feat(codecs): support encoding native Vector metrics in OTLP serializer

changelog.d/otlp-metric-encoder.feature.md

@@ -0,0 +1,3 @@
+The OTLP codec's serializer now supports native Vector `Metric` events for `Counter`, `Gauge`, `AggregatedHistogram`, and `AggregatedSummary` values, converting them into the OTLP `Sum`, `Gauge`, `Histogram`, and `Summary` protobuf types respectively. Previously, encoding a native metric event with the OTLP serializer always failed.
+
+authors: petere-datadog

lib/codecs/src/encoding/format/otlp.rs

@@ -1,10 +1,12 @@
 use crate::encoding::ProtobufSerializer;
 use bytes::BytesMut;
+use opentelemetry_proto::metrics::metric_event_to_export_request;
 use opentelemetry_proto::proto::{
     DESCRIPTOR_BYTES, LOGS_REQUEST_MESSAGE_TYPE, METRICS_REQUEST_MESSAGE_TYPE,
     RESOURCE_LOGS_JSON_FIELD, RESOURCE_METRICS_JSON_FIELD, RESOURCE_SPANS_JSON_FIELD,
     TRACES_REQUEST_MESSAGE_TYPE,
 };
+use prost::Message;
 use tokio_util::codec::Encoder;
 use vector_config_macros::configurable_component;
 use vector_core::{config::DataType, event::Event, schema};
@@ -25,7 +27,7 @@ impl OtlpSerializerConfig {
 
     /// The data type of events that are accepted by `OtlpSerializer`.
     pub fn input_type(&self) -> DataType {
-        DataType::Log | DataType::Trace
+        DataType::all_bits()
     }
 
     /// The schema required by the serializer.
@@ -98,9 +100,6 @@ impl Encoder<Event> for OtlpSerializer {
     type Error = vector_common::Error;
 
     fn encode(&mut self, event: Event, buffer: &mut BytesMut) -> Result<(), Self::Error> {
-        // Determine which descriptor to use based on top-level OTLP fields
-        // This handles events that were decoded with use_otlp_decoding enabled
-        // The deserializer uses use_json_names: true, so fields are in camelCase
         match &event {
             Event::Log(log) => {
                 if log.contains(event_path!(RESOURCE_LOGS_JSON_FIELD)) {
@@ -125,9 +124,167 @@ impl Encoder<Event> for OtlpSerializer {
                         .into())
                 }
             }
-            Event::Metric(_) => {
-                Err("OTLP serializer does not support native Vector metrics yet.".into())
+            Event::Metric(metric) => {
+                let request = metric_event_to_export_request(metric)?;
+                request.encode(buffer)?;
+                Ok(())
             }
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::{TimeZone, Utc};
+    use opentelemetry_proto::proto::collector::metrics::v1::ExportMetricsServiceRequest;
+    use vector_core::event::{Metric, MetricKind, MetricTags, MetricValue, metric::Bucket};
+
+    // `into_event_iter` always wraps attributes in `Some(MetricTags)` (via `build_metric_tags`),
+    // even when there are none, so a tag-less input compares unequal to its round-tripped output
+    // unless we give it the same empty-but-present tag set up front.
+    fn with_empty_tags(metric: Metric) -> Metric {
+        metric.with_tags(Some(MetricTags::default()))
+    }
+
+    fn round_trip_metric(metric: Metric) -> Metric {
+        let mut serializer = OtlpSerializer::new().unwrap();
+        let mut buffer = BytesMut::new();
+        serializer
+            .encode(Event::Metric(metric), &mut buffer)
+            .expect("encode should succeed");
+
+        let request =
+            ExportMetricsServiceRequest::decode(buffer.freeze()).expect("decode should succeed");
+        let mut events: Vec<Event> = request
+            .resource_metrics
+            .into_iter()
+            .flat_map(|rm| rm.into_event_iter())
+            .collect();
+
+        assert_eq!(events.len(), 1);
+        match events.remove(0) {
+            Event::Metric(metric) => metric,
+            other => panic!("expected a metric event, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn round_trip_counter() {
+        let metric = with_empty_tags(
+            Metric::new(
+                "requests",
+                MetricKind::Incremental,
+                MetricValue::Counter { value: 42.0 },
+            )
+            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
+        );
+
+        assert_eq!(metric.clone(), round_trip_metric(metric));
+    }
+
+    #[test]
+    fn round_trip_gauge() {
+        let metric = with_empty_tags(
+            Metric::new(
+                "cpu_usage",
+                MetricKind::Absolute,
+                MetricValue::Gauge { value: 12.5 },
+            )
+            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
+        );
+
+        assert_eq!(metric.clone(), round_trip_metric(metric));
+    }
+
+    #[test]
+    fn round_trip_aggregated_histogram() {
+        let metric = with_empty_tags(
+            Metric::new(
+                "latency",
+                MetricKind::Absolute,
+                MetricValue::AggregatedHistogram {
+                    buckets: vec![
+                        Bucket {
+                            upper_limit: 1.0,
+                            count: 1,
+                        },
+                        Bucket {
+                            upper_limit: 2.0,
+                            count: 2,
+                        },
+                        Bucket {
+                            upper_limit: f64::INFINITY,
+                            count: 3,
+                        },
+                    ],
+                    count: 6,
+                    sum: 10.0,
+                },
+            )
+            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
+        );
+
+        assert_eq!(metric.clone(), round_trip_metric(metric));
+    }
+
+    #[test]
+    fn round_trip_aggregated_summary() {
+        let metric = with_empty_tags(
+            Metric::new(
+                "response_time",
+                MetricKind::Absolute,
+                MetricValue::AggregatedSummary {
+                    quantiles: vec![
+                        vector_core::event::metric::Quantile {
+                            quantile: 0.5,
+                            value: 10.0,
+                        },
+                        vector_core::event::metric::Quantile {
+                            quantile: 0.99,
+                            value: 20.0,
+                        },
+                    ],
+                    count: 100,
+                    sum: 1000.0,
+                },
+            )
+            .with_timestamp(Some(Utc.timestamp_nanos(1_000_000_000))),
+        );
+
+        assert_eq!(metric.clone(), round_trip_metric(metric));
+    }
+
+    #[test]
+    fn unsupported_metric_values_return_err() {
+        let mut serializer = OtlpSerializer::new().unwrap();
+        let mut buffer = BytesMut::new();
+
+        let set_metric = Metric::new(
+            "unique_users",
+            MetricKind::Incremental,
+            MetricValue::Set {
+                values: std::iter::once("a".to_string()).collect(),
+            },
+        );
+        assert!(
+            serializer
+                .encode(Event::Metric(set_metric), &mut buffer)
+                .is_err()
+        );
+
+        let distribution_metric = Metric::new(
+            "latencies",
+            MetricKind::Incremental,
+            MetricValue::Distribution {
+                samples: Vec::new(),
+                statistic: vector_core::event::metric::StatisticKind::Histogram,
+            },
+        );
+        assert!(
+            serializer
+                .encode(Event::Metric(distribution_metric), &mut buffer)
+                .is_err()
+        );
+    }
+}

lib/opentelemetry-proto/Cargo.toml

@@ -22,4 +22,5 @@ ordered-float.workspace = true
 prost.workspace = true
 tonic.workspace = true
 vrl.workspace = true
+vector-common = { path = "../vector-common", default-features = false }
 vector-core = { path = "../vector-core", default-features = false }

lib/opentelemetry-proto/src/common.rs

@@ -1,9 +1,9 @@
 use bytes::Bytes;
 use ordered_float::NotNan;
-use vector_core::event::metric::TagValue;
+use vector_core::event::metric::{TagValue, TagValueSet};
 use vrl::value::{ObjectMap, Value};
 
-use super::proto::common::v1::{KeyValue, any_value::Value as PBValue};
+use super::proto::common::v1::{AnyValue, ArrayValue, KeyValue, any_value::Value as PBValue};
 
 impl From<PBValue> for Value {
     fn from(av: PBValue) -> Self {
@@ -37,6 +37,36 @@ impl From<PBValue> for TagValue {
     }
 }
 
+impl From<&TagValue> for AnyValue {
+    fn from(tag: &TagValue) -> Self {
+        match tag {
+            TagValue::Value(s) => Self {
+                value: Some(PBValue::StringValue(s.clone())),
+            },
+            TagValue::Bare => Self { value: None },
+        }
+    }
+}
+
+pub fn str_to_key_value(key: &str, val: &TagValue) -> KeyValue {
+    KeyValue {
+        key: key.to_string(),
+        value: Some(val.into()),
+    }
+}
+
+pub fn tag_set_to_any_value(tag_set: &TagValueSet) -> Option<AnyValue> {
+    match tag_set {
+        TagValueSet::Empty => None,
+        TagValueSet::Single(tag) => Some(tag.into()),
+        TagValueSet::Set(set) => Some(AnyValue {
+            value: Some(PBValue::ArrayValue(ArrayValue {
+                values: set.iter().map(Into::into).collect(),
+            })),
+        }),
+    }
+}
+
 pub fn kv_list_into_value(arr: Vec<KeyValue>) -> Value {
     Value::Object(
         arr.into_iter()