vector/transforms/
filter.rs

1use vector_lib::{
2    config::clone_input_definitions,
3    configurable::configurable_component,
4    internal_event::{Count, InternalEventHandle as _, Registered},
5};
6
7use crate::{
8    conditions::{AnyCondition, Condition},
9    config::{
10        DataType, GenerateConfig, Input, OutputId, TransformConfig, TransformContext,
11        TransformOutput,
12    },
13    event::Event,
14    internal_events::FilterEventsDropped,
15    schema,
16    transforms::{FunctionTransform, OutputBuffer, Transform},
17};
18
19/// Configuration for the `filter` transform.
20#[configurable_component(transform("filter", "Filter events based on a set of conditions."))]
21#[derive(Clone, Debug)]
22#[serde(deny_unknown_fields)]
23pub struct FilterConfig {
24    #[configurable(derived)]
25    /// The condition that every input event is matched against.
26    ///
27    /// If an event is matched by the condition, it is forwarded. Otherwise, the event is dropped.
28    condition: AnyCondition,
29}
30
31impl From<AnyCondition> for FilterConfig {
32    fn from(condition: AnyCondition) -> Self {
33        Self { condition }
34    }
35}
36
37impl GenerateConfig for FilterConfig {
38    fn generate_config() -> toml::Value {
39        toml::from_str(r#"condition = ".message == \"value\"""#).unwrap()
40    }
41}
42
43#[async_trait::async_trait]
44#[typetag::serde(name = "filter")]
45impl TransformConfig for FilterConfig {
46    async fn build(&self, context: &TransformContext) -> crate::Result<Transform> {
47        Ok(Transform::function(Filter::new(self.condition.build(
48            &context.enrichment_tables,
49            &context.metrics_storage,
50        )?)))
51    }
52
53    fn validate_env(&self, context: &TransformContext) -> Result<(), Vec<String>> {
54        self.condition
55            .validate(&context.enrichment_tables, &context.metrics_storage)
56            .map_err(|e| vec![e.to_string()])
57    }
58
59    fn input(&self) -> Input {
60        Input::all()
61    }
62
63    fn outputs(
64        &self,
65        _: &TransformContext,
66        input_definitions: &[(OutputId, schema::Definition)],
67    ) -> Vec<TransformOutput> {
68        vec![TransformOutput::new(
69            DataType::all_bits(),
70            clone_input_definitions(input_definitions),
71        )]
72    }
73
74    fn enable_concurrency(&self) -> bool {
75        true
76    }
77}
78
79#[derive(Clone)]
80pub struct Filter {
81    condition: Condition,
82    events_dropped: Registered<FilterEventsDropped>,
83}
84
85impl Filter {
86    pub fn new(condition: Condition) -> Self {
87        Self {
88            condition,
89            events_dropped: register!(FilterEventsDropped),
90        }
91    }
92}
93
94impl FunctionTransform for Filter {
95    fn transform(&mut self, output: &mut OutputBuffer, event: Event) {
96        let (result, event) = self.condition.check(event);
97        if result {
98            output.push(event);
99        } else {
100            self.events_dropped.emit(Count(1));
101        }
102    }
103}
104
105#[cfg(test)]
106mod test {
107    use std::sync::Arc;
108
109    use tokio::sync::mpsc;
110    use tokio_stream::wrappers::ReceiverStream;
111    use vector_lib::{
112        config::ComponentKey,
113        event::{Metric, MetricKind, MetricValue},
114    };
115
116    use super::*;
117    use crate::{
118        conditions::ConditionConfig,
119        config::schema::Definition,
120        event::{Event, LogEvent},
121        test_util::components::assert_transform_compliance,
122        transforms::test::create_topology,
123    };
124
125    const TEST_SOURCE_COMPONENT_ID: &str = "in";
126    const TEST_UPSTREAM_COMPONENT_ID: &str = "transform";
127    const TEST_SOURCE_TYPE: &str = "unit_test_stream";
128
129    fn set_expected_metadata(event: &mut Event) {
130        event.set_source_id(Arc::new(ComponentKey::from(TEST_SOURCE_COMPONENT_ID)));
131        event.set_upstream_id(Arc::new(OutputId::from(TEST_UPSTREAM_COMPONENT_ID)));
132        event.set_source_type(TEST_SOURCE_TYPE);
133        event
134            .metadata_mut()
135            .set_schema_definition(&Arc::new(Definition::default_legacy_namespace()));
136    }
137
138    #[test]
139    fn generate_config() {
140        crate::test_util::test_generate_config::<super::FilterConfig>();
141    }
142
143    #[tokio::test]
144    async fn filter_basic() {
145        assert_transform_compliance(async {
146            let transform_config = FilterConfig::from(AnyCondition::from(ConditionConfig::IsLog));
147
148            let (tx, rx) = mpsc::channel(1);
149            let (topology, mut out) =
150                create_topology(ReceiverStream::new(rx), transform_config).await;
151
152            let mut log = Event::from(LogEvent::from("message"));
153            tx.send(log.clone()).await.unwrap();
154
155            set_expected_metadata(&mut log);
156
157            assert_eq!(out.recv().await.unwrap(), log);
158
159            let metric = Event::from(Metric::new(
160                "test metric",
161                MetricKind::Incremental,
162                MetricValue::Counter { value: 1.0 },
163            ));
164            tx.send(metric).await.unwrap();
165
166            drop(tx);
167            topology.stop().await;
168            assert_eq!(out.recv().await, None);
169        })
170        .await;
171    }
172}
vector/transforms/filter.rs

vector/transforms/
filter.rs
