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

saluki_components/decoders/otlp/
mod.rs
