feat(splunk_hec source): support second-stage framing and decoding

changelog.d/splunk_hec_source_codec.enhancement.md

@@ -0,0 +1,3 @@
+The `splunk_hec` source now accepts optional per-endpoint codec configuration via `event: { framing, decoding }` and `raw: { framing, decoding }`. When `decoding` is set on an endpoint, Vector applies a second decoding pass after the HEC envelope is parsed: on `/services/collector/event` the envelope's `event` field is fed through the codec (string contents are passed as raw bytes; objects, arrays, and other JSON values are JSON-serialized first to preserve shape), and on `/services/collector/raw` the request body is fed through the codec directly. A single payload can fan out to multiple events. Decoder-produced fields win on conflict with HEC envelope metadata, and decode failures do not surface to the Splunk client. Each endpoint is configured independently; when neither endpoint is configured the source preserves its existing behavior unchanged.
+
+authors: thomasqueirozb

lib/codecs/src/decoding/decoder.rs

@@ -1,7 +1,10 @@
 use bytes::{Bytes, BytesMut};
 use smallvec::SmallVec;
 use vector_common::internal_event::emit;
-use vector_core::{config::LogNamespace, event::Event};
+use vector_core::{
+    config::LogNamespace,
+    event::{Event, EventMetadata},
+};
 
 use crate::{
     decoding::format::Deserializer as _,
@@ -53,6 +56,17 @@ impl Decoder {
         self
     }
 
+    /// Attaches a per-decode-call metadata template to the inner deserializer.
+    ///
+    /// For deserializers that support it (currently only `VrlDeserializer`) the
+    /// template is pre-populated on the synthetic event before any user program
+    /// executes, making every `%`-prefixed path readable (e.g. `%splunk_hec.host`,
+    /// `%vector.secrets.*`). For all other deserializers this is a no-op.
+    pub fn with_metadata_template(mut self, metadata: EventMetadata) -> Self {
+        self.deserializer = self.deserializer.with_metadata_template(metadata);
+        self
+    }
+
     /// Handles the framing result and parses it into a structured event, if
     /// possible.
     ///

lib/codecs/src/decoding/format/vrl.rs

@@ -13,7 +13,9 @@ use vrl::{
     value::Kind,
 };
 
-use crate::{BytesDeserializerConfig, decoding::format::Deserializer};
+use vector_core::event::EventMetadata;
+
+use crate::decoding::format::Deserializer;
 
 /// Config used to build a `VrlDeserializer`.
 #[configurable_component]
@@ -64,6 +66,7 @@ impl VrlDeserializerConfig {
             Ok(result) => Ok(VrlDeserializer {
                 program: result.program,
                 timezone: self.vrl.timezone.unwrap_or(TimeZone::Local),
+                metadata_template: None,
             }),
             Err(diagnostics) => Err(Formatter::new(&self.vrl.source, diagnostics)
                 .to_string()
@@ -94,9 +97,28 @@ impl VrlDeserializerConfig {
 pub struct VrlDeserializer {
     program: Program,
     timezone: TimeZone,
+    /// Per-call metadata template set by the source before decoding. When
+    /// present, every `%`-prefixed path in the template is accessible from
+    /// within the VRL program (e.g. `%splunk_hec.host`, `%vector.secrets.*`).
+    metadata_template: Option<EventMetadata>,
+}
+
+impl VrlDeserializer {
+    /// Attach a metadata template that will be pre-populated on each synthetic
+    /// event before the VRL program runs.
+    ///
+    /// Sources call this once per decode call with the per-request context they
+    /// have assembled (envelope fields, auth tokens, etc.). VRL can then read
+    /// those values via `%`-prefixed paths.
+    #[must_use]
+    pub fn with_metadata_template(mut self, metadata: EventMetadata) -> Self {
+        self.metadata_template = Some(metadata);
+        self
+    }
 }
 
 fn parse_bytes(bytes: Bytes, log_namespace: LogNamespace) -> Event {
+    use crate::BytesDeserializerConfig;
     let bytes_deserializer = BytesDeserializerConfig::new().build();
     let log_event = bytes_deserializer.parse_single(bytes, log_namespace);
     Event::from(log_event)
@@ -108,11 +130,14 @@ impl Deserializer for VrlDeserializer {
         bytes: Bytes,
         log_namespace: LogNamespace,
     ) -> vector_common::Result<SmallVec<[Event; 1]>> {
-        let event = parse_bytes(bytes, log_namespace);
-        match self.run_vrl(event, log_namespace) {
-            Ok(events) => Ok(events),
-            Err(e) => Err(e),
+        let mut event = parse_bytes(bytes, log_namespace);
+        if let Some(template) = &self.metadata_template {
+            // Pre-populate the synthetic event with the source-assembled metadata so
+            // every `%`-prefixed path is in scope when VRL executes. This lets
+            // user programs read `%splunk_hec.host`, `%vector.secrets.*`, etc.
+            *event.metadata_mut() = template.clone();
         }
+        self.run_vrl(event, log_namespace)
     }
 }
 
@@ -320,4 +345,55 @@ mod tests {
             .to_string();
         assert!(error.contains("aborted"));
     }
+
+    fn metadata_with_secret(key: &str, value: &str) -> EventMetadata {
+        let mut metadata = EventMetadata::default();
+        metadata.secrets_mut().insert(key, value);
+        metadata
+    }
+
+    /// A VRL program that uses `get_secret!()` can read a secret injected via
+    /// `with_metadata_template`.
+    #[test]
+    fn test_with_metadata_template_vrl_can_read_secret() {
+        // VRL program copies the injected secret into an event field so we can
+        // assert on its value. The input bytes become `.message` (Legacy namespace)
+        // and we add `.secret_value` alongside it.
+        let decoder = make_decoder(r#".secret_value = get_secret!("my_token")"#)
+            .with_metadata_template(metadata_with_secret("my_token", "super-secret"));
+
+        let bytes = Bytes::from(r#"hello"#);
+        let events = decoder
+            .parse(bytes, LogNamespace::Legacy)
+            .expect("parse should succeed");
+
+        assert_eq!(events.len(), 1);
+        assert_eq!(
+            *events[0].as_log().get("secret_value").unwrap(),
+            Value::from("super-secret")
+        );
+    }
+
+    /// Secrets explicitly set by the VRL program win over the template because
+    /// `set_secret!` runs after the template is pre-populated.
+    #[test]
+    fn test_with_metadata_template_codec_wins_on_collision() {
+        let decoder = make_decoder(r#"set_secret!("my_token", "codec-wins")"#)
+            .with_metadata_template(metadata_with_secret("my_token", "template-loses"));
+
+        let bytes = Bytes::from(r#"hello"#);
+        let events = decoder
+            .parse(bytes, LogNamespace::Legacy)
+            .expect("parse should succeed");
+
+        assert_eq!(
+            events[0]
+                .metadata()
+                .secrets()
+                .get("my_token")
+                .unwrap()
+                .as_ref(),
+            "codec-wins"
+        );
+    }
 }

lib/codecs/src/decoding/mod.rs

@@ -37,7 +37,7 @@ use smallvec::SmallVec;
 use vector_config::configurable_component;
 use vector_core::{
     config::{DataType, LogNamespace},
-    event::Event,
+    event::{Event, EventMetadata},
     schema,
 };
 
@@ -426,6 +426,12 @@ impl DeserializerConfig {
         }
     }
 
+    /// Returns `true` when this is a VRL deserializer.
+    /// Sources can use this to decide whether to call `Decoder::with_metadata_template`.
+    pub fn is_vrl(&self) -> bool {
+        matches!(self, DeserializerConfig::Vrl(_))
+    }
+
     /// Return the type of event build by this deserializer.
     pub fn output_type(&self) -> DataType {
         match self {
@@ -542,6 +548,18 @@ pub enum Deserializer {
     Vrl(VrlDeserializer),
 }
 
+impl Deserializer {
+    /// Attaches a metadata template to the inner deserializer, if it supports
+    /// one. Currently only [`VrlDeserializer`] uses this; for all other variants
+    /// this is a no-op and `self` is returned unchanged.
+    pub fn with_metadata_template(self, metadata: EventMetadata) -> Self {
+        match self {
+            Deserializer::Vrl(d) => Deserializer::Vrl(d.with_metadata_template(metadata)),
+            other => other,
+        }
+    }
+}
+
 impl format::Deserializer for Deserializer {
     fn parse(
         &self,

lib/vector-core/src/event/mod.rs

@@ -7,7 +7,7 @@ pub use finalization::{
     Finalizable,
 };
 pub use log_event::LogEvent;
-pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, WithMetadata};
+pub use metadata::{DatadogMetricOriginMetadata, EventMetadata, Secrets, WithMetadata};
 pub use metric::{Metric, MetricKind, MetricTags, MetricValue, StatisticKind};
 pub use r#ref::{EventMutRef, EventRef};
 use serde::{Deserialize, Serialize};