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 yaml_rust::Yaml;
18
19use crate::error::{
20    Error, FieldRequiredForDispatcherSnafu, Result, TableSuffixRequiredForDispatcherRuleSnafu,
21    ValueRequiredForDispatcherRuleSnafu,
22};
23use crate::{PipelineMap, Value};
24
25const FIELD: &str = "field";
26const TABLE_SUFFIX: &str = "table_suffix";
27const PIPELINE: &str = "pipeline";
28const VALUE: &str = "value";
29const RULES: &str = "rules";
30
31/// The dispatcher configuration.
32///
33/// Dispatcher in a pipeline allows user to call another pipeline and specify
34/// table name based on field matching.
35///
36/// ```yaml
37/// dispatcher:
38///   field: type
39///   rules:
40///     - value: http
41///       pipeline: http_pipeline
42///       table_suffix: http_log
43///     - value: db
44///       pipeline: db_pipeline
45///       table_suffix: db_log
46/// ```
47///
48/// If none of the rules match the value, this pipeline will continue to process
49/// current log entry
50#[derive(Debug, PartialEq)]
51pub(crate) struct Dispatcher {
52    pub field: String,
53    pub rules: Vec<Rule>,
54}
55
56/// The rule definition for dispatcher
57///
58/// - `value`: for pattern matching
59/// - `pipeline`: the pipeline to call, if it's unspecified, we use default
60///   `greptime_identity`
61/// - `table_suffix`: the table name segment that we use to construct full table
62///   name
63#[derive(Debug, PartialEq)]
64pub(crate) struct Rule {
65    pub value: Value,
66    pub table_suffix: String,
67    pub pipeline: Option<String>,
68}
69
70impl TryFrom<&Yaml> for Dispatcher {
71    type Error = Error;
72
73    fn try_from(value: &Yaml) -> Result<Self> {
74        let field = value[FIELD]
75            .as_str()
76            .map(|s| s.to_string())
77            .context(FieldRequiredForDispatcherSnafu)?;
78
79        let rules = if let Some(rules) = value[RULES].as_vec() {
80            rules
81                .iter()
82                .map(|rule| {
83                    let table_part = rule[TABLE_SUFFIX]
84                        .as_str()
85                        .map(|s| s.to_string())
86                        .context(TableSuffixRequiredForDispatcherRuleSnafu)?;
87
88                    let pipeline = rule[PIPELINE].as_str().map(|s| s.to_string());
89
90                    if rule[VALUE].is_badvalue() {
91                        ValueRequiredForDispatcherRuleSnafu.fail()?;
92                    }
93                    let value = Value::try_from(&rule[VALUE])?;
94
95                    Ok(Rule {
96                        value,
97                        table_suffix: table_part,
98                        pipeline,
99                    })
100                })
101                .collect::<Result<Vec<Rule>>>()?
102        } else {
103            vec![]
104        };
105
106        Ok(Dispatcher { field, rules })
107    }
108}
109
110impl Dispatcher {
111    /// execute dispatcher and returns matched rule if any
112    pub(crate) fn exec(&self, data: &PipelineMap) -> Option<&Rule> {
113        if let Some(value) = data.get(&self.field) {
114            for rule in &self.rules {
115                if rule.value == *value {
116                    return Some(rule);
117                }
118            }
119
120            None
121        } else {
122            debug!("field {} not found in keys {:?}", &self.field, data.keys());
123            None
124        }
125    }
126}
pipeline/dispatcher.rs

pipeline/
dispatcher.rs
