vector/transforms/throttle/
config.rs

1use std::time::Duration;
2
3use governor::clock;
4use serde_with::serde_as;
5use vector_lib::{config::clone_input_definitions, 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        _: &TransformContext,
81        input_definitions: &[(OutputId, schema::Definition)],
82    ) -> Vec<TransformOutput> {
83        // The event is not modified, so the definition is passed through as-is
84        vec![TransformOutput::new(
85            DataType::Log,
86            clone_input_definitions(input_definitions),
87        )]
88    }
89
90    fn validate(&self, _: &TransformContext) -> Result<(), Vec<String>> {
91        if self.threshold == 0 || self.window_secs.as_secs_f64() == 0.0 {
92            Err(vec![
93                "`threshold` and `window_secs` must be non-zero".to_string(),
94            ])
95        } else {
96            Ok(())
97        }
98    }
99
100    fn validate_env(&self, context: &TransformContext) -> Result<(), Vec<String>> {
101        if let Some(Err(e)) = self
102            .exclude
103            .as_ref()
104            .map(|c| c.validate(&context.enrichment_tables, &context.metrics_storage))
105        {
106            Err(vec![format!("exclude: {e}")])
107        } else {
108            Ok(())
109        }
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::ThrottleConfig;
116
117    #[test]
118    fn generate_config() {
119        crate::test_util::test_generate_config::<ThrottleConfig>();
120    }
121}
vector/transforms/throttle/config.rs

vector/transforms/throttle/
config.rs
