feat(sources): post-processing in SourceSender

Add a `PostProcessor` hook to `SourceSender` so sources can attach a VRL
program or a hard-coded Rust closure that runs on every event after schema
metadata is attached, immediately before the event is placed on the output
channel. This provides a first-class, reusable facility (e.g. for HEC token
enrichment) instead of requiring per-source workarounds.

Author: 20agbekodo

changelog.d/source_sender_post_processor.feature.md

@@ -0,0 +1,5 @@
+Add a `PostProcessor` hook to `SourceSender` that allows a hard-coded Rust closure to be applied
+to every event emitted by a source, immediately after schema metadata is attached and before the
+event is placed on the output channel.
+
+authors: 20agbekodo

lib/vector-core/src/source_sender/builder.rs

@@ -5,8 +5,8 @@ use vector_common::histogram;
 use vector_common::internal_event::DEFAULT_OUTPUT;
 
 use super::{
-    CHUNK_SIZE, LAG_TIME_NAME, Output, OutputMetrics, SEND_BATCH_LATENCY_NAME, SEND_LATENCY_NAME,
-    SourceSender, SourceSenderItem,
+    CHUNK_SIZE, LAG_TIME_NAME, Output, OutputMetrics, PostProcessor, SEND_BATCH_LATENCY_NAME,
+    SEND_LATENCY_NAME, SourceSender, SourceSenderItem,
 };
 use crate::config::{ComponentKey, OutputId, SourceOutput};
 
@@ -17,6 +17,7 @@ pub struct Builder {
     output_metrics: OutputMetrics,
     timeout: Option<Duration>,
     ewma_half_life_seconds: Option<f64>,
+    post_processor: Option<PostProcessor>,
 }
 
 impl Default for Builder {
@@ -32,6 +33,7 @@ impl Default for Builder {
             ),
             timeout: None,
             ewma_half_life_seconds: None,
+            post_processor: None,
         }
     }
 }
@@ -55,6 +57,18 @@ impl Builder {
         self
     }
 
+    /// Attach a post-processing step that will be applied to every event on **all** outputs
+    /// (default and named ports) produced by this builder.
+    ///
+    /// The processor runs after schema metadata has been attached to each event, immediately
+    /// before the event is placed on the output channel. See [`PostProcessor`] for the available
+    /// variants and their error-handling semantics.
+    #[must_use]
+    pub fn with_post_processor(mut self, post_processor: PostProcessor) -> Self {
+        self.post_processor = Some(post_processor);
+        self
+    }
+
     pub fn add_source_output(
         &mut self,
         output: SourceOutput,
@@ -75,6 +89,7 @@ impl Builder {
                     output_id,
                     self.timeout,
                     self.ewma_half_life_seconds,
+                    self.post_processor.clone(),
                 );
                 self.default_output = Some(output);
                 rx
@@ -88,6 +103,7 @@ impl Builder {
                     output_id,
                     self.timeout,
                     self.ewma_half_life_seconds,
+                    self.post_processor.clone(),
                 );
                 self.named_outputs.insert(name, output);
                 rx

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

@@ -27,3 +27,24 @@ use vector_common::internal_event::HistogramName;
 const LAG_TIME_NAME: HistogramName = HistogramName::SourceLagTimeSeconds;
 const SEND_LATENCY_NAME: HistogramName = HistogramName::SourceSendLatencySeconds;
 const SEND_BATCH_LATENCY_NAME: HistogramName = HistogramName::SourceSendBatchLatencySeconds;
+
+/// A post-processing step applied to every event that flows through a [`SourceSender`].
+///
+/// The hook executes on each event after schema metadata has been attached, immediately before the
+/// event is placed on the output channel. It is applied *globally* — to all outputs (default and
+/// named ports) produced by the same [`Builder`].
+///
+/// Currently one variant is provided:
+///
+/// - [`PostProcessor::HardCoded`]: calls an infallible Rust closure. No events are dropped by
+///   this variant.
+///
+/// If per-output post-processing is needed in the future, a `with_post_processor_for_port` API
+/// can be added without breaking this interface.
+#[derive(Clone)]
+pub enum PostProcessor {
+    /// Call a hard-coded Rust function against every event.
+    ///
+    /// The closure is infallible; no events are dropped by this variant.
+    HardCoded(std::sync::Arc<dyn Fn(&mut crate::event::Event) + Send + Sync>),
+}

lib/vector-core/src/source_sender/output.rs

@@ -24,7 +24,7 @@ use vector_common::{
 };
 use vrl::value::Value;
 
-use super::{CHUNK_SIZE, SendError, SourceSenderItem};
+use super::{CHUNK_SIZE, PostProcessor, SendError, SourceSenderItem};
 use crate::{
     EstimatedJsonEncodedSizeOf,
     config::{OutputId, log_schema},
@@ -93,6 +93,8 @@ pub(super) struct Output {
     /// `EventMetadata` for all event sent through here.
     id: Arc<OutputId>,
     timeout: Option<Duration>,
+    /// Optional post-processing step applied to every event before it is placed on the channel.
+    post_processor: Option<PostProcessor>,
 }
 
 #[derive(Clone, Default)]
@@ -129,6 +131,7 @@ impl fmt::Debug for Output {
 }
 
 impl Output {
+    #[expect(clippy::too_many_arguments)]
     pub(super) fn new_with_buffer(
         n: usize,
         output: String,
@@ -137,6 +140,7 @@ impl Output {
         output_id: OutputId,
         timeout: Option<Duration>,
         ewma_half_life_seconds: Option<f64>,
+        post_processor: Option<PostProcessor>,
     ) -> (Self, LimitedReceiver<SourceSenderItem>) {
         let limit = MemoryBufferSize::MaxEvents(NonZeroUsize::new(n).unwrap());
         let channel_metrics =
@@ -152,6 +156,7 @@ impl Output {
                 log_definition,
                 id: Arc::new(output_id),
                 timeout,
+                post_processor,
             },
             rx,
         )
@@ -168,7 +173,7 @@ impl Output {
 
     async fn send_inner(
         &mut self,
-        mut events: EventArray,
+        events: EventArray,
         unsent_event_count: &mut UnsentEventCount,
         reference: i64,
     ) -> Result<(), SendError> {
@@ -177,29 +182,49 @@ impl Output {
             .iter_events()
             .for_each(|event| self.emit_lag_time(event, reference));
 
-        events.iter_events_mut().for_each(|mut event| {
-            // attach runtime schema definitions from the source
-            if let Some(log_definition) = &self.log_definition {
-                event.metadata_mut().set_schema_definition(log_definition);
-            }
-            event.metadata_mut().set_upstream_id(Arc::clone(&self.id));
-        });
+        let post_processor = self.post_processor.clone();
+        let log_definition = self.log_definition.clone();
+        let id = Arc::clone(&self.id);
+
+        // Iterate over individual events to apply metadata and the optional post-processor.
+        // We collect into a Vec<Event> first so that we can then re-group them into
+        // EventArrays (which are homogeneous by type). We send each sub-array separately to
+        // handle mixed-type batches correctly — no events are silently dropped.
+        let processed: Vec<Event> = events
+            .into_events()
+            .map(|mut event| {
+                if let Some(ref def) = log_definition {
+                    event.metadata_mut().set_schema_definition(def);
+                }
+                event.metadata_mut().set_upstream_id(Arc::clone(&id));
+                if let Some(PostProcessor::HardCoded(ref f)) = post_processor {
+                    f(&mut event);
+                }
+                event
+            })
+            .collect();
 
-        let byte_size = events.estimated_json_encoded_size_of();
-        let count = events.len();
+        if processed.is_empty() {
+            return Ok(());
+        }
 
+        // Re-group individual events back into typed EventArrays and send each one.
+        // Using `events_into_arrays(...).collect::<Vec<_>>()` ensures ALL sub-arrays are
+        // captured — no type group is silently discarded (avoids the `.next()`-only bug).
         let send_start = Instant::now();
+        for sub_array in array::events_into_arrays(processed.into_iter(), None) {
+            let byte_size = sub_array.estimated_json_encoded_size_of();
+            let count = sub_array.len();
 
-        let send_result = self.send_with_timeout(events, send_reference).await;
+            self.send_with_timeout(sub_array, send_reference).await?;
 
+            self.events_sent.emit(CountByteSize(count, byte_size));
+            unsent_event_count.decr(count);
+        }
         if let Some(send_latency) = &self.metrics.send_latency {
             send_latency.record(send_start.elapsed().as_secs_f64());
         }
 
-        send_result?;
-
-        self.events_sent.emit(CountByteSize(count, byte_size));
-        unsent_event_count.decr(count);
         Ok(())
     }
 
@@ -295,6 +320,11 @@ impl Output {
         Ok(())
     }
 
+    /// Attach a post-processing step to this output, replacing any previously set one.
+    pub(super) fn set_post_processor(&mut self, pp: &PostProcessor) {
+        self.post_processor = Some(pp.clone());
+    }
+
     /// Calculate the difference between the reference time and the
     /// timestamp stored in the given event reference, and emit the
     /// different, as expressed in milliseconds, as a histogram.
@@ -336,3 +366,4 @@ const fn get_timestamp_millis(value: &Value) -> Option<i64> {
         _ => None,
     }
 }
+

lib/vector-core/src/source_sender/sender.rs

@@ -20,7 +20,7 @@ use vector_common::{
     json_size::JsonSize,
 };
 
-use super::{Builder, Output, SendError};
+use super::{Builder, Output, PostProcessor, SendError};
 #[cfg(any(test, feature = "test"))]
 use super::{
     LAG_TIME_NAME, OutputMetrics, SEND_BATCH_LATENCY_NAME, SEND_LATENCY_NAME, TEST_BUFFER_SIZE,
@@ -104,6 +104,20 @@ impl SourceSender {
         Builder::default()
     }
 
+    /// Attach a post-processing step to every output on this sender.
+    ///
+    /// The processor runs after schema metadata has been attached to each event, immediately
+    /// before the event is placed on the output channel. Replaces any previously set
+    /// post-processor.
+    pub fn set_post_processor(&mut self, pp: &PostProcessor) {
+        if let Some(output) = &mut self.default_output {
+            output.set_post_processor(pp);
+        }
+        for output in self.named_outputs.values_mut() {
+            output.set_post_processor(pp);
+        }
+    }
+
     #[cfg(any(test, feature = "test"))]
     pub fn new_test_sender_with_options(
         n: usize,
@@ -124,6 +138,37 @@ impl SourceSender {
             output_id,
             timeout,
             None,
+            None,
+        );
+        (
+            Self {
+                default_output: Some(default_output),
+                named_outputs: Default::default(),
+            },
+            rx,
+        )
+    }
+
+    #[cfg(any(test, feature = "test"))]
+    pub fn new_test_with_post_processor(
+        post_processor: super::PostProcessor,
+    ) -> (Self, LimitedReceiver<SourceSenderItem>) {
+        let lag_time = Some(histogram!(LAG_TIME_NAME));
+        let send_latency = Some(histogram!(SEND_LATENCY_NAME));
+        let send_batch_latency = Some(histogram!(SEND_BATCH_LATENCY_NAME));
+        let output_id = OutputId {
+            component: "test".to_string().into(),
+            port: None,
+        };
+        let (default_output, rx) = Output::new_with_buffer(
+            TEST_BUFFER_SIZE,
+            DEFAULT_OUTPUT.to_owned(),
+            OutputMetrics::new(lag_time, send_latency, send_batch_latency),
+            None,
+            output_id,
+            None,
+            None,
+            Some(post_processor),
         );
         (
             Self {
@@ -204,6 +249,7 @@ impl SourceSender {
             output_id,
             None,
             None,
+            None,
         );
         let recv = recv.into_stream().map(move |mut item| {
             item.events.iter_events_mut().for_each(|mut event| {