chore(codecs): add native OTLP metric encoder e2e suite

.github/workflows/changes.yml

@@ -447,6 +447,7 @@ jobs:
       datadog-metrics: ${{ steps.filter.outputs.datadog-metrics }}
       opentelemetry-logs: ${{ steps.filter.outputs.opentelemetry-logs }}
       opentelemetry-metrics: ${{ steps.filter.outputs.opentelemetry-metrics }}
+      opentelemetry-metrics-native: ${{ steps.filter.outputs.opentelemetry-metrics-native }}
       any: ${{ steps.detect-changes.outputs.any }}
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -480,7 +481,8 @@ jobs:
             "datadog-logs": ${{ steps.filter.outputs.datadog-logs }},
             "datadog-metrics": ${{ steps.filter.outputs.datadog-metrics }},
             "opentelemetry-logs": ${{ steps.filter.outputs.opentelemetry-logs }},
-            "opentelemetry-metrics": ${{ steps.filter.outputs.opentelemetry-metrics }}
+            "opentelemetry-metrics": ${{ steps.filter.outputs.opentelemetry-metrics }},
+            "opentelemetry-metrics-native": ${{ steps.filter.outputs.opentelemetry-metrics-native }}
           }
           EOF
           )

.github/workflows/ci-integration-review.yml

@@ -199,7 +199,14 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        service: ["datadog-logs", "datadog-metrics", "opentelemetry-logs", "opentelemetry-metrics"]
+        service:
+          [
+            "datadog-logs",
+            "datadog-metrics",
+            "opentelemetry-logs",
+            "opentelemetry-metrics",
+            "opentelemetry-metrics-native"
+          ]
     steps:
       - name: Evaluate run condition
         id: run_condition

.github/workflows/integration.yml

@@ -215,7 +215,14 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        service: ["datadog-logs", "datadog-metrics", "opentelemetry-logs", "opentelemetry-metrics"]
+        service:
+          [
+            "datadog-logs",
+            "datadog-metrics",
+            "opentelemetry-logs",
+            "opentelemetry-metrics",
+            "opentelemetry-metrics-native"
+          ]
     timeout-minutes: 90
 
     steps:

tests/e2e/opentelemetry-metrics-native/config/compose.yaml

@@ -0,0 +1,107 @@
+name: opentelemetry-vector-native-e2e
+services:
+  otel-collector-source:
+    container_name: otel-collector-source-native
+    image: otel/opentelemetry-collector-contrib:${CONFIG_COLLECTOR_VERSION}
+    init: true
+    volumes:
+      - type: bind
+        source: ../data/collector-source.yaml
+        target: /etc/otelcol-contrib/config.yaml
+        read_only: true
+    ports:
+      - "${OTEL_COLLECTOR_SOURCE_GRPC_PORT:-4317}:4317"
+      - "${OTEL_COLLECTOR_SOURCE_HTTP_PORT:-4318}:4318"
+    command: ["--config=/etc/otelcol-contrib/config.yaml"]
+
+  metrics-generator:
+    container_name: metrics-generator-native
+    build:
+      context: ../../../..
+      dockerfile: ./tests/e2e/opentelemetry-common/telemetrygen.Dockerfile
+    init: true
+    depends_on:
+      otel-collector-source:
+        condition: service_started
+      vector:
+        condition: service_started
+      otel-collector-sink:
+        condition: service_started
+    command:
+      - "-c"
+      - |
+        until nc -z otel-collector-source 4318; do
+          sleep 0.5
+        done
+        # Generate 50 Gauge metrics
+        telemetrygen metrics \
+          --otlp-endpoint=otel-collector-source:4318 \
+          --otlp-insecure \
+          --otlp-http \
+          --metrics=50 \
+          --metric-type=Gauge \
+          --otlp-metric-name=gauge_metric \
+          --telemetry-attributes='metric.type="gauge"' \
+          --rate=0
+        # Generate 50 Sum (counter) metrics
+        telemetrygen metrics \
+          --otlp-endpoint=otel-collector-source:4318 \
+          --otlp-insecure \
+          --otlp-http \
+          --metrics=50 \
+          --metric-type=Sum \
+          --otlp-metric-name=sum_metric \
+          --aggregation-temporality=cumulative \
+          --telemetry-attributes='metric.type="sum"' \
+          --rate=0
+
+  otel-collector-sink:
+    container_name: otel-collector-sink-native
+    build:
+      context: ../../../..
+      dockerfile: ./tests/e2e/opentelemetry-common/collector.Dockerfile
+      args:
+        CONFIG_COLLECTOR_VERSION: ${CONFIG_COLLECTOR_VERSION}
+    init: true
+    user: "0:0" # test only, override special user with root
+    # Wipe stale output before starting: the file exporter appends and the
+    # vector_target volume is external, so it survives `compose down --volumes`.
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - |
+        rm -rf /output/opentelemetry-metrics-native
+        mkdir -p /output/opentelemetry-metrics-native
+        exec /otelcol-contrib --config /etc/otelcol-contrib/config.yaml
+    volumes:
+      - type: bind
+        source: ../data/collector-sink.yaml
+        target: /etc/otelcol-contrib/config.yaml
+        read_only: true
+      - type: volume
+        source: vector_target
+        target: /output
+    ports:
+      - "${OTEL_COLLECTOR_SINK_HTTP_PORT:-5318}:5318"
+
+  vector:
+    container_name: vector-otel-metrics-native-e2e
+    image: ${CONFIG_VECTOR_IMAGE}
+    init: true
+    volumes:
+      - type: bind
+        source: ../data/vector_otlp.yaml
+        target: /etc/vector/vector.yaml
+        read_only: true
+    environment:
+      - VECTOR_LOG=${VECTOR_LOG:-info}
+      - FEATURES=e2e-tests-opentelemetry
+    command: ["vector", "-c", "/etc/vector/vector.yaml"]
+
+volumes:
+  vector_target:
+    external: true
+
+networks:
+  default:
+    name: ${VECTOR_NETWORK}
+    external: true

tests/e2e/opentelemetry-metrics-native/config/test.yaml

@@ -0,0 +1,26 @@
+features:
+  - e2e-tests-opentelemetry
+
+test: "e2e"
+
+test_filter: "opentelemetry::metrics_native::"
+
+runner:
+  needs_docker_socket: true
+  env:
+    OTEL_COLLECTOR_SOURCE_GRPC_PORT: "4317"
+    OTEL_COLLECTOR_SOURCE_HTTP_PORT: "4318"
+    OTEL_COLLECTOR_SINK_HTTP_PORT: "5318"
+
+matrix:
+  collector_version: ["latest"]
+
+# Only trigger this integration test if relevant OTEL source/sink files change
+paths:
+  - "src/sources/opentelemetry/**"
+  - "src/sinks/opentelemetry/**"
+  - "src/internal_events/opentelemetry_*"
+  - "tests/e2e/opentelemetry/**"
+  - "tests/e2e/opentelemetry-metrics-native/**"
+  - "lib/codecs/src/**/otlp.rs"
+  - "lib/opentelemetry-proto/**"

tests/e2e/opentelemetry-metrics-native/data/collector-sink.yaml

@@ -0,0 +1,26 @@
+receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: "0.0.0.0:5318"
+
+processors:
+  batch: {}
+
+exporters:
+  debug: {}
+  file:
+    path: /output/opentelemetry-metrics-native/collector-file-exporter.log
+    rotation:
+      max_megabytes: 10
+      max_days: 1
+
+service:
+  pipelines:
+    metrics:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [debug, file]
+  telemetry:
+    logs:
+      level: "debug"

tests/e2e/opentelemetry-metrics-native/data/collector-source.yaml

@@ -0,0 +1,33 @@
+receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: "0.0.0.0:4318"
+      grpc:
+
+processors:
+  batch: {}
+  transform/scope:
+    metric_statements:
+      - context: scope
+        statements:
+          - set(name, "vector-e2e-metrics")
+          - set(version, "1.2.3")
+
+exporters:
+  otlp/grpc:
+    endpoint: vector:4317
+    tls:
+      insecure: true
+  otlphttp/vector:
+    endpoint: http://vector:4318
+    tls:
+      insecure: true
+  debug: {}
+
+service:
+  pipelines:
+    metrics:
+      receivers: [otlp]
+      processors: [transform/scope, batch]
+      exporters: [debug, otlp/grpc, otlphttp/vector]

tests/e2e/opentelemetry-metrics-native/data/vector_otlp.yaml

@@ -0,0 +1,23 @@
+sources:
+  source0:
+    type: opentelemetry
+    grpc:
+      address: 0.0.0.0:4317
+    http:
+      address: 0.0.0.0:4318
+      keepalive:
+        max_connection_age_jitter_factor: 0.1
+        max_connection_age_secs: 300
+    # Native decoding yields Event::Metric, exercising the OTLP metric encoder.
+    use_otlp_decoding: false
+
+sinks:
+  otel_sink:
+    inputs:
+      - source0.metrics
+    type: opentelemetry
+    protocol:
+      type: http
+      uri: http://otel-collector-sink:5318/v1/metrics
+      encoding:
+        codec: otlp

tests/e2e/opentelemetry/metrics_native/mod.rs

@@ -0,0 +1,120 @@
+use crate::opentelemetry::{
+    assert_service_name_with, parse_line_to_export_type_request, read_file_helper,
+};
+
+use vector_lib::opentelemetry::proto::{
+    METRICS_REQUEST_MESSAGE_TYPE,
+    collector::metrics::v1::ExportMetricsServiceRequest,
+    common::v1::any_value::Value as AnyValueEnum,
+    metrics::v1::{Gauge, Sum, metric::Data as MetricData},
+};
+
+// telemetrygen emits 50 Gauge + 50 Sum. The source collector fans out to Vector over both gRPC
+// and HTTP, so Vector sees each metric twice: 100 gauge + 100 sum data points.
+const EXPECTED_GAUGE: usize = 100;
+const EXPECTED_SUM: usize = 100;
+const EXPECTED_SCOPE_NAME: &str = "vector-e2e-metrics";
+const EXPECTED_SCOPE_VERSION: &str = "1.2.3";
+
+fn parse_export_metrics_request(content: &str) -> Result<ExportMetricsServiceRequest, String> {
+    let mut merged = ExportMetricsServiceRequest {
+        resource_metrics: Vec::new(),
+    };
+
+    for (line_num, line) in content.lines().enumerate() {
+        let line = line.trim();
+        if line.is_empty() {
+            continue;
+        }
+        merged.resource_metrics.extend(
+            parse_line_to_export_type_request::<ExportMetricsServiceRequest>(
+                METRICS_REQUEST_MESSAGE_TYPE,
+                line,
+            )
+            .map_err(|e| format!("Line {}: {}", line_num + 1, e))?
+            .resource_metrics,
+        );
+    }
+
+    if merged.resource_metrics.is_empty() {
+        return Err("No resource metrics found in file".to_string());
+    }
+    Ok(merged)
+}
+
+/// Counts data points per (metric name, OTLP type) and validates each point:
+/// non-empty name, non-zero timestamp, a value, and a round-tripped `metric.type` attribute.
+fn tally_and_validate(request: &ExportMetricsServiceRequest) -> (usize, usize) {
+    let mut gauge = 0;
+    let mut sum = 0;
+
+    for rm in &request.resource_metrics {
+        for sm in &rm.scope_metrics {
+            let scope = sm.scope.as_ref().expect("scope_metrics has no scope");
+            assert_eq!(scope.name, EXPECTED_SCOPE_NAME, "scope.name mismatch");
+            assert_eq!(
+                scope.version, EXPECTED_SCOPE_VERSION,
+                "scope.version mismatch"
+            );
+            for metric in &sm.metrics {
+                assert!(!metric.name.is_empty(), "metric name is empty");
+                match metric.data.as_ref().expect("metric has no data") {
+                    MetricData::Gauge(Gauge { data_points }) => {
+                        assert_eq!(metric.name, "gauge_metric");
+                        for dp in data_points {
+                            assert!(dp.time_unix_nano > 0, "gauge point has zero timestamp");
+                            assert!(dp.value.is_some(), "gauge point has no value");
+                            assert_metric_type_attr(&dp.attributes, "gauge");
+                        }
+                        gauge += data_points.len();
+                    }
+                    MetricData::Sum(Sum { data_points, .. }) => {
+                        assert_eq!(metric.name, "sum_metric");
+                        for dp in data_points {
+                            assert!(dp.time_unix_nano > 0, "sum point has zero timestamp");
+                            assert!(dp.value.is_some(), "sum point has no value");
+                            assert_metric_type_attr(&dp.attributes, "sum");
+                        }
+                        sum += data_points.len();
+                    }
+                    other => panic!("unexpected metric type for {}: {other:?}", metric.name),
+                }
+            }
+        }
+    }
+    (gauge, sum)
+}
+
+fn assert_metric_type_attr(
+    attributes: &[vector_lib::opentelemetry::proto::common::v1::KeyValue],
+    expected: &str,
+) {
+    let attr = attributes
+        .iter()
+        .find(|kv| kv.key == "metric.type")
+        .expect("missing 'metric.type' attribute");
+    match attr.value.as_ref().and_then(|v| v.value.as_ref()) {
+        Some(AnyValueEnum::StringValue(s)) => assert_eq!(s, expected),
+        other => panic!("'metric.type' is not the expected string: {other:?}"),
+    }
+}
+
+#[test]
+fn vector_native_metrics_encode_to_otlp() {
+    let content = read_file_helper("metrics-native", "collector-file-exporter.log")
+        .expect("Failed to read collector file");
+    let request = parse_export_metrics_request(&content)
+        .expect("Failed to parse collector metrics as ExportMetricsServiceRequest");
+
+    // service.name from the telemetrygen resource must survive the native round-trip.
+    assert_service_name_with(
+        &request.resource_metrics,
+        "resource_metrics",
+        "telemetrygen",
+        |rm| rm.resource.as_ref(),
+    );
+
+    let (gauge, sum) = tally_and_validate(&request);
+    assert_eq!(gauge, EXPECTED_GAUGE, "gauge_metric data point count");
+    assert_eq!(sum, EXPECTED_SUM, "sum_metric data point count");
+}

tests/e2e/opentelemetry/mod.rs

@@ -1,5 +1,6 @@
 pub mod logs;
 pub mod metrics;
+pub mod metrics_native;
 
 use std::{io, path::Path, process::Command};