vector/transforms/
route.rs

1use indexmap::IndexMap;
2use vector_lib::{
3    config::clone_input_definitions, configurable::configurable_component, transform::SyncTransform,
4};
5
6use crate::{
7    conditions::{AnyCondition, Condition, ConditionConfig, VrlConfig},
8    config::{
9        DataType, GenerateConfig, Input, OutputId, TransformConfig, TransformContext,
10        TransformOutput,
11    },
12    event::Event,
13    schema,
14    transforms::Transform,
15};
16
17pub(crate) const UNMATCHED_ROUTE: &str = "_unmatched";
18
19#[derive(Clone)]
20pub struct Route {
21    conditions: Vec<(String, Condition)>,
22    reroute_unmatched: bool,
23}
24
25impl Route {
26    pub fn new(config: &RouteConfig, context: &TransformContext) -> crate::Result<Self> {
27        let mut conditions = Vec::with_capacity(config.route.len());
28        for (output_name, condition) in config.route.iter() {
29            let condition =
30                condition.build(&context.enrichment_tables, &context.metrics_storage)?;
31            conditions.push((output_name.clone(), condition));
32        }
33        Ok(Self {
34            conditions,
35            reroute_unmatched: config.reroute_unmatched,
36        })
37    }
38}
39
40impl SyncTransform for Route {
41    fn transform(&mut self, event: Event, output: &mut vector_lib::transform::TransformOutputsBuf) {
42        let mut check_failed: usize = 0;
43        for (output_name, condition) in &self.conditions {
44            let (result, event) = condition.check(event.clone());
45            if result {
46                output.push(Some(output_name), event);
47            } else {
48                check_failed += 1;
49            }
50        }
51        if self.reroute_unmatched && check_failed == self.conditions.len() {
52            output.push(Some(UNMATCHED_ROUTE), event);
53        }
54    }
55}
56
57/// Configuration for the `route` transform.
58#[configurable_component(transform(
59    "route",
60    "Split a stream of events into multiple sub-streams based on user-supplied conditions."
61))]
62#[derive(Clone, Debug)]
63#[serde(deny_unknown_fields)]
64pub struct RouteConfig {
65    /// Reroutes unmatched events to a named output instead of silently discarding them.
66    ///
67    /// Normally, if an event doesn't match any defined route, it is sent to the `<transform_name>._unmatched`
68    /// output for further processing. In some cases, you may want to simply discard unmatched events and not
69    /// process them any further.
70    ///
71    /// In these cases, `reroute_unmatched` can be set to `false` to disable the `<transform_name>._unmatched`
72    /// output and instead silently discard any unmatched events.
73    #[serde(default = "crate::serde::default_true")]
74    #[configurable(metadata(docs::human_name = "Reroute Unmatched Events"))]
75    reroute_unmatched: bool,
76
77    /// A map from route identifiers to logical conditions.
78    /// Each condition represents a filter which is applied to each event.
79    ///
80    /// The following identifiers are reserved output names and thus cannot be used as route IDs:
81    /// - `_unmatched`
82    /// - `_default`
83    ///
84    /// Each route can then be referenced as an input by other components with the name
85    /// `<transform_name>.<route_id>`. If an event doesn’t match any route, and if `reroute_unmatched`
86    /// is set to `true` (the default), it is sent to the `<transform_name>._unmatched` output.
87    /// Otherwise, the unmatched event is instead silently discarded.
88    #[configurable(metadata(docs::additional_props_description = "An individual route."))]
89    #[configurable(metadata(docs::examples = "route_examples()"))]
90    route: IndexMap<String, AnyCondition>,
91}
92
93fn route_examples() -> IndexMap<String, AnyCondition> {
94    IndexMap::from([
95        (
96            "foo-exists".to_owned(),
97            AnyCondition::Map(ConditionConfig::Vrl(VrlConfig {
98                source: "exists(.foo)".to_owned(),
99                ..Default::default()
100            })),
101        ),
102        (
103            "foo-does-not-exist".to_owned(),
104            AnyCondition::Map(ConditionConfig::Vrl(VrlConfig {
105                source: "!exists(.foo)".to_owned(),
106                ..Default::default()
107            })),
108        ),
109    ])
110}
111
112impl GenerateConfig for RouteConfig {
113    fn generate_config() -> toml::Value {
114        toml::Value::try_from(Self {
115            reroute_unmatched: true,
116            route: route_examples(),
117        })
118        .unwrap()
119    }
120}
121
122#[async_trait::async_trait]
123#[typetag::serde(name = "route")]
124impl TransformConfig for RouteConfig {
125    async fn build(&self, context: &TransformContext) -> crate::Result<Transform> {
126        let route = Route::new(self, context)?;
127        Ok(Transform::synchronous(route))
128    }
129
130    fn input(&self) -> Input {
131        Input::all()
132    }
133
134    fn validate(&self, _: &TransformContext) -> Result<(), Vec<String>> {
135        if self.route.contains_key(UNMATCHED_ROUTE) {
136            Err(vec![format!(
137                "cannot have a named output with reserved name: `{UNMATCHED_ROUTE}`"
138            )])
139        } else {
140            Ok(())
141        }
142    }
143
144    fn validate_env(&self, context: &TransformContext) -> Result<(), Vec<String>> {
145        let errors: Vec<String> = self
146            .route
147            .iter()
148            .filter_map(|(name, condition)| {
149                condition
150                    .validate(&context.enrichment_tables, &context.metrics_storage)
151                    .err()
152                    .map(|e| format!("route \"{name}\": {e}"))
153            })
154            .collect();
155
156        if errors.is_empty() {
157            Ok(())
158        } else {
159            Err(errors)
160        }
161    }
162
163    fn outputs(
164        &self,
165        _: &TransformContext,
166        input_definitions: &[(OutputId, schema::Definition)],
167    ) -> Vec<TransformOutput> {
168        let mut result: Vec<TransformOutput> = self
169            .route
170            .keys()
171            .map(|output_name| {
172                TransformOutput::new(
173                    DataType::all_bits(),
174                    clone_input_definitions(input_definitions),
175                )
176                .with_port(output_name)
177            })
178            .collect();
179        if self.reroute_unmatched {
180            result.push(
181                TransformOutput::new(
182                    DataType::all_bits(),
183                    clone_input_definitions(input_definitions),
184                )
185                .with_port(UNMATCHED_ROUTE),
186            );
187        }
188        result
189    }
190
191    fn enable_concurrency(&self) -> bool {
192        true
193    }
194}
195
196#[cfg(test)]
197mod test {
198    use std::collections::HashMap;
199
200    use indoc::indoc;
201    use vector_lib::{config::LogNamespace, transform::TransformOutputsBuf};
202
203    use super::*;
204    use crate::{
205        config::{ConfigBuilder, build_unit_tests},
206        test_util::components::{COMPONENT_MULTIPLE_OUTPUTS_TESTS, init_test},
207    };
208
209    #[test]
210    fn generate_config() {
211        crate::test_util::test_generate_config::<super::RouteConfig>();
212    }
213
214    #[test]
215    fn can_serialize_remap() {
216        // We need to serialize the config to check if a config has
217        // changed when reloading.
218        let config = serde_yaml::from_str::<RouteConfig>(indoc! {"
219            route:
220              first:
221                type: vrl
222                source: '.message == \"hello world\"'
223        "})
224        .unwrap();
225
226        assert_eq!(
227            serde_json::to_string(&config).unwrap(),
228            r#"{"reroute_unmatched":true,"route":{"first":{"type":"vrl","source":".message == \"hello world\""}}}"#
229        );
230    }
231
232    #[test]
233    fn route_pass_all_route_conditions() {
234        let output_names = vec!["first", "second", "third", UNMATCHED_ROUTE];
235        let event = Event::from_json_value(
236            serde_json::json!({"message": "hello world", "second": "second", "third": "third"}),
237            LogNamespace::Legacy,
238        )
239        .unwrap();
240        let config = serde_yaml::from_str::<RouteConfig>(indoc! {"
241            route:
242              first:
243                type: vrl
244                source: '.message == \"hello world\"'
245              second:
246                type: vrl
247                source: '.second == \"second\"'
248              third:
249                type: vrl
250                source: '.third == \"third\"'
251        "})
252        .unwrap();
253
254        let mut transform = Route::new(&config, &Default::default()).unwrap();
255        let mut outputs = TransformOutputsBuf::new_with_capacity(
256            output_names
257                .iter()
258                .map(|output_name| {
259                    TransformOutput::new(DataType::all_bits(), HashMap::new())
260                        .with_port(output_name.to_owned())
261                })
262                .collect(),
263            1,
264        );
265
266        transform.transform(event.clone(), &mut outputs);
267        for output_name in output_names {
268            let mut events: Vec<_> = outputs.drain_named(output_name).collect();
269            if output_name == UNMATCHED_ROUTE {
270                assert!(events.is_empty());
271            } else {
272                assert_eq!(events.len(), 1);
273                assert_eq!(events.pop().unwrap(), event);
274            }
275        }
276    }
277
278    #[test]
279    fn route_pass_one_route_condition() {
280        let output_names = vec!["first", "second", "third", UNMATCHED_ROUTE];
281        let event = Event::from_json_value(
282            serde_json::json!({"message": "hello world"}),
283            LogNamespace::Legacy,
284        )
285        .unwrap();
286        let config = serde_yaml::from_str::<RouteConfig>(indoc! {"
287            route:
288              first:
289                type: vrl
290                source: '.message == \"hello world\"'
291              second:
292                type: vrl
293                source: '.second == \"second\"'
294              third:
295                type: vrl
296                source: '.third == \"third\"'
297        "})
298        .unwrap();
299
300        let mut transform = Route::new(&config, &Default::default()).unwrap();
301        let mut outputs = TransformOutputsBuf::new_with_capacity(
302            output_names
303                .iter()
304                .map(|output_name| {
305                    TransformOutput::new(DataType::all_bits(), HashMap::new())
306                        .with_port(output_name.to_owned())
307                })
308                .collect(),
309            1,
310        );
311
312        transform.transform(event.clone(), &mut outputs);
313        for output_name in output_names {
314            let mut events: Vec<_> = outputs.drain_named(output_name).collect();
315            if output_name == "first" {
316                assert_eq!(events.len(), 1);
317                assert_eq!(events.pop().unwrap(), event);
318            }
319            assert_eq!(events.len(), 0);
320        }
321    }
322
323    #[test]
324    fn route_pass_no_route_condition() {
325        let output_names = vec!["first", "second", "third", UNMATCHED_ROUTE];
326        let event =
327            Event::from_json_value(serde_json::json!({"message": "NOPE"}), LogNamespace::Legacy)
328                .unwrap();
329        let config = serde_yaml::from_str::<RouteConfig>(indoc! {"
330            route:
331              first:
332                type: vrl
333                source: '.message == \"hello world\"'
334              second:
335                type: vrl
336                source: '.second == \"second\"'
337              third:
338                type: vrl
339                source: '.third == \"third\"'
340        "})
341        .unwrap();
342
343        let mut transform = Route::new(&config, &Default::default()).unwrap();
344        let mut outputs = TransformOutputsBuf::new_with_capacity(
345            output_names
346                .iter()
347                .map(|output_name| {
348                    TransformOutput::new(DataType::all_bits(), HashMap::new())
349                        .with_port(output_name.to_owned())
350                })
351                .collect(),
352            1,
353        );
354
355        transform.transform(event.clone(), &mut outputs);
356        for output_name in output_names {
357            let mut events: Vec<_> = outputs.drain_named(output_name).collect();
358            if output_name == UNMATCHED_ROUTE {
359                assert_eq!(events.len(), 1);
360                assert_eq!(events.pop().unwrap(), event);
361            }
362            assert_eq!(events.len(), 0);
363        }
364    }
365
366    #[test]
367    fn route_no_unmatched_output() {
368        let output_names = vec!["first", "second", "third", UNMATCHED_ROUTE];
369        let event =
370            Event::from_json_value(serde_json::json!({"message": "NOPE"}), LogNamespace::Legacy)
371                .unwrap();
372        let config = serde_yaml::from_str::<RouteConfig>(indoc! {"
373            reroute_unmatched: false
374            route:
375              first:
376                type: vrl
377                source: '.message == \"hello world\"'
378              second:
379                type: vrl
380                source: '.second == \"second\"'
381              third:
382                type: vrl
383                source: '.third == \"third\"'
384        "})
385        .unwrap();
386
387        let mut transform = Route::new(&config, &Default::default()).unwrap();
388        let mut outputs = TransformOutputsBuf::new_with_capacity(
389            output_names
390                .iter()
391                .map(|output_name| {
392                    TransformOutput::new(DataType::all_bits(), HashMap::new())
393                        .with_port(output_name.to_owned())
394                })
395                .collect(),
396            1,
397        );
398
399        transform.transform(event.clone(), &mut outputs);
400        for output_name in output_names {
401            let events: Vec<_> = outputs.drain_named(output_name).collect();
402            assert_eq!(events.len(), 0);
403        }
404    }
405
406    #[tokio::test]
407    async fn route_metrics_with_output_tag() {
408        init_test();
409
410        let config: ConfigBuilder = serde_yaml::from_str(indoc! {"
411            transforms:
412              foo:
413                inputs: []
414                type: route
415                route:
416                  first:
417                    type: is_log
418            tests:
419              - name: metric output
420                input:
421                  insert_at: foo
422                  value: none
423                outputs:
424                  - extract_from: foo.first
425                    conditions:
426                      - type: vrl
427                        source: \"true\"
428        "})
429        .unwrap();
430
431        let mut tests = build_unit_tests(config).await.unwrap();
432        assert!(tests.remove(0).run().await.errors.is_empty());
433        // Check that metrics were emitted with output tag
434        COMPONENT_MULTIPLE_OUTPUTS_TESTS.assert(&["output"]);
435    }
436}
vector/transforms/route.rs

vector/transforms/
route.rs
