feat(new transform): New remap transform

.meta/transforms/remap.toml.erb

@@ -0,0 +1,166 @@
+[transforms.remap]
+title = "Remap"
+allow_you_to_description = "remap one or more log fields"
+beta = false
+common = false
+function_category = "schema"
+input_types = ["log"]
+output_types = ["log"]
+requirements = {}
+
+<%= render("_partials/fields/_component_options.toml", type: "transform", name: "remap") %>
+
+[transforms.remap.options.mapping]
+type = "string"
+common = true
+required = true
+description = """\
+A mapping that describes field assignments and deletions to be performed on log \
+events.\
+"""
+examples = [
+  """.type = "foo"""",
+  """.new_field = .old_field * 2
+del(.old_field)""",
+  """only_fields(.message, .timestamp, .name)""",
+]
+
+[[transforms.remap.examples]]
+label = "Generic"
+body = """\
+The remap transform makes it easy to add, rename and remove fields with a series of statements. Given events of the following form:
+
+```json
+{
+  "name": "Foo McBarson",
+  "friends": 23,
+  "enemies": 17
+}
+```
+
+If, say, we wished to add a new field `type` with a static value `"human_person"`, a dynamic field `acquaintances` that equals the sum of the fields `friends` and `enemies`, and remove the field `enemies`, we can achieve that with the following mapping:
+
+```toml
+[transforms.remap_human_person]
+  type = "remap"
+  mapping = \"\"\"
+    .type = "human_person"
+    .acquaintances = .friends + .enemies
+    del(.enemies)
+  \"\"\"
+```
+
+And log events will be output with the following structure:
+
+```json
+{
+  "name": "Foo McBarson",
+  "type": "human_person",
+  "friends": 23,
+  "acquaintances": 40
+}
+```\
+"""
+
+[[transforms.remap.examples]]
+label = "Conditional mapping"
+body = """\
+Sometimes there are mappings that we only wish to execute given certain conditions, the remap transform allows you to express this with if statements. For example, given events of the following form:
+
+```json
+{
+  "type": "foo",
+  "foo": {
+    "body": "hello world",
+    "id": "XXX"
+  }
+}
+```
+
+And, occasionally of the following alternative form:
+
+```json
+{
+  "type": "bar",
+  "body": "hello world",
+  "id": "YYY"
+}
+```
+
+If we wished to create a consistent format between the two, where the fields `body` and `id` are always at the root of the event, we can achieve this by conditionally mapping against the field `type` of the input event:
+
+```toml
+[transforms.remap_human_person]
+  type = "remap"
+  mapping = \"\"\"
+    if .type == "foo" {
+      .body = .foo.body
+      .id = .foo.id
+      del(.foo)
+    }
+  \"\"\"
+```
+
+And with the above mapping our events will be output with the following consistent structure:
+
+```json
+{
+  "type": "foo",
+  "body": "hello world",
+  "id": "XXX"
+}
+```\
+"""
+
+[[transforms.remap.examples]]
+label = "Type Coercion"
+body = """\
+The remap transform offers the functions `string`, `int`, `float`, `bool` and `timestamp`, that attempt to coerce values into fixed types. Each of these functions take a query parameter describing the value to coerce and these queries can themselves include arithmetic and functions.
+
+The `timestamp` function also requires a second parameter describing the timestamp format to parse. All other functions have an optional second parameter that describes a default value to return if the target value does not exist or otherwise cannot be coerced.
+
+For example, given the following event:
+
+```json
+{
+  // ... existing fields
+  "bytes_in": "5667",
+  "bytes_out": "20574",
+  "host": "5.86.210.12",
+  "message": "GET /embrace/supply-chains/dynamic/vertical",
+  "status": "201",
+  "timestamp": "19/06/2019:17:20:49 -0400",
+  "user_id": "zieme4647"
+}
+```
+
+And the following configuration:
+
+```toml
+[transforms.remap_log]
+  type = "remap"
+  mapping = \"\"\"
+    .bytes_in = int(.bytes_in)
+    .bytes_out = int(.bytes_out)
+    .imaginary_bool = bool(.doesnt.exist, true)
+    .timestamp = timestamp(.timestamp, "%d/%m/%Y:%H:%M:%S %z")
+    .status = int(.status, 200)
+  \"\"\"
+```
+
+A log event will be output with the following structure:
+
+```json
+{
+  // ... existing fields
+  "bytes_in": 5667,
+  "bytes_out": 20574,
+  "host": "5.86.210.12",
+  "message": "GET /embrace/supply-chains/dynamic/vertical",
+  "status": 201,
+  "timestamp": "2019-06-19T17:20:49-04:00",
+  "user_id": "zieme4647",
+  "imaginary_bool": true
+}
+```\
+"""

Cargo.toml

@@ -127,6 +127,8 @@ bytesize = { version = "1.0.0", optional = true }
 glob = "0.2.11"
 grok = { version = "~1.0.1", optional = true }
 nom = { version = "5.1.2", optional = true }
+pest = "2.1.3"
+pest_derive = "2.1.0"
 uuid = { version = "0.7", features = ["serde", "v4"], optional = true }
 exitcode = "1.1.2"
 snafu = { version = "0.6", features = ["futures-01", "futures"] }
@@ -289,6 +291,7 @@ transforms = [
   "transforms-lua",
   "transforms-merge",
   "transforms-regex_parser",
+  "transforms-remap",
   "transforms-remove_fields",
   "transforms-remove_tags",
   "transforms-rename_fields",
@@ -316,6 +319,7 @@ transforms-logfmt_parser = ["logfmt"]
 transforms-lua = ["rlua"]
 transforms-merge = []
 transforms-regex_parser = []
+transforms-remap = []
 transforms-remove_fields = []
 transforms-remove_tags = []
 transforms-rename_fields = []

benches/bench.rs

@@ -2,16 +2,23 @@ use criterion::{criterion_group, criterion_main, Benchmark, Criterion, Throughpu
 
 use approx::assert_relative_eq;
 use futures::{compat::Future01CompatExt, future, stream, StreamExt};
+use indexmap::IndexMap;
 use rand::{
     distributions::{Alphanumeric, Uniform},
     prelude::*,
 };
 use std::convert::TryFrom;
+use string_cache::DefaultAtom as Atom;
 use vector::config::{self, TransformConfig, TransformContext};
 use vector::event::Event;
 use vector::test_util::{
     next_addr, runtime, send_lines, start_topology, wait_for_tcp, CountReceiver,
 };
+use vector::transforms::{
+    add_fields::AddFields,
+    remap::{Remap, RemapConfig},
+    Transform,
+};
 use vector::{sinks, sources, transforms};
 
 mod batch;
@@ -32,6 +39,7 @@ criterion_group!(
     benchmark_complex,
     bench_elasticsearch_index,
     benchmark_regex,
+    benchmark_remap,
 );
 criterion_main!(
     benches,
@@ -657,6 +665,90 @@ fn bench_elasticsearch_index(c: &mut Criterion) {
     );
 }
 
+fn benchmark_remap(c: &mut Criterion) {
+    let event = {
+        let mut event = Event::from("augment me");
+        event.as_mut_log().insert("copy_from", "buz");
+        event
+    };
+
+    c.bench_function("add fields with remap", |b| {
+        let conf = RemapConfig {
+            mapping: r#".foo = "bar"
+            .bar = "baz"
+            .copy = .copy_from"#
+                .to_string(),
+            drop_on_err: true,
+        };
+        let mut tform = Remap::new(conf).unwrap();
+
+        b.iter(|| {
+            let result = tform.transform(event.clone()).unwrap();
+            assert_eq!(
+                result
+                    .as_log()
+                    .get(&Atom::from("foo"))
+                    .unwrap()
+                    .to_string_lossy(),
+                "bar"
+            );
+            assert_eq!(
+                result
+                    .as_log()
+                    .get(&Atom::from("bar"))
+                    .unwrap()
+                    .to_string_lossy(),
+                "baz"
+            );
+            assert_eq!(
+                result
+                    .as_log()
+                    .get(&Atom::from("copy"))
+                    .unwrap()
+                    .to_string_lossy(),
+                "buz"
+            );
+        })
+    });
+
+    c.bench_function("add fields with add_fields", |b| {
+        let mut fields = IndexMap::new();
+        fields.insert("foo".into(), "bar".into());
+        fields.insert("bar".into(), "baz".into());
+        fields.insert("copy".into(), "{{ copy_from }}".into());
+
+        let mut tform = AddFields::new(fields, true);
+
+        b.iter(|| {
+            let result = tform.transform(event.clone()).unwrap();
+            assert_eq!(
+                result
+                    .as_log()
+                    .get(&Atom::from("foo"))
+                    .unwrap()
+                    .to_string_lossy(),
+                "bar"
+            );
+            assert_eq!(
+                result
+                    .as_log()
+                    .get(&Atom::from("bar"))
+                    .unwrap()
+                    .to_string_lossy(),
+                "baz"
+            );
+            assert_eq!(
+                result
+                    .as_log()
+                    .get(&Atom::from("copy"))
+                    .unwrap()
+                    .to_string_lossy(),
+                "buz"
+            );
+        })
+    });
+}
+
 fn random_lines(size: usize) -> impl Iterator<Item = String> {
     let mut rng = SmallRng::from_rng(thread_rng()).unwrap();