pipeline/etl/
processor.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
15pub mod cmcd;
16pub mod csv;
17pub mod date;
18pub mod decolorize;
19pub mod digest;
20pub mod dissect;
21pub mod epoch;
22pub mod filter;
23pub mod gsub;
24pub mod join;
25pub mod json_parse;
26pub mod json_path;
27pub mod letter;
28pub mod logfmt;
29pub mod regex;
30pub mod select;
31pub mod simple_extract;
32pub mod urlencoding;
33pub mod vrl_processor;
34
35use std::str::FromStr;
36
37use cmcd::CmcdProcessor;
38use csv::CsvProcessor;
39use date::DateProcessor;
40use decolorize::DecolorizeProcessor;
41use digest::DigestProcessor;
42use dissect::DissectProcessor;
43use enum_dispatch::enum_dispatch;
44use epoch::EpochProcessor;
45use gsub::GsubProcessor;
46use join::JoinProcessor;
47use json_path::JsonPathProcessor;
48use letter::LetterProcessor;
49use logfmt::LogfmtProcessor;
50use regex::RegexProcessor;
51use snafu::{OptionExt, ResultExt};
52use urlencoding::UrlEncodingProcessor;
53use vrl::value::Value as VrlValue;
54
55use crate::error::{
56    FailedParseFieldFromStringSnafu, FieldMustBeTypeSnafu, InvalidFieldRenameSnafu,
57    ProcessorKeyMustBeStringSnafu, ProcessorMustBeMapSnafu, ProcessorMustHaveStringKeySnafu,
58    Result, UnsupportedProcessorSnafu,
59};
60use crate::etl::field::{Field, Fields};
61use crate::etl::processor::filter::FilterProcessor;
62use crate::etl::processor::json_parse::JsonParseProcessor;
63use crate::etl::processor::select::SelectProcessor;
64use crate::etl::processor::simple_extract::SimpleExtractProcessor;
65use crate::etl::processor::vrl_processor::VrlProcessor;
66
67const FIELD_NAME: &str = "field";
68const FIELDS_NAME: &str = "fields";
69const IGNORE_MISSING_NAME: &str = "ignore_missing";
70const METHOD_NAME: &str = "method";
71const PATTERN_NAME: &str = "pattern";
72const PATTERNS_NAME: &str = "patterns";
73const SEPARATOR_NAME: &str = "separator";
74const TARGET_FIELDS_NAME: &str = "target_fields";
75const JSON_PATH_NAME: &str = "json_path";
76const JSON_PATH_RESULT_INDEX_NAME: &str = "result_index";
77const KEY_NAME: &str = "key";
78const TYPE_NAME: &str = "type";
79const RENAME_TO_KEY: &str = "rename_to";
80
81/// Macro to extract a string value from a YAML map
82#[macro_export]
83macro_rules! yaml_map_get_str {
84    ($map:expr, $key:expr, $value:expr) => {
85        $map.get(&yaml_rust::Yaml::String($key.to_string()))
86            .and_then(|v| v.as_str())
87            .with_context(|| InvalidFieldRenameSnafu {
88                value: $value.clone(),
89            })
90    };
91}
92
93lazy_static::lazy_static! {
94    static ref STRING_FN: fn(&str, &yaml_rust::Yaml) -> Result<String> = |_, v| {
95        Ok(v.as_str().unwrap_or_default().into())
96    };
97
98    static ref STRING_OR_HASH_FN: fn(&str, &yaml_rust::Yaml) -> Result<Field> = |field, v| {
99        match v {
100            yaml_rust::Yaml::String(s) => Field::from_str(s),
101            yaml_rust::Yaml::Hash(m) => {
102                let key = yaml_map_get_str!(m, KEY_NAME, v)?;
103                let rename_to = yaml_map_get_str!(m, RENAME_TO_KEY, v)?;
104                Ok(Field::new(key, Some(rename_to.to_string())))
105            }
106            _ => FieldMustBeTypeSnafu {
107                field,
108                ty: "string or key-rename_to map",
109            }
110            .fail(),
111        }
112    };
113}
114
115/// Processor trait defines the interface for all processors.
116///
117/// A processor is a transformation that can be applied to a field in a document
118/// It can be used to extract, transform, or enrich data
119/// Now Processor only have one input field. In the future, we may support multiple input fields.
120/// The output of a processor is a map of key-value pairs that will be merged into the document when you use exec_map method.
121#[enum_dispatch(ProcessorKind)]
122pub trait Processor: std::fmt::Debug + Send + Sync + 'static {
123    /// Get the processor's kind
124    fn kind(&self) -> &str;
125
126    /// Whether to ignore missing
127    fn ignore_missing(&self) -> bool;
128
129    /// Execute the processor on a vector which be preprocessed by the pipeline
130    fn exec_mut(&self, val: VrlValue) -> Result<VrlValue>;
131}
132
133#[derive(Debug)]
134#[enum_dispatch]
135pub enum ProcessorKind {
136    Cmcd(CmcdProcessor),
137    Csv(CsvProcessor),
138    Dissect(DissectProcessor),
139    Gsub(GsubProcessor),
140    Join(JoinProcessor),
141    Letter(LetterProcessor),
142    Regex(RegexProcessor),
143    UrlEncoding(UrlEncodingProcessor),
144    Epoch(EpochProcessor),
145    Date(DateProcessor),
146    JsonPath(JsonPathProcessor),
147    JsonParse(JsonParseProcessor),
148    Logfmt(LogfmtProcessor),
149    SimpleJsonPath(SimpleExtractProcessor),
150    Decolorize(DecolorizeProcessor),
151    Digest(DigestProcessor),
152    Select(SelectProcessor),
153    Vrl(VrlProcessor),
154    Filter(FilterProcessor),
155}
156
157#[derive(Debug, Default)]
158pub struct Processors {
159    /// A ordered list of processors
160    /// The order of processors is important
161    /// The output of the first processor will be the input of the second processor
162    pub processors: Vec<ProcessorKind>,
163}
164
165impl std::ops::Deref for Processors {
166    type Target = Vec<ProcessorKind>;
167
168    fn deref(&self) -> &Self::Target {
169        &self.processors
170    }
171}
172
173impl std::ops::DerefMut for Processors {
174    fn deref_mut(&mut self) -> &mut Self::Target {
175        &mut self.processors
176    }
177}
178
179impl Processors {
180    /// Parses processors using the server's VRL host access policy.
181    pub fn from_yaml(vec: &[yaml_rust::Yaml], allow_vrl_host_access: bool) -> Result<Self> {
182        let mut processors_builders = vec![];
183        for doc in vec {
184            let processor = parse_processor(doc, allow_vrl_host_access)?;
185            processors_builders.push(processor);
186        }
187        Ok(Processors {
188            processors: processors_builders,
189        })
190    }
191}
192
193fn parse_processor(doc: &yaml_rust::Yaml, allow_vrl_host_access: bool) -> Result<ProcessorKind> {
194    let map = doc.as_hash().context(ProcessorMustBeMapSnafu)?;
195
196    let key = map.keys().next().context(ProcessorMustHaveStringKeySnafu)?;
197
198    let value = map
199        .get(key)
200        .unwrap()
201        .as_hash()
202        .context(ProcessorMustBeMapSnafu)?;
203
204    let str_key = key.as_str().context(ProcessorKeyMustBeStringSnafu)?;
205
206    let processor = match str_key {
207        cmcd::PROCESSOR_CMCD => ProcessorKind::Cmcd(CmcdProcessor::try_from(value)?),
208        csv::PROCESSOR_CSV => ProcessorKind::Csv(CsvProcessor::try_from(value)?),
209        dissect::PROCESSOR_DISSECT => ProcessorKind::Dissect(DissectProcessor::try_from(value)?),
210        epoch::PROCESSOR_EPOCH => ProcessorKind::Epoch(EpochProcessor::try_from(value)?),
211        date::PROCESSOR_DATE => ProcessorKind::Date(DateProcessor::try_from(value)?),
212        gsub::PROCESSOR_GSUB => ProcessorKind::Gsub(GsubProcessor::try_from(value)?),
213        join::PROCESSOR_JOIN => ProcessorKind::Join(JoinProcessor::try_from(value)?),
214        letter::PROCESSOR_LETTER => ProcessorKind::Letter(LetterProcessor::try_from(value)?),
215        regex::PROCESSOR_REGEX => ProcessorKind::Regex(RegexProcessor::try_from(value)?),
216        urlencoding::PROCESSOR_URL_ENCODING => {
217            ProcessorKind::UrlEncoding(UrlEncodingProcessor::try_from(value)?)
218        }
219        json_path::PROCESSOR_JSON_PATH => {
220            ProcessorKind::JsonPath(json_path::JsonPathProcessor::try_from(value)?)
221        }
222        decolorize::PROCESSOR_DECOLORIZE => {
223            ProcessorKind::Decolorize(DecolorizeProcessor::try_from(value)?)
224        }
225        digest::PROCESSOR_DIGEST => ProcessorKind::Digest(DigestProcessor::try_from(value)?),
226        simple_extract::PROCESSOR_SIMPLE_EXTRACT => {
227            ProcessorKind::SimpleJsonPath(SimpleExtractProcessor::try_from(value)?)
228        }
229        json_parse::PROCESSOR_JSON_PARSE => {
230            ProcessorKind::JsonParse(JsonParseProcessor::try_from(value)?)
231        }
232        logfmt::PROCESSOR_LOGFMT => ProcessorKind::Logfmt(LogfmtProcessor::try_from(value)?),
233        vrl_processor::PROCESSOR_VRL => {
234            ProcessorKind::Vrl(VrlProcessor::from_yaml(value, allow_vrl_host_access)?)
235        }
236        select::PROCESSOR_SELECT => ProcessorKind::Select(SelectProcessor::try_from(value)?),
237        filter::PROCESSOR_FILTER => ProcessorKind::Filter(FilterProcessor::try_from(value)?),
238        _ => return UnsupportedProcessorSnafu { processor: str_key }.fail(),
239    };
240
241    Ok(processor)
242}
243
244pub(crate) fn yaml_string(v: &yaml_rust::Yaml, field: &str) -> Result<String> {
245    v.as_str()
246        .map(|s| s.to_string())
247        .context(FieldMustBeTypeSnafu {
248            field,
249            ty: "string",
250        })
251}
252
253pub(crate) fn yaml_strings(v: &yaml_rust::Yaml, field: &str) -> Result<Vec<String>> {
254    yaml_list(v, *STRING_FN, field)
255}
256
257pub(crate) fn yaml_list<T>(
258    v: &yaml_rust::Yaml,
259    conv_fn: impl Fn(&str, &yaml_rust::Yaml) -> Result<T>,
260    field: &str,
261) -> Result<Vec<T>> {
262    v.as_vec()
263        .context(FieldMustBeTypeSnafu { field, ty: "list" })?
264        .iter()
265        .map(|v| conv_fn(field, v))
266        .collect()
267}
268
269pub(crate) fn yaml_bool(v: &yaml_rust::Yaml, field: &str) -> Result<bool> {
270    v.as_bool().context(FieldMustBeTypeSnafu {
271        field,
272        ty: "boolean",
273    })
274}
275
276pub(crate) fn yaml_parse_string<T>(v: &yaml_rust::Yaml, field: &str) -> Result<T>
277where
278    T: std::str::FromStr,
279    T::Err: std::error::Error + Send + Sync + 'static,
280{
281    yaml_string(v, field)?
282        .parse::<T>()
283        .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
284        .context(FailedParseFieldFromStringSnafu { field })
285}
286
287pub(crate) fn yaml_parse_strings<T>(v: &yaml_rust::Yaml, field: &str) -> Result<Vec<T>>
288where
289    T: std::str::FromStr,
290    T::Err: std::error::Error + Send + Sync + 'static,
291{
292    yaml_strings(v, field).and_then(|v| {
293        v.into_iter()
294            .map(|s| {
295                s.parse::<T>()
296                    .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
297                    .context(FailedParseFieldFromStringSnafu { field })
298            })
299            .collect()
300    })
301}
302
303pub(crate) fn yaml_new_fields(v: &yaml_rust::Yaml, field: &str) -> Result<Fields> {
304    yaml_list(v, *STRING_OR_HASH_FN, field).map(Fields::new)
305}
306
307pub(crate) fn yaml_new_field(v: &yaml_rust::Yaml, field: &str) -> Result<Field> {
308    STRING_OR_HASH_FN(field, v)
309}
pipeline/etl/processor.rs

pipeline/etl/
processor.rs
