saluki_components/decoders/otlp/
mod.rs

1use std::time::Duration;
2
3use agent_data_plane_config::domains;
4use async_trait::async_trait;
5use datadog_agent_apm::semantics::SemanticRegistryProvider;
6use otlp_protos::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest;
7use prost::Message;
8use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
9use saluki_core::{
10    components::{
11        decoders::{Decoder, DecoderBuilder, DecoderContext},
12        BuildContext,
13    },
14    data_model::{event::EventType, payload::PayloadType},
15    topology::interconnect::EventBufferManager,
16};
17use saluki_error::GenericError;
18use tokio::{
19    select,
20    time::{interval, MissedTickBehavior},
21};
22use tracing::{debug, error, warn};
23
24use crate::common::otlp::traces::translator::OtlpTracesTranslator;
25use crate::common::otlp::{
26    build_metrics, Metrics, OTLP_LOGS_GRPC_SERVICE_PATH, OTLP_METRICS_GRPC_SERVICE_PATH, OTLP_TRACES_GRPC_SERVICE_PATH,
27};
28
29/// Configuration for the OTLP decoder.
30pub struct OtlpDecoderConfiguration {
31    /// Resolved OTLP trace ingestion settings.
32    traces: domains::otlp::Traces,
33
34    /// Maximum length of a span's resource name, in bytes.
35    ///
36    /// Defaults to `usize::MAX`, meaning resource names are not truncated.
37    max_resource_len: usize,
38
39    /// Supplies attribute mappings for trace translation. Defaults to embedded mappings without remote updates.
40    semantic_registry: SemanticRegistryProvider,
41}
42
43impl OtlpDecoderConfiguration {
44    /// Creates a new `OtlpDecoderConfiguration` from the resolved OTLP trace configuration.
45    pub fn from_configuration(traces: &domains::otlp::Traces) -> Self {
46        Self {
47            traces: traces.clone(),
48            max_resource_len: usize::MAX,
49            semantic_registry: SemanticRegistryProvider::default(),
50        }
51    }
52
53    /// Sets the maximum length of a span's resource name, in bytes.
54    ///
55    /// Resource names longer than this limit are truncated to the limit, on a UTF-8 boundary. Defaults to `usize::MAX`
56    /// (unbounded), which matches the decoder's behavior before resource name truncation was introduced.
57    ///
58    /// If set to `0`, every resource name is truncated to an empty string. Pass through the configured value
59    /// unmodified, including `0`, so the configured behavior is always honored.
60    pub fn with_max_resource_len(mut self, max_resource_len: usize) -> Self {
61        self.max_resource_len = max_resource_len;
62        self
63    }
64
65    /// Sets the provider of attribute mappings used by registry-based trace lookups.
66    ///
67    /// Pass a subscribed provider to receive remote mapping updates. The translator reads one registry snapshot per
68    /// `ResourceSpans` batch, so an update cannot change mappings midway through that batch.
69    ///
70    /// Defaults to embedded mappings without remote updates.
71    pub fn with_semantic_registry(mut self, semantic_registry: SemanticRegistryProvider) -> Self {
72        self.semantic_registry = semantic_registry;
73        self
74    }
75}
76
77#[async_trait]
78impl DecoderBuilder for OtlpDecoderConfiguration {
79    fn input_payload_type(&self) -> PayloadType {
80        PayloadType::Grpc
81    }
82
83    fn output_event_type(&self) -> EventType {
84        EventType::Trace
85    }
86
87    async fn build(&self, context: BuildContext) -> Result<Box<dyn Decoder + Send>, GenericError> {
88        let metrics = build_metrics(context.component_context());
89        Ok(Box::new(OtlpDecoder {
90            traces_translator: self.traces_translator(),
91            metrics,
92        }))
93    }
94}
95
96impl OtlpDecoderConfiguration {
97    fn traces_translator(&self) -> OtlpTracesTranslator {
98        OtlpTracesTranslator::new(
99            self.traces.clone(),
100            self.max_resource_len,
101            self.semantic_registry.clone(),
102        )
103    }
104}
105
106impl MemoryBounds for OtlpDecoderConfiguration {
107    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
108        builder.minimum().with_single_value::<OtlpDecoder>("decoder struct");
109    }
110}
111
112/// OTLP decoder.
113pub struct OtlpDecoder {
114    traces_translator: OtlpTracesTranslator,
115    metrics: Metrics,
116}
117
118#[async_trait]
119impl Decoder for OtlpDecoder {
120    async fn run(self: Box<Self>, mut context: DecoderContext) -> Result<(), GenericError> {
121        let Self {
122            mut traces_translator,
123            metrics,
124        } = *self;
125        let mut health = context.take_health_handle();
126        health.mark_ready();
127
128        debug!("OTLP decoder started.");
129
130        // Set a buffer flush interval of 100ms to ensure we flush buffered events periodically.
131        let mut buffer_flush = interval(Duration::from_millis(100));
132        buffer_flush.set_missed_tick_behavior(MissedTickBehavior::Delay);
133
134        let mut event_buffer_manager = EventBufferManager::default();
135
136        loop {
137            select! {
138                maybe_payload = context.payloads().next() => {
139                    let payload = match maybe_payload {
140                        Some(payload) => payload,
141                        None => {
142                            debug!("Payloads stream closed, shutting down decoder.");
143                            break;
144                        }
145                    };
146
147                    let grpc_payload = match payload.try_into_grpc_payload() {
148                        Some(grpc) => grpc,
149                        None => {
150                            warn!("Received non-gRPC payload in OTLP decoder. Dropping payload.");
151                            continue;
152                        }
153                    };
154
155                    match grpc_payload.service_path() {
156                        path if path == &*OTLP_TRACES_GRPC_SERVICE_PATH => {
157                            let (_, _, _, body) = grpc_payload.into_parts();
158                            let request = match ExportTraceServiceRequest::decode(body.into_bytes()) {
159                                Ok(req) => req,
160                                Err(e) => {
161                                    error!(error = %e, "Failed to decode OTLP trace request.");
162                                    continue;
163                                }
164                            };
165
166                            for resource_spans in request.resource_spans {
167                                for trace_event in traces_translator.translate_spans(resource_spans, &metrics) {
168                                    if let Some(event_buffer) = event_buffer_manager.try_push(trace_event) {
169                                        if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
170                                            error!(error = %e, "Failed to dispatch trace events.");
171                                        }
172                                    }
173                                }
174                            }
175                        }
176                        path if path == &*OTLP_METRICS_GRPC_SERVICE_PATH => {
177                            warn!("OTLP metrics decoding not yet implemented. Dropping metrics payload.");
178                        }
179                        path if path == &*OTLP_LOGS_GRPC_SERVICE_PATH => {
180                            warn!("OTLP logs decoding not yet implemented. Dropping logs payload.");
181                        }
182                        path => {
183                            warn!(service_path = path, "Received gRPC payload with unknown service path. Dropping payload.");
184                        }
185                    }
186                },
187                _ = buffer_flush.tick() => {
188                    if let Some(event_buffer) = event_buffer_manager.consume() {
189                        if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
190                            error!(error = %e, "Failed to dispatch buffered trace events.");
191                        }
192                    }
193                },
194                _ = health.live() => continue,
195            }
196        }
197
198        if let Some(event_buffer) = event_buffer_manager.consume() {
199            if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
200                error!(error = %e, "Failed to dispatch final trace events.");
201            }
202        }
203
204        debug!("OTLP decoder stopped.");
205
206        Ok(())
207    }
208}
209
210#[cfg(test)]
211mod tests {
212    use datadog_agent_apm::semantics::test_util::TestRegistryPublisher;
213
214    use super::*;
215    use crate::common::otlp::traces::translator::{custom_status_code_registry, translate_custom_status_code};
216
217    #[test]
218    fn injected_provider_reaches_the_traces_translator() {
219        let (publisher, provider) = TestRegistryPublisher::new();
220        let traces = domains::otlp::Traces {
221            string_interner_size: std::num::NonZeroUsize::new(64 * 1024).unwrap(),
222            ..Default::default()
223        };
224        let mut translator = OtlpDecoderConfiguration::from_configuration(&traces)
225            .with_semantic_registry(provider)
226            .traces_translator();
227        let metrics = Metrics::for_tests();
228        assert_eq!(translate_custom_status_code(&mut translator, &metrics), None);
229
230        publisher.assign([("a", custom_status_code_registry())]);
231        assert_eq!(translate_custom_status_code(&mut translator, &metrics), Some(418.0));
232    }
233}
saluki_components/decoders/otlp/mod.rs

saluki_components/decoders/otlp/
mod.rs
