feat(new transform): Add buffered gate transform

Cargo.toml

@@ -608,6 +608,7 @@ transforms-logs = [
   "transforms-aws_ec2_metadata",
   "transforms-dedupe",
   "transforms-filter",
+  "transforms-gate",
   "transforms-log_to_metric",
   "transforms-lua",
   "transforms-metric_to_log",
@@ -634,6 +635,7 @@ transforms-aggregate = []
 transforms-aws_ec2_metadata = ["dep:arc-swap"]
 transforms-dedupe = ["transforms-impl-dedupe"]
 transforms-filter = []
+transforms-gate = []
 transforms-log_to_metric = []
 transforms-lua = ["dep:mlua", "vector-lib/lua"]
 transforms-metric_to_log = []
@@ -647,6 +649,7 @@ transforms-throttle = ["dep:governor"]
 
 # Implementations of transforms
 transforms-impl-sample = []
+transforms-impl-gate = []
 transforms-impl-dedupe = ["dep:lru"]
 transforms-impl-reduce = []
 

src/internal_events/gate.rs

@@ -0,0 +1,14 @@
+use vector_lib::internal_event::{ComponentEventsDropped, Count, INTENTIONAL, Registered};
+
+vector_lib::registered_event!(
+    GateEventsDropped => {
+        events_dropped: Registered<ComponentEventsDropped<'static, INTENTIONAL>>
+            = register!(ComponentEventsDropped::<INTENTIONAL>::from(
+                "The gate was closed."
+            )),
+    }
+
+    fn emit(&self, data: Count) {
+        self.events_dropped.emit(data);
+    }
+);

src/internal_events/mod.rs

@@ -53,6 +53,8 @@ mod exec;
 mod file_descriptor;
 #[cfg(feature = "transforms-filter")]
 mod filter;
+#[cfg(feature = "transforms-gate")]
+mod gate;
 #[cfg(feature = "sources-fluent")]
 mod fluent;
 #[cfg(feature = "sources-gcp_pubsub")]
@@ -195,6 +197,8 @@ pub(crate) use self::file::*;
 pub(crate) use self::file_descriptor::*;
 #[cfg(feature = "transforms-filter")]
 pub(crate) use self::filter::*;
+#[cfg(feature = "transforms-gate")]
+pub(crate) use self::gate::*;
 #[cfg(feature = "sources-fluent")]
 pub(crate) use self::fluent::*;
 #[cfg(feature = "sources-gcp_pubsub")]

src/transforms/gate/config.rs

@@ -0,0 +1,96 @@
+use vector_lib::config::{clone_input_definitions, LogNamespace};
+use vector_lib::configurable::configurable_component;
+
+use crate::{
+    conditions::AnyCondition,
+    config::{
+        DataType, GenerateConfig, Input, OutputId, TransformConfig, TransformContext,
+        TransformOutput,
+    },
+    schema,
+    transforms::Transform,
+};
+
+use super::transform::Gate;
+
+
+/// Configuration for the `gate` transform.
+#[configurable_component(transform(
+"gate",
+"Open or close an event stream based on supplied criteria"
+))]
+#[derive(Clone, Debug)]
+#[serde(deny_unknown_fields)]
+pub struct GateConfig {
+    /// A logical condition used to pass events through without gating.
+    pub pass_when: Option<AnyCondition>,
+
+    /// A logical condition used to open the gate.
+    pub open_when: Option<AnyCondition>,
+
+    /// A logical condition used to close the gate.
+    pub close_when: Option<AnyCondition>,
+
+    /// Maximum number of events to keep in the buffer.
+    pub max_events: Option<usize>,
+
+    /// Automatically close the gate after the buffer has been flushed.
+    pub auto_close: Option<bool>,
+
+    /// Keep the gate open for additional number of events after the buffer has been flushed.
+    pub tail_events: Option<usize>,
+}
+
+impl GenerateConfig for GateConfig {
+    fn generate_config() -> toml::Value {
+        toml::Value::try_from(Self {
+            pass_when: None::<AnyCondition>,
+            open_when: None::<AnyCondition>,
+            close_when: None::<AnyCondition>,
+            max_events: None::<usize>,
+            auto_close: None::<bool>,
+            tail_events: None::<usize>,
+        }).unwrap()
+    }
+}
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "gate")]
+impl TransformConfig for GateConfig {
+    async fn build(&self, context: &TransformContext) -> crate::Result<Transform> {
+        Ok(Transform::function(Gate::new(
+            self.pass_when
+                .as_ref()
+                .map(|condition| condition.build(&context.enrichment_tables))
+                .transpose()?,
+            self.open_when
+                .as_ref()
+                .map(|condition| condition.build(&context.enrichment_tables))
+                .transpose()?,
+            self.close_when
+                .as_ref()
+                .map(|condition| condition.build(&context.enrichment_tables))
+                .transpose()?,
+            self.max_events.unwrap_or(100),
+            self.auto_close.unwrap_or(true),
+            self.tail_events.unwrap_or(10),
+        ).unwrap()))
+    }
+
+    fn input(&self) -> Input {
+        Input::new(DataType::Log | DataType::Trace)
+    }
+
+    fn outputs(
+        &self,
+        _: vector_lib::enrichment::TableRegistry,
+        input_definitions: &[(OutputId, schema::Definition)],
+        _: LogNamespace,
+    ) -> Vec<TransformOutput> {
+        // The event is not modified, so the definition is passed through as-is
+        vec![TransformOutput::new(
+            DataType::Log | DataType::Trace,
+            clone_input_definitions(input_definitions),
+        )]
+    }
+}

src/transforms/gate/mod.rs

@@ -0,0 +1,3 @@
+pub mod config;
+pub mod transform;
+mod state;

src/transforms/gate/state.rs

@@ -0,0 +1,14 @@
+use crate::sinks::prelude::configurable_component;
+
+/// Gate state
+#[configurable_component]
+#[derive(Clone, Debug, Copy)]
+#[derive(PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub enum GateState {
+    /// Gate is open
+    Open,
+
+    /// Gate is closed
+    Closed,
+}