pipeline/etl/processor/
logfmt.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
15//! The `logfmt` processor parses a [logfmt](https://brandur.org/logfmt) line,
16//! such as `level=info msg="hello world" duration=12ms`, into an object.
17
18use std::collections::BTreeMap;
19
20use snafu::OptionExt as _;
21use vrl::prelude::Bytes;
22use vrl::value::{KeyString, Value as VrlValue};
23
24use crate::Processor;
25use crate::error::{
26    Error, FieldMustBeTypeSnafu, KeyMustBeStringSnafu, LogfmtParseSnafu,
27    ProcessorMissingFieldSnafu, Result, ValueMustBeMapSnafu,
28};
29use crate::etl::field::Fields;
30use crate::etl::processor::{
31    FIELD_NAME, FIELDS_NAME, IGNORE_MISSING_NAME, yaml_bool, yaml_new_field, yaml_new_fields,
32};
33
34pub(crate) const PROCESSOR_LOGFMT: &str = "logfmt";
35
36/// Parses logfmt strings into objects.
37///
38/// Every `key=value` pair becomes an entry of the output object with a string
39/// value. A bare key without `=` (e.g. `debug` in `debug msg=hi`) is treated
40/// as a flag and set to `true`. When a key appears more than once, the last
41/// value wins. The output is written to the target field, or replaces the
42/// input field in place when no target is given.
43#[derive(Debug, Default)]
44pub struct LogfmtProcessor {
45    fields: Fields,
46    ignore_missing: bool,
47}
48
49impl TryFrom<&yaml_rust::yaml::Hash> for LogfmtProcessor {
50    type Error = Error;
51
52    fn try_from(value: &yaml_rust::yaml::Hash) -> std::result::Result<Self, Self::Error> {
53        let mut fields = Fields::default();
54        let mut ignore_missing = false;
55
56        for (k, v) in value.iter() {
57            let key = k
58                .as_str()
59                .with_context(|| KeyMustBeStringSnafu { k: k.clone() })?;
60            match key {
61                FIELD_NAME => {
62                    fields = Fields::one(yaml_new_field(v, FIELD_NAME)?);
63                }
64                FIELDS_NAME => {
65                    fields = yaml_new_fields(v, FIELDS_NAME)?;
66                }
67                IGNORE_MISSING_NAME => {
68                    ignore_missing = yaml_bool(v, IGNORE_MISSING_NAME)?;
69                }
70                _ => {}
71            }
72        }
73
74        Ok(LogfmtProcessor {
75            fields,
76            ignore_missing,
77        })
78    }
79}
80
81impl LogfmtProcessor {
82    fn process_field(&self, val: &VrlValue) -> Result<VrlValue> {
83        let Some(line) = val.as_str() else {
84            return FieldMustBeTypeSnafu {
85                field: val.to_string(),
86                ty: "string",
87            }
88            .fail();
89        };
90        let pairs = parse_logfmt(&line).map_err(|reason| {
91            LogfmtParseSnafu {
92                input: line.to_string(),
93                reason,
94            }
95            .build()
96        })?;
97        Ok(VrlValue::Object(pairs))
98    }
99}
100
101impl Processor for LogfmtProcessor {
102    fn kind(&self) -> &str {
103        PROCESSOR_LOGFMT
104    }
105
106    fn ignore_missing(&self) -> bool {
107        self.ignore_missing
108    }
109
110    fn exec_mut(&self, mut val: VrlValue) -> Result<VrlValue> {
111        for field in self.fields.iter() {
112            let index = field.input_field();
113            let val = val.as_object_mut().context(ValueMustBeMapSnafu)?;
114            match val.get(index) {
115                Some(VrlValue::Null) | None => {
116                    if !self.ignore_missing {
117                        return ProcessorMissingFieldSnafu {
118                            processor: self.kind(),
119                            field: field.input_field(),
120                        }
121                        .fail();
122                    }
123                }
124                Some(v) => {
125                    let processed = self.process_field(v)?;
126                    let output_index = field.target_or_input_field();
127                    val.insert(KeyString::from(output_index.to_string()), processed);
128                }
129            }
130        }
131        Ok(val)
132    }
133}
134
135/// Parses a logfmt line into key-value pairs.
136///
137/// Follows the grammar used by common logfmt implementations (e.g. go-logfmt):
138/// - pairs are separated by whitespace;
139/// - a key is a run of printable characters other than `=`, `"` and whitespace;
140/// - a value is either unquoted (ends at the next whitespace) or double-quoted,
141///   where `\"`, `\\`, `\n`, `\r`, `\t`, `\/`, `\b`, `\f` and `\uXXXX` escapes are supported;
142/// - `key=` yields an empty string and a bare `key` yields `true`.
143fn parse_logfmt(line: &str) -> std::result::Result<BTreeMap<KeyString, VrlValue>, String> {
144    let mut output = BTreeMap::new();
145    let mut chars = line.char_indices().peekable();
146
147    loop {
148        // Skip separators.
149        while chars.next_if(|(_, c)| c.is_whitespace()).is_some() {}
150        let Some(&(key_start, first)) = chars.peek() else {
151            break;
152        };
153        if first == '=' || first == '"' {
154            return Err(format!("unexpected '{first}' at position {key_start}"));
155        }
156
157        let mut key_end = line.len();
158        while let Some(&(i, c)) = chars.peek() {
159            if c == '=' || c.is_whitespace() {
160                key_end = i;
161                break;
162            }
163            if c == '"' || c.is_control() {
164                return Err(format!("invalid character {c:?} in key at position {i}"));
165            }
166            chars.next();
167        }
168        let key = KeyString::from(&line[key_start..key_end]);
169
170        if chars.next_if(|(_, c)| *c == '=').is_none() {
171            // Bare key, e.g. `debug`.
172            output.insert(key, VrlValue::Boolean(true));
173            continue;
174        }
175
176        let value = match chars.peek() {
177            Some(&(_, '"')) => {
178                chars.next();
179                parse_quoted_value(&mut chars)?
180            }
181            Some(&(value_start, _)) => {
182                let mut value_end = line.len();
183                while let Some(&(i, c)) = chars.peek() {
184                    if c.is_whitespace() {
185                        value_end = i;
186                        break;
187                    }
188                    if c == '"' || c == '=' {
189                        return Err(format!(
190                            "unexpected '{c}' in unquoted value at position {i}"
191                        ));
192                    }
193                    chars.next();
194                }
195                line[value_start..value_end].to_string()
196            }
197            None => String::new(),
198        };
199        output.insert(key, VrlValue::Bytes(Bytes::from(value)));
200    }
201
202    Ok(output)
203}
204
205/// Parses the rest of a double-quoted value; the opening quote is already consumed.
206fn parse_quoted_value(
207    chars: &mut std::iter::Peekable<std::str::CharIndices<'_>>,
208) -> std::result::Result<String, String> {
209    let mut value = String::new();
210    loop {
211        let Some((i, c)) = chars.next() else {
212            return Err("unterminated quoted value".to_string());
213        };
214        match c {
215            '"' => {
216                if let Some(&(j, next)) = chars.peek()
217                    && !next.is_whitespace()
218                {
219                    return Err(format!(
220                        "expect whitespace after quoted value at position {j}, found '{next}'"
221                    ));
222                }
223                return Ok(value);
224            }
225            '\\' => {
226                let Some((_, escaped)) = chars.next() else {
227                    return Err("unterminated escape sequence".to_string());
228                };
229                match escaped {
230                    '"' => value.push('"'),
231                    '\\' => value.push('\\'),
232                    '/' => value.push('/'),
233                    'n' => value.push('\n'),
234                    'r' => value.push('\r'),
235                    't' => value.push('\t'),
236                    'b' => value.push('\u{8}'),
237                    'f' => value.push('\u{c}'),
238                    'u' => {
239                        let hex: String = chars.by_ref().take(4).map(|(_, c)| c).collect();
240                        let decoded = (hex.len() == 4)
241                            .then(|| u32::from_str_radix(&hex, 16).ok())
242                            .flatten()
243                            .and_then(char::from_u32)
244                            .ok_or_else(|| format!("invalid unicode escape '\\u{hex}'"))?;
245                        value.push(decoded);
246                    }
247                    other => {
248                        return Err(format!("invalid escape '\\{other}' at position {i}"));
249                    }
250                }
251            }
252            c => value.push(c),
253        }
254    }
255}
256
257#[cfg(test)]
258mod tests {
259    use super::*;
260
261    fn obj(pairs: &[(&str, VrlValue)]) -> BTreeMap<KeyString, VrlValue> {
262        pairs
263            .iter()
264            .map(|(k, v)| (KeyString::from(*k), v.clone()))
265            .collect()
266    }
267
268    fn s(v: &str) -> VrlValue {
269        VrlValue::Bytes(Bytes::from(v.to_string()))
270    }
271
272    #[test]
273    fn test_parse_logfmt() {
274        let parsed = parse_logfmt(
275            r#"ts=2026-10-14T10:00:00Z level=info msg="hello \"world\"" path=/api/v1 empty= debug"#,
276        )
277        .unwrap();
278        assert_eq!(
279            parsed,
280            obj(&[
281                ("ts", s("2026-10-14T10:00:00Z")),
282                ("level", s("info")),
283                ("msg", s(r#"hello "world""#)),
284                ("path", s("/api/v1")),
285                ("empty", s("")),
286                ("debug", VrlValue::Boolean(true)),
287            ])
288        );
289    }
290
291    #[test]
292    fn test_parse_logfmt_whitespace_escapes_and_duplicates() {
293        let parsed = parse_logfmt("  a=1\tb=\"x\\ny\\u00e9\"  a=2  ").unwrap();
294        assert_eq!(parsed, obj(&[("a", s("2")), ("b", s("x\nyé"))]));
295        assert!(parse_logfmt("").unwrap().is_empty());
296        assert!(parse_logfmt("   ").unwrap().is_empty());
297        assert_eq!(parse_logfmt("k=\"\"").unwrap(), obj(&[("k", s(""))]));
298        assert_eq!(parse_logfmt("k=").unwrap(), obj(&[("k", s(""))]));
299    }
300
301    #[test]
302    fn test_parse_logfmt_invalid() {
303        for input in [
304            "=value",
305            "\"key\"=value",
306            "msg=\"unterminated",
307            "msg=\"bad\\q\"",
308            "msg=\"a\"b",
309            "k=a\"b",
310            "k=a=b",
311            "msg=\"\\u12\"",
312        ] {
313            assert!(parse_logfmt(input).is_err(), "expect error for {input:?}");
314        }
315    }
316
317    #[test]
318    fn test_logfmt_processor() {
319        let processor = LogfmtProcessor::default();
320        let result = processor.process_field(&s("level=warn code=42")).unwrap();
321        assert_eq!(
322            result,
323            VrlValue::Object(obj(&[("code", s("42")), ("level", s("warn"))]))
324        );
325        assert!(processor.process_field(&VrlValue::Integer(1)).is_err());
326        assert!(processor.process_field(&s("msg=\"oops")).is_err());
327    }
328}
pipeline/etl/processor/logfmt.rs

pipeline/etl/processor/
logfmt.rs
