feat(opentelemetry source): Support per-signal OTLP decoding configuration (#24455)

Allow independent configuration of OTLP decoding for logs, metrics, and traces.
Maintains backward compatibility with boolean configuration.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: pront

src/sources/opentelemetry/config.rs

@@ -47,6 +47,100 @@ pub const LOGS: &str = "logs";
 pub const METRICS: &str = "metrics";
 pub const TRACES: &str = "traces";
 
+/// Configuration for OTLP decoding behavior.
+///
+/// This configuration controls how OpenTelemetry Protocol (OTLP) data is decoded for each
+/// signal type (logs, metrics, traces).
+///
+/// When a signal is configured to use OTLP decoding (`true`), the raw OTLP format is preserved,
+/// allowing the data to be forwarded to downstream OTLP collectors without transformation.
+/// When set to `false`, the signal is converted to Vector's native event format, enabling
+/// compatibility with Vector's transforms and sinks.
+///
+/// # Important Considerations
+///
+/// When OTLP decoding is enabled for metrics:
+/// - Metrics are parsed as logs while preserving the OTLP format
+/// - Vector's metric transforms will NOT be compatible with this output
+/// - The events can be forwarded directly to a downstream OTLP collector
+///
+/// # Configuration Examples
+///
+/// This configuration supports both a simple boolean form (for backward compatibility)
+/// and a per-signal configuration:
+///
+/// ## Simple boolean form (applies to all signals):
+/// ```yaml
+/// use_otlp_decoding: true  # All signals preserve OTLP format
+/// # or
+/// use_otlp_decoding: false # All signals use Vector native format (default)
+/// ```
+///
+/// ## Per-signal configuration:
+/// ```yaml
+/// use_otlp_decoding:
+///   logs: false     # Convert to Vector native format
+///   metrics: false  # Convert to Vector native format
+///   traces: true    # Preserve OTLP format
+/// ```
+#[configurable_component]
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct OtlpDecodingConfig {
+    /// Whether to use OTLP decoding for logs.
+    ///
+    /// When `true`, logs preserve their OTLP format.
+    /// When `false` (default), logs are converted to Vector's native format.
+    #[serde(default)]
+    pub logs: bool,
+
+    /// Whether to use OTLP decoding for metrics.
+    ///
+    /// When `true`, metrics preserve their OTLP format but are processed as logs.
+    /// When `false` (default), metrics are converted to Vector's native metric format.
+    #[serde(default)]
+    pub metrics: bool,
+
+    /// Whether to use OTLP decoding for traces.
+    ///
+    /// When `true`, traces preserve their OTLP format.
+    /// When `false` (default), traces are converted to Vector's native format.
+    #[serde(default)]
+    pub traces: bool,
+}
+
+impl From<bool> for OtlpDecodingConfig {
+    /// Converts a boolean value to an OtlpDecodingConfig.
+    ///
+    /// This provides backward compatibility with the previous boolean configuration.
+    /// - `true` enables OTLP decoding for all signals
+    /// - `false` disables OTLP decoding for all signals (uses Vector native format)
+    fn from(value: bool) -> Self {
+        Self {
+            logs: value,
+            metrics: value,
+            traces: value,
+        }
+    }
+}
+
+impl OtlpDecodingConfig {
+    /// Returns true if any signal is configured to use OTLP decoding.
+    pub const fn any_enabled(&self) -> bool {
+        self.logs || self.metrics || self.traces
+    }
+
+    /// Returns true if all signals are configured to use OTLP decoding.
+    pub const fn all_enabled(&self) -> bool {
+        self.logs && self.metrics && self.traces
+    }
+
+    /// Returns true if signals have mixed configuration (some enabled, some disabled).
+    pub const fn is_mixed(&self) -> bool {
+        self.any_enabled() && !self.all_enabled()
+    }
+}
+
 /// Configuration for the `opentelemetry` source.
 #[configurable_component(source("opentelemetry", "Receive OTLP data through gRPC or HTTP."))]
 #[derive(Clone, Debug)]
@@ -67,14 +161,14 @@ pub struct OpentelemetryConfig {
     #[serde(default)]
     pub log_namespace: Option<bool>,
 
-    /// Setting this field will override the legacy mapping of OTEL protos to Vector events and use the proto directly.
+    /// Configuration for how OTLP data is decoded for each signal type.
     ///
-    /// One major caveat here is that the incoming metrics will be parsed as logs but they will preserve the OTLP format.
-    /// This means that components that work on metrics, will not be compatible with this output.
-    /// However, these events can be forwarded directly to a downstream OTEL collector.
+    /// This field supports both a simple boolean form (for backward compatibility) and a
+    /// per-signal configuration. See [`OtlpDecodingConfig`] for detailed documentation
+    /// and configuration examples.
     #[configurable(derived)]
-    #[serde(default)]
-    pub use_otlp_decoding: bool,
+    #[serde(default, deserialize_with = "bool_or_struct")]
+    pub use_otlp_decoding: OtlpDecodingConfig,
 }
 
 /// Configuration for the `opentelemetry` gRPC server.
@@ -152,18 +246,24 @@ impl GenerateConfig for OpentelemetryConfig {
             http: example_http_config(),
             acknowledgements: Default::default(),
             log_namespace: None,
-            use_otlp_decoding: false,
+            use_otlp_decoding: OtlpDecodingConfig::default(),
         })
         .unwrap()
     }
 }
 
 impl OpentelemetryConfig {
-    fn get_signal_deserializer(
+    pub(crate) fn get_signal_deserializer(
         &self,
         signal_type: OtlpSignalType,
     ) -> vector_common::Result<Option<OtlpDeserializer>> {
-        if self.use_otlp_decoding {
+        let should_use_otlp = match signal_type {
+            OtlpSignalType::Logs => self.use_otlp_decoding.logs,
+            OtlpSignalType::Metrics => self.use_otlp_decoding.metrics,
+            OtlpSignalType::Traces => self.use_otlp_decoding.traces,
+        };
+
+        if should_use_otlp {
             Ok(Some(OtlpDeserializer::new_with_signals(IndexSet::from([
                 signal_type,
             ]))))
@@ -183,6 +283,17 @@ impl SourceConfig for OpentelemetryConfig {
 
         let grpc_tls_settings = MaybeTlsSettings::from_config(self.grpc.tls.as_ref(), true)?;
 
+        // Log info message when using mixed OTLP decoding formats
+        if self.use_otlp_decoding.is_mixed() {
+            info!(
+                message = "Using mixed OTLP decoding configuration.",
+                logs_otlp = self.use_otlp_decoding.logs,
+                metrics_otlp = self.use_otlp_decoding.metrics,
+                traces_otlp = self.use_otlp_decoding.traces,
+                note = "Signals with OTLP decoding enabled will preserve raw format; others will use Vector native format."
+            );
+        }
+
         let logs_deserializer = self.get_signal_deserializer(OtlpSignalType::Logs)?;
         let metrics_deserializer = self.get_signal_deserializer(OtlpSignalType::Metrics)?;
         let traces_deserializer = self.get_signal_deserializer(OtlpSignalType::Traces)?;
@@ -352,16 +463,25 @@ impl SourceConfig for OpentelemetryConfig {
             }
         };
 
-        let metrics_output = if self.use_otlp_decoding {
+        let logs_output = if self.use_otlp_decoding.logs {
+            SourceOutput::new_maybe_logs(DataType::Log, Definition::any()).with_port(LOGS)
+        } else {
+            SourceOutput::new_maybe_logs(DataType::Log, schema_definition).with_port(LOGS)
+        };
+
+        let metrics_output = if self.use_otlp_decoding.metrics {
             SourceOutput::new_maybe_logs(DataType::Log, Definition::any()).with_port(METRICS)
         } else {
             SourceOutput::new_metrics().with_port(METRICS)
         };
-        vec![
-            SourceOutput::new_maybe_logs(DataType::Log, schema_definition).with_port(LOGS),
-            metrics_output,
-            SourceOutput::new_traces().with_port(TRACES),
-        ]
+
+        let traces_output = if self.use_otlp_decoding.traces {
+            SourceOutput::new_maybe_logs(DataType::Log, Definition::any()).with_port(TRACES)
+        } else {
+            SourceOutput::new_traces().with_port(TRACES)
+        };
+
+        vec![logs_output, metrics_output, traces_output]
     }
 
     fn resources(&self) -> Vec<Resource> {