saluki_components/forwarders/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::logs::v1::logs_service_client::LogsServiceClient;
6use otlp_protos::opentelemetry::proto::collector::logs::v1::ExportLogsServiceRequest;
7use otlp_protos::opentelemetry::proto::collector::metrics::v1::metrics_service_client::MetricsServiceClient;
8use otlp_protos::opentelemetry::proto::collector::metrics::v1::ExportMetricsServiceRequest;
9use otlp_protos::opentelemetry::proto::collector::trace::v1::{
10    trace_service_client::TraceServiceClient, ExportTraceServiceRequest,
11};
12use prost::Message;
13use saluki_common::buf::FrozenChunkedBytesBuffer;
14use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
15use saluki_core::data_model::payload::Payload;
16use saluki_core::{
17    components::{forwarders::*, ComponentContext},
18    data_model::payload::PayloadType,
19};
20use saluki_error::ErrorContext as _;
21use saluki_error::GenericError;
22use stringtheory::MetaString;
23use tokio::select;
24use tonic::transport::Channel;
25use tracing::{debug, error, warn};
26
27use crate::common::otlp::{OTLP_LOGS_GRPC_SERVICE_PATH, OTLP_METRICS_GRPC_SERVICE_PATH, OTLP_TRACES_GRPC_SERVICE_PATH};
28
29/// OTLP forwarder configuration.
30///
31/// Forwards OTLP metrics and logs to the Core Agent, and traces to the Trace Agent.
32#[derive(Clone)]
33pub struct OtlpForwarderConfiguration {
34    core_agent_otlp_grpc_endpoint: String,
35    core_agent_traces_internal_port: u16,
36}
37
38impl OtlpForwarderConfiguration {
39    /// Creates a new `OtlpForwarderConfiguration` from the resolved OTLP trace configuration.
40    pub fn from_configuration(traces: &domains::otlp::Traces, core_agent_otlp_grpc_endpoint: String) -> Self {
41        Self {
42            core_agent_otlp_grpc_endpoint,
43            core_agent_traces_internal_port: traces.internal_port,
44        }
45    }
46}
47
48#[async_trait]
49impl ForwarderBuilder for OtlpForwarderConfiguration {
50    fn input_payload_type(&self) -> PayloadType {
51        PayloadType::Grpc
52    }
53
54    async fn build(&self, _context: ComponentContext) -> Result<Box<dyn Forwarder + Send>, GenericError> {
55        let trace_agent_endpoint = format!("http://localhost:{}", self.core_agent_traces_internal_port);
56        let trace_agent_channel = Channel::from_shared(trace_agent_endpoint.clone())
57            .error_context("Failed to construct gRPC channel due to an invalid endpoint.")?
58            .connect_lazy();
59        let trace_agent_client = TraceServiceClient::new(trace_agent_channel);
60
61        let normalized_endpoint = normalize_endpoint(&self.core_agent_otlp_grpc_endpoint);
62        let core_agent_grpc_channel = Channel::from_shared(normalized_endpoint)
63            .error_context("Failed to construct gRPC channel due to an invalid endpoint.")?
64            .connect_timeout(Duration::from_secs(5))
65            .connect_lazy();
66
67        let core_agent_metrics_client = MetricsServiceClient::new(core_agent_grpc_channel.clone());
68        let core_agent_logs_client = LogsServiceClient::new(core_agent_grpc_channel);
69
70        Ok(Box::new(OtlpForwarder {
71            trace_agent_client,
72            core_agent_metrics_client,
73            core_agent_logs_client,
74        }))
75    }
76}
77
78impl MemoryBounds for OtlpForwarderConfiguration {
79    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
80        builder.minimum().with_single_value::<OtlpForwarder>("component struct");
81    }
82}
83
84struct OtlpForwarder {
85    trace_agent_client: TraceServiceClient<Channel>,
86    core_agent_metrics_client: MetricsServiceClient<Channel>,
87    core_agent_logs_client: LogsServiceClient<Channel>,
88}
89
90#[async_trait]
91impl Forwarder for OtlpForwarder {
92    async fn run(mut self: Box<Self>, mut context: ForwarderContext) -> Result<(), GenericError> {
93        let Self {
94            mut trace_agent_client,
95            mut core_agent_metrics_client,
96            mut core_agent_logs_client,
97        } = *self;
98
99        let mut health = context.take_health_handle();
100
101        health.mark_ready();
102        debug!("OTLP forwarder started.");
103
104        loop {
105            select! {
106                _ = health.live() => continue,
107                maybe_payload = context.payloads().next() => match maybe_payload {
108                    Some(payload) => match payload {
109                        Payload::Grpc(grpc_payload) => {
110                            // Extract the parts of the payload, and make sure we have an OTLP payload, otherwise
111                            // we skip it and move on.
112                            let (_, endpoint, service_path, body) = grpc_payload.into_parts();
113                            match &service_path {
114                                path if *path == OTLP_TRACES_GRPC_SERVICE_PATH => {
115                                    export_traces(&mut trace_agent_client, &endpoint, &service_path, body).await;
116                                }
117                                path if *path == OTLP_METRICS_GRPC_SERVICE_PATH => {
118                                    export_metrics(&mut core_agent_metrics_client, &endpoint, &service_path, body).await;
119                                }
120                                path if *path == OTLP_LOGS_GRPC_SERVICE_PATH => {
121                                    export_logs(&mut core_agent_logs_client, &endpoint, &service_path, body).await;
122                                }
123                                _ => {
124                                    warn!(service_path = %service_path, "Received gRPC payload with unknown service path. Skipping.");
125                                    continue;
126                                }
127                            }
128
129                        },
130                        _ => continue,
131                    },
132                    None => break,
133                },
134            }
135        }
136
137        debug!("OTLP forwarder stopped.");
138
139        Ok(())
140    }
141}
142
143async fn export_traces(
144    trace_agent_client: &mut TraceServiceClient<Channel>, endpoint: &MetaString, service_path: &MetaString,
145    body: FrozenChunkedBytesBuffer,
146) {
147    // Decode the raw request payload into a typed body so we can export it.
148    //
149    // TODO: This is suboptimal since we know the payload should be valid as it was decoded when it
150    // was ingested, and only after that converted to raw bytes. It would be nice to just forward
151    // the bytes as-is without decoding it again here just to satisfy the client interface, but no
152    // such API currently exists.
153    let body = body.into_bytes();
154    let request = match ExportTraceServiceRequest::decode(body) {
155        Ok(req) => req,
156        Err(e) => {
157            error!(error = %e, "Failed to decode trace export request from payload.");
158            return;
159        }
160    };
161
162    match trace_agent_client.export(request).await {
163        Ok(response) => {
164            let resp = response.into_inner();
165            if let Some(partial_success) = resp.partial_success {
166                if partial_success.rejected_spans > 0 {
167                    warn!(
168                        rejected_spans = partial_success.rejected_spans,
169                        error = %partial_success.error_message,
170                        "Trace export partially failed."
171                    );
172                }
173            }
174        }
175        Err(e) => {
176            error!(error = %e, %endpoint, %service_path, "Failed to export traces to Trace Agent.");
177        }
178    }
179}
180
181async fn export_metrics(
182    core_agent_grpc_client: &mut MetricsServiceClient<Channel>, endpoint: &MetaString, service_path: &MetaString,
183    body: FrozenChunkedBytesBuffer,
184) {
185    // Decode the raw request payload into a typed body so we can export it.
186    //
187    // TODO: This is suboptimal since we know the payload should be valid as it was decoded when it
188    // was ingested, and only after that converted to raw bytes. It would be nice to just forward
189    // the bytes as-is without decoding it again here just to satisfy the client interface, but no
190    // such API currently exists.
191    let body = body.into_bytes();
192
193    let request = match ExportMetricsServiceRequest::decode(body) {
194        Ok(req) => req,
195        Err(e) => {
196            error!(error = %e, "Failed to decode metrics or logs export request from payload.");
197            return;
198        }
199    };
200
201    match core_agent_grpc_client.export(request).await {
202        Ok(response) => {
203            let resp = response.into_inner();
204            if let Some(partial_success) = resp.partial_success {
205                if partial_success.rejected_data_points > 0 {
206                    warn!(
207                        rejected_data_points = partial_success.rejected_data_points,
208                        error = %partial_success.error_message,
209                        "Metrics export partially failed."
210                    );
211                }
212            }
213        }
214        Err(e) => {
215            error!(error = %e, %endpoint, %service_path, "Failed to export metrics to Core Agent.");
216        }
217    }
218}
219
220async fn export_logs(
221    core_agent_grpc_client: &mut LogsServiceClient<Channel>, endpoint: &MetaString, service_path: &MetaString,
222    body: FrozenChunkedBytesBuffer,
223) {
224    // Decode the raw request payload into a typed body so we can export it.
225    //
226    // TODO: This is suboptimal since we know the payload should be valid as it was decoded when it
227    // was ingested, and only after that converted to raw bytes. It would be nice to just forward
228    // the bytes as-is without decoding it again here just to satisfy the client interface, but no
229    // such API currently exists.
230    let body = body.into_bytes();
231
232    let request = match ExportLogsServiceRequest::decode(body) {
233        Ok(req) => req,
234        Err(e) => {
235            error!(error = %e, "Failed to decode logs export request from payload.");
236            return;
237        }
238    };
239
240    match core_agent_grpc_client.export(request).await {
241        Ok(response) => {
242            let resp = response.into_inner();
243            if let Some(partial_success) = resp.partial_success {
244                if partial_success.rejected_log_records > 0 {
245                    warn!(
246                        rejected_log_records = partial_success.rejected_log_records,
247                        error = %partial_success.error_message,
248                        "Trace export partially failed."
249                    );
250                }
251            }
252        }
253        Err(e) => {
254            error!(error = %e, %endpoint, %service_path, "Failed to export metrics to Core Agent.");
255        }
256    }
257}
258
259fn normalize_endpoint(endpoint: &str) -> String {
260    if endpoint.starts_with("http://") || endpoint.starts_with("https://") {
261        endpoint.to_string()
262    } else {
263        format!("https://{}", endpoint)
264    }
265}
266
267#[cfg(test)]
268mod tests {
269    use super::normalize_endpoint;
270
271    #[test]
272    fn normalize_endpoint_prepends_https_only_when_scheme_missing() {
273        // Endpoints that already carry an `http://`/`https://` scheme are passed through untouched; anything else
274        // (a bare host:port, as the Core Agent gRPC endpoint is typically configured) gets an `https://` prefix.
275        let cases = [
276            ("http://localhost:4317", "http://localhost:4317"),
277            ("https://api.datadoghq.com:443", "https://api.datadoghq.com:443"),
278            ("localhost:4317", "https://localhost:4317"),
279            ("127.0.0.1:5003", "https://127.0.0.1:5003"),
280        ];
281
282        for (input, expected) in cases {
283            assert_eq!(expected, normalize_endpoint(input), "normalize_endpoint({input:?})");
284        }
285    }
286}
saluki_components/forwarders/otlp/mod.rs

saluki_components/forwarders/otlp/
mod.rs
