[processor/transform] Add merge_maps and ParseJSON functions (#16551)

* Add merge_maps and ParseJSON

* add changelog

.chloggen/tp-add-new-funcs.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: transformprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add the `merge_maps` and `ParseJSON` functions.
+
+# One or more tracking issues related to the change
+issues: [16551]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

processor/transformprocessor/internal/common/functions.go

@@ -29,6 +29,7 @@ func Functions[K any]() map[string]interface{} {
 		"Split":                ottlfuncs.Split[K],
 		"Int":                  ottlfuncs.Int[K],
 		"ConvertCase":          ottlfuncs.ConvertCase[K],
+		"ParseJSON":            ottlfuncs.ParseJSON[K],
 		"keep_keys":            ottlfuncs.KeepKeys[K],
 		"set":                  ottlfuncs.Set[K],
 		"truncate_all":         ottlfuncs.TruncateAll[K],
@@ -39,6 +40,7 @@ func Functions[K any]() map[string]interface{} {
 		"replace_all_patterns": ottlfuncs.ReplaceAllPatterns[K],
 		"delete_key":           ottlfuncs.DeleteKey[K],
 		"delete_matching_keys": ottlfuncs.DeleteMatchingKeys[K],
+		"merge_maps":           ottlfuncs.MergeMaps[K],
 	}
 }
 

processor/transformprocessor/internal/logs/processor_test.go

@@ -298,6 +298,12 @@ func Test_ProcessLogs_LogContext(t *testing.T) {
 				td.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Attributes().PutStr("test", "OperationA")
 			},
 		},
+		{
+			statement: `merge_maps(attributes, ParseJSON("{\"json_test\":\"pass\"}"), "insert") where body == "operationA"`,
+			want: func(td plog.Logs) {
+				td.ResourceLogs().At(0).ScopeLogs().At(0).LogRecords().At(0).Attributes().PutStr("json_test", "pass")
+			},
+		},
 	}
 
 	for _, tt := range tests {

processor/transformprocessor/internal/metrics/processor_test.go

@@ -469,6 +469,13 @@ func Test_ProcessMetrics_DataPointContext(t *testing.T) {
 				v01.AppendEmpty().SetStr("C")
 			},
 		},
+		{
+			statements: []string{`merge_maps(attributes, ParseJSON("{\"json_test\":\"pass\"}"), "insert") where metric.name == "operationA"`},
+			want: func(td pmetric.Metrics) {
+				td.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().At(0).Attributes().PutStr("json_test", "pass")
+				td.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().At(1).Attributes().PutStr("json_test", "pass")
+			},
+		},
 	}
 
 	for _, tt := range tests {

processor/transformprocessor/internal/traces/processor_test.go

@@ -326,22 +326,31 @@ func Test_ProcessTraces_TraceContext(t *testing.T) {
 			want: func(td ptrace.Traces) {
 				td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes().PutStr("test", "operationa")
 			},
-		}, {
+		},
+		{
 			statement: `set(attributes["test"], ConvertCase(name, "upper")) where name == "operationA"`,
 			want: func(td ptrace.Traces) {
 				td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes().PutStr("test", "OPERATIONA")
 			},
-		}, {
+		},
+		{
 			statement: `set(attributes["test"], ConvertCase(name, "snake")) where name == "operationA"`,
 			want: func(td ptrace.Traces) {
 				td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes().PutStr("test", "operation_a")
 			},
-		}, {
+		},
+		{
 			statement: `set(attributes["test"], ConvertCase(name, "camel")) where name == "operationA"`,
 			want: func(td ptrace.Traces) {
 				td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes().PutStr("test", "OperationA")
 			},
 		},
+		{
+			statement: `merge_maps(attributes, ParseJSON("{\"json_test\":\"pass\"}"), "insert") where name == "operationA"`,
+			want: func(td ptrace.Traces) {
+				td.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes().PutStr("json_test", "pass")
+			},
+		},
 	}
 
 	for _, tt := range tests {