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

pipeline/etl/
processor.rs
