chore(config): Pass the extra context to sources and transforms too (v…

…ectordotdev#19779)

* chore(config): Pass the extra context to sources and transforms too

* Add comments on new members

* Fix compile error in socket source

src/config/sink.rs

@@ -238,6 +238,9 @@ pub struct SinkContext {
     pub schema: schema::Options,
     pub app_name: String,
     pub app_name_slug: String,
+
+    /// Extra context data provided by the running app and shared across all components. This can be
+    /// used to pass shared settings or other data from outside the components.
     pub extra_context: ExtraContext,
 }
 

src/config/source.rs

@@ -17,7 +17,7 @@ use vector_lib::{
 };
 
 use super::{schema, ComponentKey, ProxyConfig, Resource};
-use crate::{shutdown::ShutdownSignal, SourceSender};
+use crate::{extra_context::ExtraContext, shutdown::ShutdownSignal, SourceSender};
 
 pub type BoxedSource = Box<dyn SourceConfig>;
 
@@ -131,6 +131,10 @@ pub struct SourceContext {
     /// Given a source can expose multiple [`SourceOutput`] channels, the ID is tied to the identifier of
     /// that `SourceOutput`.
     pub schema_definitions: HashMap<Option<String>, schema::Definition>,
+
+    /// Extra context data provided by the running app and shared across all components. This can be
+    /// used to pass shared settings or other data from outside the components.
+    pub extra_context: ExtraContext,
 }
 
 impl SourceContext {
@@ -151,6 +155,7 @@ impl SourceContext {
                 acknowledgements: false,
                 schema_definitions: HashMap::default(),
                 schema: Default::default(),
+                extra_context: Default::default(),
             },
             shutdown,
         )
@@ -170,6 +175,7 @@ impl SourceContext {
             acknowledgements: false,
             schema_definitions: schema_definitions.unwrap_or_default(),
             schema: Default::default(),
+            extra_context: Default::default(),
         }
     }
 

src/config/transform.rs

@@ -19,6 +19,7 @@ use vector_lib::{
 use super::schema::Options as SchemaOptions;
 use super::OutputId;
 use super::{id::Inputs, ComponentKey};
+use crate::extra_context::ExtraContext;
 
 pub type BoxedTransform = Box<dyn TransformConfig>;
 
@@ -96,7 +97,6 @@ where
     }
 }
 
-#[derive(Debug)]
 pub struct TransformContext {
     // This is optional because currently there are a lot of places we use `TransformContext` that
     // may not have the relevant data available (e.g. tests). In the future it'd be nice to make it
@@ -121,6 +121,10 @@ pub struct TransformContext {
     pub merged_schema_definition: schema::Definition,
 
     pub schema: SchemaOptions,
+
+    /// Extra context data provided by the running app and shared across all components. This can be
+    /// used to pass shared settings or other data from outside the components.
+    pub extra_context: ExtraContext,
 }
 
 impl Default for TransformContext {
@@ -132,6 +136,7 @@ impl Default for TransformContext {
             schema_definitions: HashMap::from([(None, HashMap::new())]),
             merged_schema_definition: schema::Definition::any(),
             schema: SchemaOptions::default(),
+            extra_context: Default::default(),
         }
     }
 }

src/sources/socket/mod.rs

@@ -948,6 +948,7 @@ mod test {
                 acknowledgements: false,
                 schema: Default::default(),
                 schema_definitions: HashMap::default(),
+                extra_context: Default::default(),
             })
             .await
             .unwrap();

src/topology/builder.rs

@@ -330,6 +330,7 @@ impl<'a> Builder<'a> {
                 acknowledgements: source.sink_acknowledgements,
                 schema_definitions,
                 schema: self.config.schema,
+                extra_context: self.extra_context.clone(),
             };
             let source = source.inner.build(context).await;
             let server = match source {
@@ -464,6 +465,7 @@ impl<'a> Builder<'a> {
                 schema_definitions,
                 merged_schema_definition: merged_definition.clone(),
                 schema: self.config.schema,
+                extra_context: self.extra_context.clone(),
             };
 
             let node = TransformNode::from_parts(