pipeline/
dispatcher.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use common_telemetry::debug;
16use snafu::OptionExt;
17use vrl::value::Value as VrlValue;
18use yaml_rust::Yaml;
19
20use crate::error::{
21    Error, FieldRequiredForDispatcherSnafu, Result, TableSuffixRequiredForDispatcherRuleSnafu,
22    ValueRequiredForDispatcherRuleSnafu,
23};
24use crate::etl::ctx_req::TABLE_SUFFIX_KEY;
25use crate::etl::value::yaml_to_vrl_value;
26
27const FIELD: &str = "field";
28const PIPELINE: &str = "pipeline";
29const VALUE: &str = "value";
30const RULES: &str = "rules";
31
32/// The dispatcher configuration.
33///
34/// Dispatcher in a pipeline allows user to call another pipeline and specify
35/// table name based on field matching.
36///
37/// ```yaml
38/// dispatcher:
39///   field: type
40///   rules:
41///     - value: http
42///       pipeline: http_pipeline
43///       table_suffix: http_log
44///     - value: db
45///       pipeline: db_pipeline
46///       table_suffix: db_log
47/// ```
48///
49/// If none of the rules match the value, this pipeline will continue to process
50/// current log entry
51#[derive(Debug, PartialEq)]
52pub(crate) struct Dispatcher {
53    pub field: String,
54    pub rules: Vec<Rule>,
55}
56
57/// The rule definition for dispatcher
58///
59/// - `value`: for pattern matching
60/// - `pipeline`: the pipeline to call, if it's unspecified, we use default
61///   `greptime_identity`
62/// - `table_suffix`: the table name segment that we use to construct full table
63///   name
64#[derive(Debug, PartialEq)]
65pub(crate) struct Rule {
66    pub value: VrlValue,
67    pub table_suffix: String,
68    pub pipeline: Option<String>,
69}
70
71impl TryFrom<&Yaml> for Dispatcher {
72    type Error = Error;
73
74    fn try_from(value: &Yaml) -> Result<Self> {
75        let field = value[FIELD]
76            .as_str()
77            .map(|s| s.to_string())
78            .context(FieldRequiredForDispatcherSnafu)?;
79
80        let rules = if let Some(rules) = value[RULES].as_vec() {
81            rules
82                .iter()
83                .map(|rule| {
84                    let table_part = rule[TABLE_SUFFIX_KEY]
85                        .as_str()
86                        .map(|s| s.to_string())
87                        .context(TableSuffixRequiredForDispatcherRuleSnafu)?;
88
89                    let pipeline = rule[PIPELINE].as_str().map(|s| s.to_string());
90
91                    if rule[VALUE].is_badvalue() {
92                        ValueRequiredForDispatcherRuleSnafu.fail()?;
93                    }
94
95                    let value = yaml_to_vrl_value(&rule[VALUE])?;
96
97                    Ok(Rule {
98                        value,
99                        table_suffix: table_part,
100                        pipeline,
101                    })
102                })
103                .collect::<Result<Vec<Rule>>>()?
104        } else {
105            vec![]
106        };
107
108        Ok(Dispatcher { field, rules })
109    }
110}
111
112impl Dispatcher {
113    /// execute dispatcher and returns matched rule if any
114    pub(crate) fn exec(&self, data: &VrlValue) -> Option<&Rule> {
115        let data = data.as_object()?;
116        if let Some(value) = data.get(self.field.as_str()) {
117            for rule in &self.rules {
118                if rule.value == *value {
119                    return Some(rule);
120                }
121            }
122
123            None
124        } else {
125            debug!("field {} not found in keys {:?}", &self.field, data);
126            None
127        }
128    }
129}
pipeline/dispatcher.rs

pipeline/
dispatcher.rs
