emit_otlp 1.18.0

Emit diagnostic events to an OpenTelemetry-compatible collector.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

use std::{future::Future, pin::Pin, sync::Arc, time::Duration};

use futures_util::{stream::FuturesUnordered, StreamExt};

use crate::{
    client::{Channel, ClientEventEncoder, OtlpBuilder, OtlpInner, OtlpTransport},
    data::{
        logs::{LogsEventEncoder, LogsRequestEncoder},
        metrics::{MetricsEventEncoder, MetricsRequestEncoder},
        traces::{TracesEventEncoder, TracesRequestEncoder},
    },
    internal_metrics::InternalMetrics,
    Error,
};

pub(super) type Handle = std::thread::JoinHandle<()>;

impl OtlpBuilder {
    pub(super) fn try_spawn_inner_imp(
        otlp_logs: Option<(
            ClientEventEncoder<LogsEventEncoder>,
            emit_batcher::Sender<Channel>,
        )>,
        process_otlp_logs: Option<(
            OtlpTransport<LogsRequestEncoder>,
            emit_batcher::Receiver<Channel>,
        )>,
        otlp_traces: Option<(
            ClientEventEncoder<TracesEventEncoder>,
            emit_batcher::Sender<Channel>,
        )>,
        process_otlp_traces: Option<(
            OtlpTransport<TracesRequestEncoder>,
            emit_batcher::Receiver<Channel>,
        )>,
        otlp_metrics: Option<(
            ClientEventEncoder<MetricsEventEncoder>,
            emit_batcher::Sender<Channel>,
        )>,
        process_otlp_metrics: Option<(
            OtlpTransport<MetricsRequestEncoder>,
            emit_batcher::Receiver<Channel>,
        )>,
        metrics: Arc<InternalMetrics>,
    ) -> Result<OtlpInner, Error> {
        let receive = async move {
            let processors =
                FuturesUnordered::<Pin<Box<dyn Future<Output = ()> + Send + 'static>>>::new();

            if let Some((transport, receiver)) = process_otlp_logs {
                let transport = Arc::new(transport);

                processors.push(Box::pin(emit_batcher::tokio::exec(
                    receiver,
                    move |batch| {
                        let transport = transport.clone();

                        async move { transport.send(batch).await }
                    },
                )));
            }

            if let Some((transport, receiver)) = process_otlp_traces {
                let transport = Arc::new(transport);

                processors.push(Box::pin(emit_batcher::tokio::exec(
                    receiver,
                    move |batch| {
                        let transport = transport.clone();

                        async move { transport.send(batch).await }
                    },
                )));
            }

            if let Some((transport, receiver)) = process_otlp_metrics {
                let transport = Arc::new(transport);

                processors.push(Box::pin(emit_batcher::tokio::exec(
                    receiver,
                    move |batch| {
                        let transport = transport.clone();

                        async move { transport.send(batch).await }
                    },
                )));
            }

            // Process batches from each signal independently
            // This ensures one signal becoming unavailable doesn't
            // block the others
            let _ = processors.into_future().await;
        };

        // Spawn a background thread to process batches
        // This is a safe way to ensure users of `Otlp` can never
        // deadlock waiting on the processing of batches
        let handle = std::thread::Builder::new()
            .name("emit_otlp_worker".into())
            .spawn(move || {
                tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .unwrap()
                    .block_on(receive);
            })
            .map_err(|e| Error::new("failed to spawn background worker", e))?;

        Ok(OtlpInner {
            otlp_logs,
            otlp_traces,
            otlp_metrics,
            metrics,
            _handle: handle,
        })
    }
}

pub(crate) async fn flush(sender: &emit_batcher::Sender<Channel>, timeout: Duration) -> bool {
    emit_batcher::tokio::flush(sender, timeout).await
}