enhancement(vrl): add encode_key_value function (#7751)

Co-authored-by: Jean Mertz <git@jeanmertz.com>
Signed-off-by: prognant <pierre.rognant@datadoghq.com>

docs/reference/remap/functions/encode_key_value.cue

@@ -0,0 +1,89 @@
+package metadata
+
+remap: functions: encode_key_value: {
+	category: "Codec"
+	description: #"""
+		Encodes the `value` to in key/value format with customizable delimiters. Default delimiters match
+		the [logfmt](\(urls.logfmt)) format.
+		"""#
+	notices: [
+		"""
+			If `fields_ordering` is specified then the function is fallible else it is infallible.
+			""",
+	]
+
+	arguments: [
+		{
+			name:        "value"
+			description: "The value to convert to a string."
+			required:    true
+			type: ["object"]
+		},
+		{
+			name:        "fields_ordering"
+			description: "The ordering of fields to preserve. Any fields not in this list will appear unordered, after any ordered fields."
+			required:    false
+			type: ["array"]
+		},
+		{
+			name:        "key_value_delimiter"
+			description: "The string that separates the key from the value."
+			required:    false
+			default:     "="
+			type: ["string"]
+		},
+		{
+			name:        "field_delimiter"
+			description: "The string that separates each key/value pair."
+			required:    false
+			default:     " "
+			type: ["string"]
+		},
+	]
+	internal_failure_reasons: [
+		"`fields_ordering` contains a non-string element",
+	]
+	return: types: ["string"]
+
+	examples: [
+		{
+			title: "Encode with default delimiters (no ordering)"
+			source: """
+				encode_key_value({"ts": "2021-06-05T17:20:00Z", "msg": "This is a message", "lvl": "info"})
+				"""
+			return: #"lvl=info msg="This is a message" ts=2021-06-05T17:20:00Z"#
+		},
+		{
+			title: "Encode with default delimiters (fields ordering)"
+			source: """
+				encode_key_value!({"ts": "2021-06-05T17:20:00Z", "msg": "This is a message", "lvl": "info", "log_id": 12345}, ["ts", "lvl", "msg"])
+				"""
+			return: #"ts=2021-06-05T17:20:00Z lvl=info msg="This is a message" log_id=12345"#
+		},
+		{
+			title: "Encode with default delimiters (nested fields)"
+			source: """
+				encode_key_value({"agent": {"name": "vector"}, "log": {"file": {"path": "my.log"}}, "event": "log"})
+				"""
+			return: #"agent.name=vector event=log log.file.path=my.log"#
+		},
+		{
+			title: "Encode with default delimiters (nested fields ordering)"
+			source: """
+				encode_key_value!({"agent": {"name": "vector"}, "log": {"file": {"path": "my.log"}}, "event": "log"}, ["event", "log.file.path", "agent.name"])
+				"""
+			return: #"event=log log.file.path=my.log agent.name=vector"#
+		},
+		{
+			title: "Encode with custom delimiters (no ordering)"
+			source: """
+				encode_key_value(
+					{"ts": "2021-06-05T17:20:00Z", "msg": "This is a message", "lvl": "info"},
+					field_delimiter: ",",
+					key_value_delimiter: ":"
+				)
+				"""
+			return: #"lvl:info,msg:"This is a message",ts:2021-06-05T17:20:00Z"#
+		},
+	]
+}

docs/reference/remap/functions/encode_logfmt.cue

@@ -5,6 +5,7 @@ remap: functions: encode_logfmt: {
 	description: #"""
 		Encodes the `value` to [logfmt](\#(urls.logfmt)).
 		"""#
+	notices:     functions.encode_key_value.notices
 
 	arguments: [
 		{
@@ -29,7 +30,7 @@ remap: functions: encode_logfmt: {
 		{
 			title: "Encode to logfmt (no ordering)"
 			source: """
-				encode_logfmt!({"ts": "2021-06-05T17:20:00Z", "msg": "This is a message", "lvl": "info"})
+				encode_logfmt({"ts": "2021-06-05T17:20:00Z", "msg": "This is a message", "lvl": "info"})
 				"""
 			return: #"lvl=info msg="This is a message" ts=2021-06-05T17:20:00Z"#
 		},
@@ -43,7 +44,7 @@ remap: functions: encode_logfmt: {
 		{
 			title: "Encode to logfmt (nested fields)"
 			source: """
-				encode_logfmt!({"agent": {"name": "vector"}, "log": {"file": {"path": "my.log"}}, "event": "log"})
+				encode_logfmt({"agent": {"name": "vector"}, "log": {"file": {"path": "my.log"}}, "event": "log"})
 				"""
 			return: #"agent.name=vector event=log log.file.path=my.log"#
 		},

docs/reference/remap/functions/parse_logfmt.cue

@@ -9,11 +9,7 @@ remap: functions: parse_logfmt: {
 		* `"` characters can be escaped by `\`.
 		* As per this [logfmt specification](\#(urls.logfmt_specs)) standlone key are accepted and will be associated with the boolean value `true`.
 		"""#
-	notices: [
-		"""
-			All values are returned as strings, it is recommended to manually coerce values as you see fit.
-			""",
-	]
+	notices:     functions.encode_key_value.notices
 
 	arguments: [
 		{

lib/vrl/stdlib/Cargo.toml

@@ -55,6 +55,7 @@ default = [
     "downcase",
     "encode_base64",
     "encode_json",
+    "encode_key_value",
     "encode_logfmt",
     "encode_percent",
     "ends_with",
@@ -159,8 +160,9 @@ del = []
 downcase = []
 encode_base64 = ["base64"]
 encode_json = ["serde_json"]
-encode_logfmt = []
-encode_percent= ["percent-encoding"]
+encode_key_value = []
+encode_logfmt = ["encode_key_value"]
+encode_percent = ["percent-encoding"]
 ends_with = []
 exists = []
 flatten = []

lib/vrl/stdlib/benches/benches.rs

@@ -20,6 +20,7 @@ criterion_group!(
               //del,
               downcase,
               encode_base64,
+              encode_key_value,
               encode_json,
               encode_logfmt,
               encode_percent,
@@ -211,6 +212,59 @@ bench_function! {
     }
 }
 
+bench_function! {
+    encode_key_value => vrl_stdlib::EncodeKeyValue;
+
+    encode_complex_value {
+        args: func_args![value:
+            btreemap! {
+                "msg" => r#"result: {"authz": false, "length": 42}\n"#,
+                "severity" => "    panic"
+            },
+            key_value_delimiter: "==",
+            field_delimiter: "!!!"
+            ],
+        want: Ok(r#"msg=="result: {\"authz\": false, \"length\": 42}\\n"!!!severity=="    panic""#),
+    }
+
+    encode_key_value {
+        args: func_args![value:
+            btreemap! {
+                "mow" => "vvo",
+                "vvo" => "pkc",
+                "pkc" => "hrb",
+                "hrb" => "tsn",
+                "tsn" => "can",
+                "can" => "pnh",
+                "pnh" => "sin",
+                "sin" => "syd"
+            },
+            key_value_delimiter: ":",
+            field_delimiter: ","
+        ],
+        want: Ok(r#"can:pnh,hrb:tsn,mow:vvo,pkc:hrb,pnh:sin,sin:syd,tsn:can,vvo:pkc"#),
+    }
+
+    fields_ordering {
+        args: func_args![value:
+            btreemap! {
+                "mow" => "vvo",
+                "vvo" => "pkc",
+                "pkc" => "hrb",
+                "hrb" => "tsn",
+                "tsn" => "can",
+                "can" => "pnh",
+                "pnh" => "sin",
+                "sin" => "syd"
+            },
+            fields_ordering: value!(["mow", "vvo", "pkc", "hrb", "tsn", "can", "pnh", "sin"]),
+            key_value_delimiter: ":",
+            field_delimiter: ","
+        ],
+        want: Ok(r#"mow:vvo,vvo:pkc,pkc:hrb,hrb:tsn,tsn:can,can:pnh,pnh:sin,sin:syd"#),
+    }
+}
+
 bench_function! {
     encode_json => vrl_stdlib::EncodeJson;