saluki_components/decoders/otlp/
mod.rs

1use std::time::Duration;
2
3use agent_data_plane_config::domains;
4use async_trait::async_trait;
5use otlp_protos::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest;
6use prost::Message;
7use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
8use saluki_core::{
9    components::{
10        decoders::{Decoder, DecoderBuilder, DecoderContext},
11        ComponentContext,
12    },
13    data_model::{event::EventType, payload::PayloadType},
14    topology::interconnect::EventBufferManager,
15};
16use saluki_error::GenericError;
17use tokio::{
18    select,
19    time::{interval, MissedTickBehavior},
20};
21use tracing::{debug, error, warn};
22
23use crate::common::otlp::traces::translator::OtlpTracesTranslator;
24use crate::common::otlp::{
25    build_metrics, Metrics, OTLP_LOGS_GRPC_SERVICE_PATH, OTLP_METRICS_GRPC_SERVICE_PATH, OTLP_TRACES_GRPC_SERVICE_PATH,
26};
27
28/// Configuration for the OTLP decoder.
29#[derive(Default)]
30pub struct OtlpDecoderConfiguration {
31    /// Resolved OTLP trace ingestion settings.
32    traces: domains::otlp::Traces,
33}
34
35impl OtlpDecoderConfiguration {
36    /// Creates a new `OtlpDecoderConfiguration` from the resolved OTLP trace configuration.
37    pub fn from_configuration(traces: &domains::otlp::Traces) -> Self {
38        Self { traces: traces.clone() }
39    }
40}
41
42#[async_trait]
43impl DecoderBuilder for OtlpDecoderConfiguration {
44    fn input_payload_type(&self) -> PayloadType {
45        PayloadType::Grpc
46    }
47
48    fn output_event_type(&self) -> EventType {
49        EventType::Trace
50    }
51
52    async fn build(&self, context: ComponentContext) -> Result<Box<dyn Decoder + Send>, GenericError> {
53        let metrics = build_metrics(&context);
54        let traces_translator = OtlpTracesTranslator::new(self.traces.clone());
55
56        Ok(Box::new(OtlpDecoder {
57            traces_translator,
58            metrics,
59        }))
60    }
61}
62
63impl MemoryBounds for OtlpDecoderConfiguration {
64    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
65        builder.minimum().with_single_value::<OtlpDecoder>("decoder struct");
66    }
67}
68
69/// OTLP decoder.
70pub struct OtlpDecoder {
71    traces_translator: OtlpTracesTranslator,
72    metrics: Metrics,
73}
74
75#[async_trait]
76impl Decoder for OtlpDecoder {
77    async fn run(self: Box<Self>, mut context: DecoderContext) -> Result<(), GenericError> {
78        let Self {
79            mut traces_translator,
80            metrics,
81        } = *self;
82        let mut health = context.take_health_handle();
83        health.mark_ready();
84
85        debug!("OTLP decoder started.");
86
87        // Set a buffer flush interval of 100ms to ensure we flush buffered events periodically.
88        let mut buffer_flush = interval(Duration::from_millis(100));
89        buffer_flush.set_missed_tick_behavior(MissedTickBehavior::Delay);
90
91        let mut event_buffer_manager = EventBufferManager::default();
92
93        loop {
94            select! {
95                maybe_payload = context.payloads().next() => {
96                    let payload = match maybe_payload {
97                        Some(payload) => payload,
98                        None => {
99                            debug!("Payloads stream closed, shutting down decoder.");
100                            break;
101                        }
102                    };
103
104                    let grpc_payload = match payload.try_into_grpc_payload() {
105                        Some(grpc) => grpc,
106                        None => {
107                            warn!("Received non-gRPC payload in OTLP decoder. Dropping payload.");
108                            continue;
109                        }
110                    };
111
112                    match grpc_payload.service_path() {
113                        path if path == &*OTLP_TRACES_GRPC_SERVICE_PATH => {
114                            let (_, _, _, body) = grpc_payload.into_parts();
115                            let request = match ExportTraceServiceRequest::decode(body.into_bytes()) {
116                                Ok(req) => req,
117                                Err(e) => {
118                                    error!(error = %e, "Failed to decode OTLP trace request.");
119                                    continue;
120                                }
121                            };
122
123                            for resource_spans in request.resource_spans {
124                                for trace_event in traces_translator.translate_spans(resource_spans, &metrics) {
125                                    if let Some(event_buffer) = event_buffer_manager.try_push(trace_event) {
126                                        if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
127                                            error!(error = %e, "Failed to dispatch trace events.");
128                                        }
129                                    }
130                                }
131                            }
132                        }
133                        path if path == &*OTLP_METRICS_GRPC_SERVICE_PATH => {
134                            warn!("OTLP metrics decoding not yet implemented. Dropping metrics payload.");
135                        }
136                        path if path == &*OTLP_LOGS_GRPC_SERVICE_PATH => {
137                            warn!("OTLP logs decoding not yet implemented. Dropping logs payload.");
138                        }
139                        path => {
140                            warn!(service_path = path, "Received gRPC payload with unknown service path. Dropping payload.");
141                        }
142                    }
143                },
144                _ = buffer_flush.tick() => {
145                    if let Some(event_buffer) = event_buffer_manager.consume() {
146                        if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
147                            error!(error = %e, "Failed to dispatch buffered trace events.");
148                        }
149                    }
150                },
151                _ = health.live() => continue,
152            }
153        }
154
155        if let Some(event_buffer) = event_buffer_manager.consume() {
156            if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
157                error!(error = %e, "Failed to dispatch final trace events.");
158            }
159        }
160
161        debug!("OTLP decoder stopped.");
162
163        Ok(())
164    }
165}
saluki_components/decoders/otlp/mod.rs

saluki_components/decoders/otlp/
mod.rs
