[pkg/ottl] Add merge_maps function

.chloggen/ottl-merge-maps.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: pkg/ottl
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add new `merge` function to OTTL, which allows merging maps.
+
+# One or more tracking issues related to the change
+issues: [16461]
+
+# (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:

pkg/ottl/ottlfuncs/README.md

@@ -16,6 +16,7 @@ List of available Functions:
 - [delete_matching_keys](#delete_matching_keys)
 - [keep_keys](#keep_keys)
 - [limit](#limit)
+- [merge_maps](#merge_maps)
 - [replace_all_matches](#replace_all_matches)
 - [replace_all_patterns](#replace_all_patterns)
 - [replace_match](#replace_match)
@@ -238,6 +239,31 @@ Examples:
 
 - `limit(resource.attributes, 50, ["http.host", "http.method"])`
 
+### merge_maps
+
+`merge_maps(target, source, strategy)`
+
+The `merge_maps` function merges the source map into the target map using the supplied strategy to handle conflicts.
+
+`target` is a `pdata.Map` type field. `source` is a `pdata.Map` type field. `strategy` is a string that must be one of `insert`, `update`, or `upsert`.
+
+If strategy is:
+- `insert`: Insert the value from `source` into `target` where the key does not already exist.
+- `update`: Update the entry in `target` with the value from `source` where the key does exist.
+- `upsert`: Performs insert or update. Insert the value from `source` into `target` where the key does not already exist and update the entry in `target` with the value from `source` where the key does exist.
+
+`merge_maps` is a special case of the [`set` function](#set). If you need to completely override `target`, use `set` instead.
+
+Examples:
+
+- `merge_maps(attributes, ParseJSON(body), "upsert")`
+
+
+- `merge_maps(attributes, ParseJSON(attributes["kubernetes"]), "update")`
+
+
+- `merge_maps(attributes, resource.attributes, "insert")`
+
 ### replace_all_matches
 
 `replace_all_matches(target, pattern, replacement)`

pkg/ottl/ottlfuncs/func_merge_maps.go

@@ -0,0 +1,85 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ottlfuncs // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs"
+
+import (
+	"context"
+	"fmt"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+)
+
+const (
+	INSERT = "insert"
+	UPDATE = "update"
+	UPSERT = "upsert"
+)
+
+// MergeMaps function merges the source map into the target map using the supplied strategy to handle conflicts.
+// Strategy definitions:
+//
+//	insert: Insert the value from `source` into `target` where the key does not already exist.
+//	update: Update the entry in `target` with the value from `source` where the key does exist
+//	upsert: Performs insert or update. Insert the value from `source` into `target` where the key does not already exist and update the entry in `target` with the value from `source` where the key does exist.
+func MergeMaps[K any](target ottl.Getter[K], source ottl.Getter[K], strategy string) (ottl.ExprFunc[K], error) {
+	if strategy != INSERT && strategy != UPDATE && strategy != UPSERT {
+		return nil, fmt.Errorf("invalid value for strategy, %v, must be 'insert', 'update' or 'upsert'", strategy)
+	}
+
+	return func(ctx context.Context, tCtx K) (interface{}, error) {
+		targetVal, err := target.Get(ctx, tCtx)
+		if err != nil {
+			return nil, err
+		}
+
+		if targetMap, ok := targetVal.(pcommon.Map); ok {
+			val, err := source.Get(ctx, tCtx)
+			if err != nil {
+				return nil, err
+			}
+
+			if valueMap, ok := val.(pcommon.Map); ok {
+				switch strategy {
+				case INSERT:
+					valueMap.Range(func(k string, v pcommon.Value) bool {
+						if _, ok := targetMap.Get(k); !ok {
+							tv := targetMap.PutEmpty(k)
+							v.CopyTo(tv)
+						}
+						return true
+					})
+				case UPDATE:
+					valueMap.Range(func(k string, v pcommon.Value) bool {
+						if tv, ok := targetMap.Get(k); ok {
+							v.CopyTo(tv)
+						}
+						return true
+					})
+				case UPSERT:
+					valueMap.Range(func(k string, v pcommon.Value) bool {
+						tv := targetMap.PutEmpty(k)
+						v.CopyTo(tv)
+						return true
+					})
+				default:
+					return nil, fmt.Errorf("unknown strategy, %v", strategy)
+				}
+			}
+		}
+		return nil, nil
+	}, nil
+}

pkg/ottl/ottlfuncs/func_merge_maps_test.go

@@ -0,0 +1,165 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ottlfuncs
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+)
+
+func Test_MergeMaps(t *testing.T) {
+
+	input := pcommon.NewMap()
+	input.PutStr("attr1", "value1")
+
+	targetGetter := &ottl.StandardGetSetter[pcommon.Map]{
+		Getter: func(ctx context.Context, tCtx pcommon.Map) (interface{}, error) {
+			return tCtx, nil
+		},
+	}
+
+	tests := []struct {
+		name     string
+		source   ottl.Getter[pcommon.Map]
+		strategy string
+		want     func(pcommon.Map)
+	}{
+		{
+			name: "Upsert no conflicting keys",
+			source: ottl.StandardGetSetter[pcommon.Map]{
+				Getter: func(ctx context.Context, _ pcommon.Map) (interface{}, error) {
+					m := pcommon.NewMap()
+					m.PutStr("attr2", "value2")
+					return m, nil
+				},
+			},
+			strategy: UPSERT,
+			want: func(expectedValue pcommon.Map) {
+				expectedValue.PutStr("attr1", "value1")
+				expectedValue.PutStr("attr2", "value2")
+			},
+		},
+		{
+			name: "Upsert conflicting key",
+			source: ottl.StandardGetSetter[pcommon.Map]{
+				Getter: func(ctx context.Context, _ pcommon.Map) (interface{}, error) {
+					m := pcommon.NewMap()
+					m.PutStr("attr1", "value3")
+					m.PutStr("attr2", "value2")
+					return m, nil
+				},
+			},
+			strategy: UPSERT,
+			want: func(expectedValue pcommon.Map) {
+				expectedValue.PutStr("attr1", "value3")
+				expectedValue.PutStr("attr2", "value2")
+			},
+		},
+		{
+			name: "Insert no conflicting keys",
+			source: ottl.StandardGetSetter[pcommon.Map]{
+				Getter: func(ctx context.Context, _ pcommon.Map) (interface{}, error) {
+					m := pcommon.NewMap()
+					m.PutStr("attr2", "value2")
+					return m, nil
+				},
+			},
+			strategy: INSERT,
+			want: func(expectedValue pcommon.Map) {
+				expectedValue.PutStr("attr1", "value1")
+				expectedValue.PutStr("attr2", "value2")
+			},
+		},
+		{
+			name: "Insert conflicting key",
+			source: ottl.StandardGetSetter[pcommon.Map]{
+				Getter: func(ctx context.Context, _ pcommon.Map) (interface{}, error) {
+					m := pcommon.NewMap()
+					m.PutStr("attr1", "value3")
+					m.PutStr("attr2", "value2")
+					return m, nil
+				},
+			},
+			strategy: INSERT,
+			want: func(expectedValue pcommon.Map) {
+				expectedValue.PutStr("attr1", "value1")
+				expectedValue.PutStr("attr2", "value2")
+			},
+		},
+		{
+			name: "Update no conflicting keys",
+			source: ottl.StandardGetSetter[pcommon.Map]{
+				Getter: func(ctx context.Context, _ pcommon.Map) (interface{}, error) {
+					m := pcommon.NewMap()
+					m.PutStr("attr2", "value2")
+					return m, nil
+				},
+			},
+			strategy: UPDATE,
+			want: func(expectedValue pcommon.Map) {
+				expectedValue.PutStr("attr1", "value1")
+			},
+		},
+		{
+			name: "Update conflicting key",
+			source: ottl.StandardGetSetter[pcommon.Map]{
+				Getter: func(ctx context.Context, _ pcommon.Map) (interface{}, error) {
+					m := pcommon.NewMap()
+					m.PutStr("attr1", "value3")
+					return m, nil
+				},
+			},
+			strategy: UPDATE,
+			want: func(expectedValue pcommon.Map) {
+				expectedValue.PutStr("attr1", "value3")
+			},
+		},
+		{
+			name: "non-map value leaves target unchanged",
+			source: ottl.StandardGetSetter[pcommon.Map]{
+				Getter: func(ctx context.Context, _ pcommon.Map) (interface{}, error) {
+					return nil, nil
+				},
+			},
+			strategy: UPSERT,
+			want: func(expectedValue pcommon.Map) {
+				expectedValue.PutStr("attr1", "value1")
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			scenarioMap := pcommon.NewMap()
+			input.CopyTo(scenarioMap)
+
+			exprFunc, err := MergeMaps[pcommon.Map](targetGetter, tt.source, tt.strategy)
+			assert.NoError(t, err)
+
+			result, err := exprFunc(context.Background(), scenarioMap)
+			assert.NoError(t, err)
+			assert.Nil(t, result)
+
+			expected := pcommon.NewMap()
+			tt.want(expected)
+
+			assert.Equal(t, expected, scenarioMap)
+		})
+	}
+}