vector/transforms/aggregate/
config.rs

1use std::collections::HashMap;
2
3use vector_lib::configurable::configurable_component;
4
5use super::Aggregate;
6use crate::{
7    config::{DataType, Input, OutputId, TransformConfig, TransformContext, TransformOutput},
8    schema,
9    transforms::Transform,
10};
11
12/// Configuration for the `aggregate` transform.
13#[configurable_component(transform("aggregate", "Aggregate metrics passing through a topology."))]
14#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
15#[serde(deny_unknown_fields)]
16pub struct AggregateConfig {
17    /// The interval between flushes, in milliseconds.
18    ///
19    /// During this time frame, metrics (beta) with the same series data (name, namespace, tags, and so on) are aggregated.
20    #[serde(default = "default_interval_ms")]
21    #[configurable(metadata(docs::human_name = "Flush Interval"))]
22    pub interval_ms: u64,
23    /// Function to use for aggregation.
24    ///
25    /// Some of the functions may only function on incremental and some only on absolute metrics.
26    #[serde(default = "default_mode")]
27    #[configurable(derived)]
28    pub mode: AggregationMode,
29}
30
31#[configurable_component]
32#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
33#[configurable(description = "The aggregation mode to use.")]
34pub enum AggregationMode {
35    /// Default mode. Sums incremental metrics and uses the latest value for absolute metrics.
36    #[default]
37    Auto,
38
39    /// Sums incremental metrics; absolute metrics pass through unchanged.
40    Sum,
41
42    /// Returns the latest value for absolute metrics; incremental metrics pass through unchanged.
43    Latest,
44
45    /// Counts metrics for incremental and absolute metrics
46    Count,
47
48    /// Returns difference between latest value for absolute; incremental metrics pass through unchanged.
49    Diff,
50
51    /// Max value of absolute metric; incremental metrics pass through unchanged.
52    Max,
53
54    /// Min value of absolute metric; incremental metrics pass through unchanged.
55    Min,
56
57    /// Mean value of absolute metric; incremental metrics pass through unchanged.
58    Mean,
59
60    /// Stdev value of absolute metric; incremental metrics pass through unchanged.
61    Stdev,
62}
63const fn default_mode() -> AggregationMode {
64    AggregationMode::Auto
65}
66
67const fn default_interval_ms() -> u64 {
68    10 * 1000
69}
70
71impl_generate_config_from_default!(AggregateConfig);
72
73#[async_trait::async_trait]
74#[typetag::serde(name = "aggregate")]
75impl TransformConfig for AggregateConfig {
76    async fn build(&self, _context: &TransformContext) -> crate::Result<Transform> {
77        Aggregate::new(self).map(Transform::event_task)
78    }
79
80    fn input(&self) -> Input {
81        Input::metric()
82    }
83
84    fn outputs(
85        &self,
86        _: &TransformContext,
87        _: &[(OutputId, schema::Definition)],
88    ) -> Vec<TransformOutput> {
89        vec![TransformOutput::new(DataType::Metric, HashMap::new())]
90    }
91}
vector/transforms/aggregate/config.rs

vector/transforms/aggregate/
config.rs
