feat(throttle): reimplement throttle transform with basic rate limiting

The native `throttle` transform is backed by the `governor` library
which provides a very flexible GCRA implementation for rate limiting.
The library does not provide any way to explicitly initialize the quota
state, however, meaning that we cannot rehydrate the initial state of
the transform from state persistence (rocksdb).

For our purposes, we do not need the flexibity and advanced features
(e.g. burst rate capacity) provided by this crate. Instead, this adds
our own implementation `mezmo_throttle` which is identical, but instead
uses a very simple leaky bucket implementation. The state is represented
by a `HashMap` storing types that can natively be represented as a
JSON string for storage.

Ref: LOG-19565

Author: mdeltito

Cargo.toml

@@ -661,6 +661,7 @@ transforms-metrics-mezmo = [
   "transforms-mezmo_aggregate",
   "transforms-mezmo_tag_cardinality_limit",
   "transforms-mezmo_log_to_metric",
+  "transforms-mezmo_throttle",
   "transforms-protobuf_to_metric",
 ]
 
@@ -680,6 +681,7 @@ transforms-mezmo_log_to_metric = []
 transforms-mezmo_log_clustering = ["dep:lru", "dep:blake2", "dep:base64", "dep:tokio-postgres"]
 transforms-mezmo_log_classification = ["dep:grok"]
 transforms-mezmo_tag_cardinality_limit = ["dep:bloomy", "dep:hashbrown"]
+transforms-mezmo_throttle = []
 transforms-remap = []
 transforms-route = []
 transforms-sample = []

src/transforms/mezmo_throttle/config.rs

@@ -0,0 +1,58 @@
+use super::*;
+
+/// Configuration for the `mezmo_throttle` transform.
+#[serde_as]
+#[configurable_component(transform(
+    "mezmo_throttle",
+    "Rate limit logs passing through a topology."
+))]
+#[derive(Clone, Debug, Default)]
+#[serde(deny_unknown_fields)]
+pub struct MezmoThrottleConfig {
+    /// The number of events allowed for a given bucket per configured `window_secs`.
+    ///
+    /// Each unique key has its own `threshold`.
+    pub(super) threshold: u32,
+
+    /// The time window in which the configured `threshold` is applied, in seconds.
+    #[serde_as(as = "serde_with::DurationSecondsWithFrac<f64>")]
+    #[configurable(metadata(docs::human_name = "Time Window"))]
+    pub(super) window_secs: Duration,
+
+    /// The value to group events into separate buckets to be rate limited independently.
+    ///
+    /// If left unspecified, or if the event doesn't have `key_field`, then the event is not rate
+    /// limited separately.
+    #[configurable(metadata(docs::examples = "{{ message }}", docs::examples = "{{ hostname }}",))]
+    pub(super) key_field: Option<Template>,
+
+    /// A logical condition used to exclude events from sampling.
+    pub(super) exclude: Option<AnyCondition>,
+}
+
+impl_generate_config_from_default!(MezmoThrottleConfig);
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "mezmo_throttle")]
+impl TransformConfig for MezmoThrottleConfig {
+    async fn build(&self, context: &TransformContext) -> crate::Result<Transform> {
+        Throttle::new(self, context, ThrottleClock {}).map(Transform::event_task)
+    }
+
+    fn input(&self) -> Input {
+        Input::log()
+    }
+
+    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,
+            clone_input_definitions(input_definitions),
+        )]
+    }
+}