feat(opentelemetry source): allow to add headers to metadata for OpenTelemetry metrics and traces

changelog.d/24619-opentelemetry-header-enrichment.enhancement.md

@@ -0,0 +1,7 @@
+`opentelemetry` source: Implemented header enrichment for OTLP metrics and traces. Unlike logs, which support enriching
+the event itself or its metadata, depending on `log_namespace` settings, for metrics and traces this setting is ignored
+and header values are added to the event metadata.
+
+Issue: https://github.com/vectordotdev/vector/issues/24619
+
+authors: ozanichkovsky

src/sources/opentelemetry/config.rs

@@ -202,13 +202,14 @@ pub struct HttpConfig {
     #[serde(default)]
     pub keepalive: KeepaliveConfig,
 
-    /// A list of HTTP headers to include in the log event.
+    /// A list of HTTP headers to include in the event.
     ///
     /// Accepts the wildcard (`*`) character for headers matching a specified pattern.
     ///
-    /// Specifying "*" results in all headers included in the log event.
+    /// Specifying "*" results in all headers included in the event.
     ///
-    /// These headers are not included in the JSON payload if a field with a conflicting name exists.
+    /// For log events in legacy namespace mode, headers are not included if a field with a conflicting name exists.
+    /// For metrics and traces, headers are always added to event metadata.
     #[serde(default)]
     #[configurable(metadata(docs::examples = "User-Agent"))]
     #[configurable(metadata(docs::examples = "X-My-Custom-Header"))]

src/sources/opentelemetry/http.rs

@@ -109,16 +109,19 @@ pub(crate) fn build_warp_filter(
     );
     let metrics_filters = build_warp_metrics_filter(
         acknowledgements,
+        log_namespace,
         out.clone(),
         bytes_received.clone(),
         events_received.clone(),
+        headers.clone(),
         metrics_deserializer,
     );
     let trace_filters = build_warp_trace_filter(
         acknowledgements,
         out.clone(),
         bytes_received,
         events_received,
+        headers.clone(),
         traces_deserializer,
     );
     log_filters
@@ -257,12 +260,14 @@ fn build_warp_log_filter(
 }
 fn build_warp_metrics_filter(
     acknowledgements: bool,
+    log_namespace: LogNamespace,
     source_sender: SourceSender,
     bytes_received: Registered<BytesReceived>,
     events_received: Registered<EventsReceived>,
+    headers_cfg: Vec<HttpConfigParamKind>,
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
-    let make_events = move |encoding_header: Option<String>, _headers: HeaderMap, body: Bytes| {
+    let make_events = move |encoding_header: Option<String>, headers: HeaderMap, body: Bytes| {
         decompress_body(encoding_header.as_deref(), body)
             .inspect_err(|err| {
                 // Other status codes are already handled by `sources::util::decompress_body` (tech debt).
@@ -276,15 +281,14 @@ fn build_warp_metrics_filter(
             .and_then(|decoded_body| {
                 bytes_received.emit(ByteSize(decoded_body.len()));
                 if let Some(d) = deserializer.as_ref() {
-                    parse_with_deserializer(
-                        d,
-                        decoded_body,
-                        LogNamespace::default(),
-                        &events_received,
-                    )
+                    parse_with_deserializer(d, decoded_body, log_namespace, &events_received)
                 } else {
                     decode_metrics_body(decoded_body, &events_received)
                 }
+                .map(|mut events| {
+                    enrich_events(&mut events, &headers_cfg, &headers, log_namespace);
+                    events
+                })
             })
     };
 
@@ -301,9 +305,10 @@ fn build_warp_trace_filter(
     source_sender: SourceSender,
     bytes_received: Registered<BytesReceived>,
     events_received: Registered<EventsReceived>,
+    headers_cfg: Vec<HttpConfigParamKind>,
     deserializer: Option<OtlpDeserializer>,
 ) -> BoxedFilter<(Response,)> {
-    let make_events = move |encoding_header: Option<String>, _headers: HeaderMap, body: Bytes| {
+    let make_events = move |encoding_header: Option<String>, headers: HeaderMap, body: Bytes| {
         decompress_body(encoding_header.as_deref(), body)
             .inspect_err(|err| {
                 // Other status codes are already handled by `sources::util::decompress_body` (tech debt).
@@ -326,6 +331,10 @@ fn build_warp_trace_filter(
                 } else {
                     decode_trace_body(decoded_body, &events_received)
                 }
+                .map(|mut events| {
+                    enrich_events(&mut events, &headers_cfg, &headers, LogNamespace::default());
+                    events
+                })
             })
     };