codecs/encoding/format/
otlp.rs

1use crate::encoding::ProtobufSerializer;
2use bytes::BytesMut;
3use opentelemetry_proto::metrics::metric_event_to_export_request;
4use opentelemetry_proto::proto::{
5    DESCRIPTOR_BYTES, LOGS_REQUEST_MESSAGE_TYPE, METRICS_REQUEST_MESSAGE_TYPE,
6    RESOURCE_LOGS_JSON_FIELD, RESOURCE_METRICS_JSON_FIELD, RESOURCE_SPANS_JSON_FIELD,
7    TRACES_REQUEST_MESSAGE_TYPE,
8};
9use prost::Message;
10use tokio_util::codec::Encoder;
11use vector_config_macros::configurable_component;
12use vector_core::{config::DataType, event::Event, schema};
13use vrl::{event_path, protobuf::encode::Options};
14
15/// Config used to build an `OtlpSerializer`.
16#[configurable_component]
17#[derive(Debug, Clone, Default)]
18pub struct OtlpSerializerConfig {
19    // No configuration options needed - OTLP serialization is opinionated
20}
21
22impl OtlpSerializerConfig {
23    /// Build the `OtlpSerializer` from this configuration.
24    pub fn build(&self) -> Result<OtlpSerializer, crate::encoding::BuildError> {
25        OtlpSerializer::new()
26    }
27
28    /// The data type of events that are accepted by `OtlpSerializer`.
29    pub fn input_type(&self) -> DataType {
30        DataType::all_bits()
31    }
32
33    /// The schema required by the serializer.
34    pub fn schema_requirement(&self) -> schema::Requirement {
35        schema::Requirement::empty()
36    }
37}
38
39/// Serializer that converts an `Event` to bytes using the OTLP (OpenTelemetry Protocol) protobuf format.
40///
41/// This serializer encodes events using the OTLP protobuf specification, which is the recommended
42/// encoding format for OpenTelemetry data. The output is suitable for sending to OTLP-compatible
43/// endpoints with `content-type: application/x-protobuf`.
44///
45/// # Implementation approach
46///
47/// This serializer converts Vector's internal event representation to the appropriate OTLP message type
48/// based on the top-level field in the event:
49/// - `resourceLogs` → `ExportLogsServiceRequest`
50/// - `resourceMetrics` → `ExportMetricsServiceRequest`
51/// - `resourceSpans` → `ExportTraceServiceRequest`
52///
53/// The implementation is the inverse of what the `opentelemetry` source does when decoding,
54/// ensuring round-trip compatibility.
55#[derive(Debug, Clone)]
56#[allow(dead_code)] // Fields will be used once encoding is implemented
57pub struct OtlpSerializer {
58    logs_descriptor: ProtobufSerializer,
59    metrics_descriptor: ProtobufSerializer,
60    traces_descriptor: ProtobufSerializer,
61    options: Options,
62}
63
64impl OtlpSerializer {
65    /// Creates a new OTLP serializer with the appropriate message descriptors.
66    pub fn new() -> vector_common::Result<Self> {
67        let options = Options {
68            use_json_names: true,
69            allow_lossy_string_coercion: true,
70        };
71
72        let logs_descriptor = ProtobufSerializer::new_from_bytes(
73            DESCRIPTOR_BYTES,
74            LOGS_REQUEST_MESSAGE_TYPE,
75            &options,
76        )?;
77
78        let metrics_descriptor = ProtobufSerializer::new_from_bytes(
79            DESCRIPTOR_BYTES,
80            METRICS_REQUEST_MESSAGE_TYPE,
81            &options,
82        )?;
83
84        let traces_descriptor = ProtobufSerializer::new_from_bytes(
85            DESCRIPTOR_BYTES,
86            TRACES_REQUEST_MESSAGE_TYPE,
87            &options,
88        )?;
89
90        Ok(Self {
91            logs_descriptor,
92            metrics_descriptor,
93            traces_descriptor,
94            options,
95        })
96    }
97}
98
99impl Encoder<Event> for OtlpSerializer {
100    type Error = vector_common::Error;
101
102    fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> {
103        match event {
104            Event::Log(log) => {
105                if log.contains(event_path!(RESOURCE_LOGS_JSON_FIELD)) {
106                    self.logs_descriptor.encode(Event::Log(log), buffer)
107                } else if log.contains(event_path!(RESOURCE_METRICS_JSON_FIELD)) {
108                    // Currently the OTLP metrics are Vector logs (not metrics).
109                    self.metrics_descriptor.encode(Event::Log(log), buffer)
110                } else {
111                    Err(format!(
112                        "Log event does not contain OTLP top-level fields ({RESOURCE_LOGS_JSON_FIELD} or {RESOURCE_METRICS_JSON_FIELD})",
113                    )
114                        .into())
115                }
116            }
117            Event::Trace(trace) => {
118                if trace.contains(event_path!(RESOURCE_SPANS_JSON_FIELD)) {
119                    self.traces_descriptor.encode(Event::Trace(trace), buffer)
120                } else {
121                    Err(format!(
122                        "Trace event does not contain OTLP top-level field ({RESOURCE_SPANS_JSON_FIELD})",
123                    )
124                        .into())
125                }
126            }
127            Event::Metric(metric) => {
128                let request = metric_event_to_export_request(metric)?;
129                request.encode(buffer).map_err(Into::into)
130            }
131        }
132    }
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138    use chrono::{TimeZone, Utc};
139    use opentelemetry_proto::proto::collector::metrics::v1::ExportMetricsServiceRequest;
140    use vector_core::event::{Metric, MetricKind, MetricTags, MetricValue, metric::Bucket};
141
142    // `into_event_iter` always wraps attributes in `Some(MetricTags)` (via `build_metric_tags`),
143    // even when there are none, so a tag-less input compares unequal to its round-tripped output
144    // unless we give it the same empty-but-present tag set up front.
145    fn with_empty_tags(metric: Metric) -> Metric {
146        metric.with_tags(Some(MetricTags::default()))
147    }
148
149    fn round_trip_metric(metric: Metric) -> Metric {
150        let mut serializer = OtlpSerializer::new().unwrap();
151        let mut buffer = BytesMut::new();
152        serializer
153            .encode(Event::Metric(metric), &mut buffer)
154            .expect("encode should succeed");
155
156        let request =
157            ExportMetricsServiceRequest::decode(buffer.freeze()).expect("decode should succeed");
158        let mut events: Vec<Event> = request
159            .resource_metrics
160            .into_iter()
161            .flat_map(|rm| rm.into_event_iter())
162            .collect();
163
164        assert_eq!(events.len(), 1);
165        match events.remove(0) {
166            Event::Metric(metric) => metric,
167            other => panic!("expected a metric event, got {other:?}"),
168        }
169    }
170
171    #[test]
172    fn round_trip_counter() {
173        let metric = with_empty_tags(
174            Metric::new(
175                "requests",
176                MetricKind::Incremental,
177                MetricValue::Counter { value: 42.0 },
178            )
179            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
180        );
181
182        assert_eq!(metric.clone(), round_trip_metric(metric));
183    }
184
185    #[test]
186    fn round_trip_gauge() {
187        let metric = with_empty_tags(
188            Metric::new(
189                "cpu_usage",
190                MetricKind::Absolute,
191                MetricValue::Gauge { value: 12.5 },
192            )
193            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
194        );
195
196        assert_eq!(metric.clone(), round_trip_metric(metric));
197    }
198
199    #[test]
200    fn round_trip_aggregated_histogram() {
201        let metric = with_empty_tags(
202            Metric::new(
203                "latency",
204                MetricKind::Absolute,
205                MetricValue::AggregatedHistogram {
206                    buckets: vec![
207                        Bucket {
208                            upper_limit: 1.0,
209                            count: 1,
210                        },
211                        Bucket {
212                            upper_limit: 2.0,
213                            count: 2,
214                        },
215                        Bucket {
216                            upper_limit: f64::INFINITY,
217                            count: 3,
218                        },
219                    ],
220                    count: 6,
221                    sum: 10.0,
222                },
223            )
224            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
225        );
226
227        assert_eq!(metric.clone(), round_trip_metric(metric));
228    }
229
230    #[test]
231    fn round_trip_aggregated_summary() {
232        let metric = with_empty_tags(
233            Metric::new(
234                "response_time",
235                MetricKind::Absolute,
236                MetricValue::AggregatedSummary {
237                    quantiles: vec![
238                        vector_core::event::metric::Quantile {
239                            quantile: 0.5,
240                            value: 10.0,
241                        },
242                        vector_core::event::metric::Quantile {
243                            quantile: 0.99,
244                            value: 20.0,
245                        },
246                    ],
247                    count: 100,
248                    sum: 1000.0,
249                },
250            )
251            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
252        );
253
254        assert_eq!(metric.clone(), round_trip_metric(metric));
255    }
256
257    #[test]
258    fn unsupported_metric_values_return_err() {
259        let mut serializer = OtlpSerializer::new().unwrap();
260        let mut buffer = BytesMut::new();
261
262        let set_metric = Metric::new(
263            "unique_users",
264            MetricKind::Incremental,
265            MetricValue::Set {
266                values: std::iter::once("a".to_string()).collect(),
267            },
268        );
269        assert!(
270            serializer
271                .encode(Event::Metric(set_metric), &mut buffer)
272                .is_err()
273        );
274
275        let distribution_metric = Metric::new(
276            "latencies",
277            MetricKind::Incremental,
278            MetricValue::Distribution {
279                samples: Vec::new(),
280                statistic: vector_core::event::metric::StatisticKind::Histogram,
281            },
282        );
283        assert!(
284            serializer
285                .encode(Event::Metric(distribution_metric), &mut buffer)
286                .is_err()
287        );
288    }
289}
codecs/encoding/format/otlp.rs

codecs/encoding/format/
otlp.rs
