vector/transforms/throttle/
config.rs

1use governor::clock;
2use serde_with::serde_as;
3use std::time::Duration;
4use vector_lib::config::{clone_input_definitions, LogNamespace};
5use vector_lib::configurable::configurable_component;
6
7use super::transform::Throttle;
8use crate::{
9    conditions::AnyCondition,
10    config::{DataType, Input, OutputId, TransformConfig, TransformContext, TransformOutput},
11    schema,
12    template::Template,
13    transforms::Transform,
14};
15
16/// Configuration of internal metrics for the Throttle transform.
17#[configurable_component]
18#[derive(Clone, Debug, PartialEq, Eq, Default)]
19#[serde(deny_unknown_fields)]
20pub struct ThrottleInternalMetricsConfig {
21    /// Whether or not to emit the `events_discarded_total` internal metric with the `key` tag.
22    ///
23    /// If true, the counter will be incremented for each discarded event, including the key value
24    /// associated with the discarded event. If false, the counter will not be emitted. Instead, the
25    /// number of discarded events can be seen through the `component_discarded_events_total` internal
26    /// metric.
27    ///
28    /// Note that this defaults to false because the `key` tag has potentially unbounded cardinality.
29    /// Only set this to true if you know that the number of unique keys is bounded.
30    #[serde(default)]
31    pub emit_events_discarded_per_key: bool,
32}
33
34/// Configuration for the `throttle` transform.
35#[serde_as]
36#[configurable_component(transform("throttle", "Rate limit logs passing through a topology."))]
37#[derive(Clone, Debug, Default)]
38#[serde(deny_unknown_fields)]
39pub struct ThrottleConfig {
40    /// The number of events allowed for a given bucket per configured `window_secs`.
41    ///
42    /// Each unique key has its own `threshold`.
43    pub threshold: u32,
44
45    /// The time window in which the configured `threshold` is applied, in seconds.
46    #[serde_as(as = "serde_with::DurationSecondsWithFrac<f64>")]
47    #[configurable(metadata(docs::human_name = "Time Window"))]
48    pub window_secs: Duration,
49
50    /// The value to group events into separate buckets to be rate limited independently.
51    ///
52    /// If left unspecified, or if the event doesn't have `key_field`, then the event is not rate
53    /// limited separately.
54    #[configurable(metadata(docs::examples = "{{ message }}", docs::examples = "{{ hostname }}",))]
55    pub key_field: Option<Template>,
56
57    /// A logical condition used to exclude events from sampling.
58    pub exclude: Option<AnyCondition>,
59
60    #[configurable(derived)]
61    #[serde(default)]
62    pub internal_metrics: ThrottleInternalMetricsConfig,
63}
64
65impl_generate_config_from_default!(ThrottleConfig);
66
67#[async_trait::async_trait]
68#[typetag::serde(name = "throttle")]
69impl TransformConfig for ThrottleConfig {
70    async fn build(&self, context: &TransformContext) -> crate::Result<Transform> {
71        Throttle::new(self, context, clock::MonotonicClock).map(Transform::event_task)
72    }
73
74    fn input(&self) -> Input {
75        Input::log()
76    }
77
78    fn outputs(
79        &self,
80        _: vector_lib::enrichment::TableRegistry,
81        input_definitions: &[(OutputId, schema::Definition)],
82        _: LogNamespace,
83    ) -> Vec<TransformOutput> {
84        // The event is not modified, so the definition is passed through as-is
85        vec![TransformOutput::new(
86            DataType::Log,
87            clone_input_definitions(input_definitions),
88        )]
89    }
90}
91
92#[cfg(test)]
93mod tests {
94    use super::ThrottleConfig;
95
96    #[test]
97    fn generate_config() {
98        crate::test_util::test_generate_config::<ThrottleConfig>();
99    }
100}
vector/transforms/throttle/config.rs

vector/transforms/throttle/
config.rs
