Repository: rcoh/angle-grinder (language: rust)

Issue:
Support logfmt input
https://www.brandur.org/logfmt describes the logfmt input, which is supported by a variety of platforms and logging libraries. I'd like to consume them with angle-grinder.


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 5a34f3b..8cb06ea 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -26,6 +26,7 @@ dependencies = [
  "im 13.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "itertools 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "lazy_static 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "logfmt 0.0.2 (git+https://github.com/brandur/logfmt?tag=v0.0.2)",
  "maplit 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "nom 4.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "nom_locate 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -743,6 +744,11 @@ dependencies = [
  "cfg-if 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "logfmt"
+version = "0.0.2"
+source = "git+https://github.com/brandur/logfmt?tag=v0.0.2#ee7b87deb117715bb382f03e1b191e57c5071bbf"
+
 [[package]]
 name = "maplit"
 version = "1.0.1"
@@ -2101,6 +2107,7 @@ dependencies = [
 "checksum libflate 0.1.19 (registry+https://github.com/rust-lang/crates.io-index)" = "bff3ac7d6f23730d3b533c35ed75eef638167634476a499feef16c428d74b57b"
 "checksum lock_api 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "62ebf1391f6acad60e5c8b43706dde4582df75c06698ab44511d15016bc2442c"
 "checksum log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)" = "c84ec4b527950aa83a329754b01dbe3f58361d1c5efacd1f6d68c494d08a17c6"
+"checksum logfmt 0.0.2 (git+https://github.com/brandur/logfmt?tag=v0.0.2)" = "<none>"
 "checksum maplit 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "08cbb6b4fef96b6d77bfc40ec491b1690c779e77b05cd9f07f787ed376fd4c43"
 "checksum matches 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "7ffc5c5338469d4d3ea17d269fa8ea3512ad247247c30bd2df69e68309ed0a08"
 "checksum memchr 2.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "db4c41318937f6e76648f42826b1d9ade5c09cafb5aef7e351240a70f39206e9"
diff --git a/Cargo.toml b/Cargo.toml
index 9aa3cef..832844b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -38,6 +38,7 @@ annotate-snippets = { version = "0.5.0", features = ["ansi_term"] }
 atty = "0.2.0"
 lazy_static = "1.2.0"
 im = "13.0.0"
+logfmt = { git = "https://github.com/brandur/logfmt", tag = "v0.0.2" }
 
 [dev-dependencies]
 assert_cli = "0.6.3"
diff --git a/README.md b/README.md
index 6784832..ac32061 100644
--- a/README.md
+++ b/README.md
@@ -95,6 +95,22 @@ Given input like:
 ```
 ![json.gif](/screen_shots/json.gif)
 
+##### Logfmt
+`logfmt [from other_field]`: Extract logfmt-serialized rows into fields for later use. If the row is _not_ valid logfmt, then it is dropped. Optionally, `from other_field` can be specified. Logfmt is a an output format commonly used by Heroku and Splunk, described at https://www.brandur.org/logfmt.
+
+*Examples*:
+```agrind
+* | logfmt
+```
+
+Given input like:
+```
+{"key": "blah", "nested_key": "some=logfmt data=more"}
+```
+```agrind
+* | json | logfmt from nested_key | fields some
+```
+
 ##### Parse
 `parse "* pattern * otherpattern *" [from field] as a,b,c [nodrop]`: Parse text that matches the pattern into variables. Lines that don't match the pattern will be dropped unless `nodrop` is specified. `*` is equivalent to regular expression `.*` and is greedy.
 By default, `parse` operates on the raw text of the message. With `from field_name`, parse will instead process input from a specific column.
diff --git a/src/lang.rs b/src/lang.rs
index 1c769e0..894f349 100644
--- a/src/lang.rs
+++ b/src/lang.rs
@@ -66,7 +66,9 @@ pub const VALID_AGGREGATES: &'static [&str] = &[
     "sort",
 ];
 
-pub const VALID_INLINE: &'static [&str] = &["parse", "limit", "json", "total", "fields", "where"];
+pub const VALID_INLINE: &'static [&str] = &[
+    "parse", "limit", "json", "logfmt", "total", "fields", "where",
+];
 
 lazy_static! {
     pub static ref VALID_OPERATORS: Vec<&'static str> =
@@ -216,6 +218,9 @@ pub enum InlineOperator {
     Json {
         input_column: Option<String>,
     },
+    Logfmt {
+        input_column: Option<String>,
+    },
     Parse {
         pattern: Keyword,
         fields: Vec<String>,
@@ -392,6 +397,12 @@ named!(json<Span, Positioned<InlineOperator>>, with_pos!(ws!(do_parse!(
     (InlineOperator::Json { input_column: from_column_opt.map(|s|s.to_string()) })
 ))));
 
+named!(logfmt<Span, Positioned<InlineOperator>>, with_pos!(ws!(do_parse!(
+    tag!("logfmt") >>
+    from_column_opt: opt!(ws!(preceded!(tag!("from"), ident))) >>
+    (InlineOperator::Logfmt { input_column: from_column_opt.map(|s|s.to_string()) })
+))));
+
 named!(whre<Span, Positioned<InlineOperator>>, with_pos!(ws!(do_parse!(
     tag!("where") >>
     ex: opt!(with_pos!(expr)) >>
@@ -561,7 +572,7 @@ named!(p_nn<Span, Positioned<AggregateFunction>>, ws!(
 ));
 
 named!(inline_operator<Span, Operator>,
-    map!(alt_complete!(parse | json | fields | whre | limit | total), Operator::Inline)
+    map!(alt_complete!(parse | json | logfmt | fields | whre | limit | total), Operator::Inline)
 );
 
 named!(aggregate_function<Span, Positioned<AggregateFunction>>, do_parse!(
@@ -939,6 +950,15 @@ mod tests {
                 value: InlineOperator::Json { input_column: None }
             })
         );
+        expect!(
+            operator,
+            "  logfmt",
+            Operator::Inline(Positioned {
+                start_pos: QueryPosition(2),
+                end_pos: QueryPosition(8),
+                value: InlineOperator::Logfmt { input_column: None }
+            })
+        );
         expect!(
             operator,
             r#" parse "[key=*]" from field as v "#,
@@ -1219,4 +1239,29 @@ mod tests {
             }
         );
     }
+
+    #[test]
+    fn logfmt_operator() {
+        let query_str = r#"* | logfmt from col | sort by foo dsc "#;
+        expect!(
+            query,
+            query_str,
+            Query {
+                search: Search::And(vec![]),
+                operators: vec![
+                    Operator::Inline(Positioned {
+                        start_pos: QueryPosition(4),
+                        end_pos: QueryPosition(20),
+                        value: InlineOperator::Logfmt {
+                            input_column: Some("col".to_string()),
+                        }
+                    }),
+                    Operator::Sort(SortOperator {
+                        sort_cols: vec!["foo".to_string()],
+                        direction: SortMode::Descending,
+                    }),
+                ],
+            }
+        );
+    }
 }
diff --git a/src/operator.rs b/src/operator.rs
index 846aed6..074513f 100644
--- a/src/operator.rs
+++ b/src/operator.rs
@@ -2,6 +2,7 @@ extern crate itertools;
 extern crate quantiles;
 extern crate regex;
 //extern crate regex_syntax;
+extern crate logfmt;
 extern crate serde_json;
 
 use self::quantiles::ckms::CKMS;
@@ -896,6 +897,37 @@ impl UnaryPreAggFunction for ParseJson {
     }
 }
 
+#[derive(Clone)]
+pub struct ParseLogfmt {
+    input_column: Option<Expr>,
+}
+
+impl ParseLogfmt {
+    pub fn new(input_column: Option<String>) -> ParseLogfmt {
+        ParseLogfmt {
+            input_column: input_column.map(Expr::Column),
+        }
+    }
+}
+
+impl UnaryPreAggFunction for ParseLogfmt {
+    fn process(&self, rec: Record) -> Result<Option<Record>, EvalError> {
+        let pairs = {
+            let inp = get_input(&rec, &self.input_column)?;
+            // Record includes the trailing newline, while logfmt considers that part of the
+            // message if present. Trim any trailing whitespace.
+            logfmt::parse(&inp.trim_end())
+        };
+        let res = {
+            pairs.into_iter().fold(rec, |record, pair| match &pair.val {
+                None => record.put(&pair.key, data::Value::None),
+                Some(val) => record.put(&pair.key, data::Value::from_string(val)),
+            })
+        };
+        Ok(Some(res))
+    }
+}
+
 /// The definition for a limit operator, which is a positive number used to specify whether
 /// the first N rows should be passed through to the downstream operators.  Negative limits are
 /// not supported at this time.
@@ -1072,6 +1104,22 @@ mod tests {
         );
     }
 
+    #[test]
+    fn logfmt() {
+        let rec = Record::new(&(r#"k1=5 k2=5.5 k3="a str" k4="#.to_string() + "\n"));
+        let parser = ParseLogfmt::new(None);
+        let rec = parser.process(rec).unwrap().unwrap();
+        assert_eq!(
+            rec.data,
+            hashmap! {
+                "k1".to_string() => Value::Int(5),
+                "k2".to_string() => Value::from_float(5.5),
+                "k3".to_string() => Value::Str("a str".to_string()),
+                "k4".to_string() => Value::Str("".to_string())
+            }
+        );
+    }
+
     #[test]
     fn fields_only() {
         let rec = Record::new("");
diff --git a/src/typecheck.rs b/src/typecheck.rs
index 4dfa42a..aa7151d 100644
--- a/src/typecheck.rs
+++ b/src/typecheck.rs
@@ -81,6 +81,9 @@ impl lang::Positioned<lang::InlineOperator> {
             lang::InlineOperator::Json { input_column } => {
                 Ok(Box::new(operator::ParseJson::new(input_column)))
             }
+            lang::InlineOperator::Logfmt { input_column } => {
+                Ok(Box::new(operator::ParseLogfmt::new(input_column)))
+            }
             lang::InlineOperator::Parse {
                 pattern,
                 fields,


Test patch:
diff --git a/test_files/test_logfmt.log b/test_files/test_logfmt.log
new file mode 100644
index 0000000..8644208
--- /dev/null
+++ b/test_files/test_logfmt.log
@@ -0,0 +1,3 @@
+level=info msg="Stopping all fetchers" tag=stopping_fetchers id=ConsumerFetcherManager-1382721708341 module=kafka.consumer.ConsumerFetcherManager
+level=info msg="Starting all fetchers" tag=starting_fetchers id=ConsumerFetcherManager-1382721708342 module=kafka.consumer.ConsumerFetcherManager
+level=warn msg="Fetcher failed to start" tag=errored_fetchers id=ConsumerFetcherManager-1382721708342 module=kafka.consumer.ConsumerFetcherManager
diff --git a/test_files/test_nested_logfmt.log b/test_files/test_nested_logfmt.log
new file mode 100644
index 0000000..3c8be00
--- /dev/null
+++ b/test_files/test_nested_logfmt.log
@@ -0,0 +1,1 @@
+{"key": "blah", "nested_key": "some=logfmt data=more"}
diff --git a/tests/integration.rs b/tests/integration.rs
index fd6b278..7c5b969 100644
--- a/tests/integration.rs
+++ b/tests/integration.rs
@@ -74,6 +74,11 @@ mod integration {
         structured_test(include_str!("structured_tests/parse_nodrop.toml"));
     }
 
+    #[test]
+    fn logfmt_operator() {
+        structured_test(include_str!("structured_tests/logfmt.toml"));
+    }
+
     #[test]
     fn sum_operator() {
         structured_test(include_str!("structured_tests/sum.toml"));
diff --git a/tests/structured_tests/logfmt.toml b/tests/structured_tests/logfmt.toml
new file mode 100644
index 0000000..c3a92fe
--- /dev/null
+++ b/tests/structured_tests/logfmt.toml
@@ -0,0 +1,11 @@
+query = """* | logfmt | fields thing_a, thing_b"""
+input = """
+thing_a=5 thing_b=red
+thing_a=6 thing_b=yellow
+thing_a=7 thing_b=blue
+"""
+output = """
+[thing_a=5]              [thing_b=red]
+[thing_a=6]              [thing_b=yellow]
+[thing_a=7]              [thing_b=blue]
+"""