add drop_on_abort option

Signed-off-by: Jean Mertz <git@jeanmertz.com>

benches/remap.rs

@@ -46,7 +46,8 @@ fn benchmark_remap(c: &mut Criterion) {
                     .copy = string!(.copy_from)
                 "#}
                 .to_string(),
-                drop_on_err: true,
+                drop_on_error: true,
+                drop_on_abort: true,
             })
             .unwrap(),
         );
@@ -106,7 +107,8 @@ fn benchmark_remap(c: &mut Criterion) {
         let mut tform: Box<dyn FunctionTransform> = Box::new(
             Remap::new(RemapConfig {
                 source: ".bar = parse_json!(string!(.foo))".to_owned(),
-                drop_on_err: false,
+                drop_on_error: true,
+                drop_on_abort: true,
             })
             .unwrap(),
         );
@@ -175,7 +177,8 @@ fn benchmark_remap(c: &mut Criterion) {
                     .timestamp = parse_timestamp!(string!(.timestamp), format: "%d/%m/%Y:%H:%M:%S %z")
                 "#}
                 .to_owned(),
-                drop_on_err: true,
+                drop_on_error: true,
+                drop_on_abort: true,
             })
             .unwrap(),
         );

docs/reference/components/transforms/remap.cue

@@ -69,9 +69,14 @@ components: transforms: "remap": {
 				Drop the event if the VRL program returns an error at runtime.
 				"""
 			required: false
-			type: bool: {
-				default: false
-			}
+			type: bool: default: false
+		}
+		drop_on_abort: {
+			description: """
+				Drop the event if the VRL program is manually aborted through the `abort` statement.
+				"""
+			required: false
+			type: bool: default: true
 		}
 	}
 

src/internal_events/remap.rs

@@ -45,3 +45,22 @@ impl InternalEvent for RemapConditionExecutionError {
         counter!("processing_errors_total", 1);
     }
 }
+
+#[derive(Debug)]
+pub struct RemapMappingAbort {
+    /// If set to true, the remap transform has dropped the event after an abort
+    /// during mapping. This internal event will reflect that in its messaging.
+    pub event_dropped: bool,
+}
+
+impl InternalEvent for RemapMappingAbort {
+    fn emit_logs(&self) {
+        let message = if self.event_dropped {
+            "Event mapping aborted; discarding event."
+        } else {
+            "Event mapping aborted."
+        };
+
+        debug!(message, internal_log_rate_secs = 30)
+    }
+}

src/transforms/remap.rs

@@ -1,20 +1,21 @@
 use crate::{
     config::{DataType, GlobalOptions, TransformConfig, TransformDescription},
     event::Event,
-    internal_events::RemapMappingError,
+    internal_events::{RemapMappingAbort, RemapMappingError},
     transforms::{FunctionTransform, Transform},
     Result,
 };
 use serde::{Deserialize, Serialize};
 use vrl::diagnostic::Formatter;
-use vrl::{Program, Runtime};
+use vrl::{Program, Runtime, Terminate};
 
 #[derive(Deserialize, Serialize, Debug, Clone, Derivative)]
 #[serde(deny_unknown_fields, default)]
 #[derivative(Default)]
 pub struct RemapConfig {
     pub source: String,
     pub drop_on_error: bool,
+    pub drop_on_abort: bool,
 }
 
 inventory::submit! {
@@ -47,6 +48,7 @@ impl TransformConfig for RemapConfig {
 pub struct Remap {
     program: Program,
     drop_on_error: bool,
+    drop_on_abort: bool,
 }
 
 impl Remap {
@@ -60,6 +62,7 @@ impl Remap {
         Ok(Remap {
             program,
             drop_on_error: config.drop_on_error,
+            drop_on_abort: config.drop_on_abort,
         })
     }
 }
@@ -84,9 +87,20 @@ impl FunctionTransform for Remap {
 
         match result {
             Ok(_) => output.push(event),
-            Err(error) => {
+            Err(Terminate::Abort) => {
+                emit!(RemapMappingAbort {
+                    event_dropped: self.drop_on_abort,
+                });
+
+                if self.drop_on_abort {
+                    return;
+                }
+
+                output.push(event)
+            }
+            Err(Terminate::Error(error)) => {
                 emit!(RemapMappingError {
-                    error: error.to_string(),
+                    error,
                     event_dropped: self.drop_on_error,
                 });
 

tests/behavior/transforms/remap.toml

@@ -1,3 +1,45 @@
+[transforms.remap_abort]
+  inputs = []
+  type = "remap"
+  drop_on_abort = false
+  source = """
+    .foo = false
+    abort
+    .bar = false
+  """
+[[tests]]
+  name = "remap_abort"
+  [tests.input]
+    insert_at = "remap_abort"
+    type = "log"
+    [tests.input.log_fields]
+      foo = true
+      bar = true
+  [[tests.outputs]]
+    extract_from = "remap_abort"
+    [[tests.outputs.conditions]]
+      type = "remap"
+      source = ".foo == false && .bar == true"
+
+[transforms.remap_abort_drop_on_abort]
+  inputs = []
+  type = "remap"
+  drop_on_abort = true
+  source = """
+    .foo = false
+    abort
+    .bar = false
+  """
+[[tests]]
+  name = "remap_abort_drop_on_abort"
+  no_outputs_from = ["remap_abort_drop_on_abort"]
+  [tests.input]
+    insert_at = "remap_abort_drop_on_abort"
+    type = "log"
+    [tests.input.log_fields]
+      foo = true
+      bar = true
+
 [transforms.remap_nested]
   inputs = []
   type = "remap"
@@ -83,7 +125,7 @@
 [transforms.remap_arithmetic_error]
   inputs = []
   type = "remap"
-  drop_on_err = true
+  drop_on_error = true
   source = """
     a = 10
     b = 0
@@ -319,7 +361,7 @@
 [transforms.remap_function_upcase_error]
   inputs = []
   type = "remap"
-  drop_on_err = true
+  drop_on_error = true
   source = """
     .a = upcase(string!(.a))
     .b = upcase(string!(.b))
@@ -337,7 +379,7 @@
 [transforms.remap_function_downcase]
   inputs = []
   type = "remap"
-  drop_on_err = true
+  drop_on_error = true
   source = """
     .a = downcase(string!(.a))
     .b = downcase(string!(.b))
@@ -371,7 +413,7 @@
 [transforms.remap_function_downcase_error]
   inputs = []
   type = "remap"
-  drop_on_err = true
+  drop_on_error = true
   source = """
     .a = downcase(string!(.a))
     .b = downcase(string!(.b))
@@ -442,7 +484,7 @@
 [transforms.remap_function_sha1_error]
   inputs = []
   type = "remap"
-  drop_on_err = true
+  drop_on_error = true
   source = """
     .a = sha1(string!(.a))
     .b = sha1(string!(.b))
@@ -487,7 +529,7 @@
 [transforms.remap_function_md5_error]
   inputs = []
   type = "remap"
-  drop_on_err = true
+  drop_on_error = true
   source = """
     .a = md5(string!(.a))
     .b = md5(string!(.b))
@@ -1331,7 +1373,7 @@
 # [transforms.remap_function_assert_pass]
 #   inputs = []
 #   type = "remap"
-#   drop_on_err = true
+#   drop_on_error = true
 #   source = """
 #     assert!(.foo, message: "assert failed")
 #     .check = "checked"
@@ -1354,7 +1396,7 @@
 # [transforms.remap_function_assert_fail]
 #   inputs = []
 #   type = "remap"
-#   drop_on_err = true
+#   drop_on_error = true
 #   source = """
 #     assert!(.foo, message: "assert failed")
 #   """