[feat]: add attribute processor

go.mod

@@ -4,6 +4,7 @@ go 1.25.0
 
 require (
 	github.com/go-viper/mapstructure/v2 v2.5.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.146.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.146.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.146.0
 	github.com/stretchr/testify v1.11.1
@@ -13,6 +14,7 @@ require (
 	go.opentelemetry.io/collector/confmap/xconfmap v0.146.1
 	go.opentelemetry.io/collector/consumer v1.52.0
 	go.opentelemetry.io/collector/consumer/xconsumer v0.146.1
+	go.opentelemetry.io/collector/featuregate v1.52.0
 	go.opentelemetry.io/collector/pdata v1.52.0
 	go.opentelemetry.io/collector/pdata/pprofile v0.146.1
 	go.opentelemetry.io/collector/processor v1.52.0
@@ -63,7 +65,7 @@ require (
 	github.com/ua-parser/uap-go v0.0.0-20250326155420-f7f5a2f9f5bc // indirect
 	github.com/zeebo/xxh3 v1.1.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/collector/featuregate v1.52.0 // indirect
+	go.opentelemetry.io/collector/client v1.52.0 // indirect
 	go.opentelemetry.io/collector/internal/componentalias v0.146.1 // indirect
 	go.opentelemetry.io/collector/pipeline v1.52.0 // indirect
 	go.opentelemetry.io/collector/pipeline/xpipeline v0.146.1 // indirect

go.sum

@@ -92,6 +92,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.146.0
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.146.0/go.mod h1:gPJiIQKJOi01p4kVBRrXR9PyBvNxbfIL37943khkNNI=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.146.0 h1:4ECxx5C6LLZaLUfNApaw/bC/AxxBIewns2MvgalKxH0=
 github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.146.0/go.mod h1:AFhMSPTpRw83Pmlbym0tMJK3gsRmpXKbWuFNwRtBqjw=
+github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.146.0 h1:Vlo7twXttfAGCCHJOxhgRucm+QkkBftRoaWpts1alcQ=
+github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.146.0/go.mod h1:K9VapHTZooyV62uroDROqxcTVvmsqrnM6WnCbQKXl48=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.146.0 h1:Zdbp2A9Uy3ESgHJS+cNcuBhmaD4r/ehXMkEu6lvMShk=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.146.0/go.mod h1:SBOdXc9oeAUMBcrncgrecTFJzrpS7EjfOyUFnw8Kv6E=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.146.0 h1:1y1N+2CVbWTxd3ZEOe3XgzEusE4Ucm4CX407yHrXJ2s=
@@ -118,6 +120,8 @@ github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs=
 github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
+go.opentelemetry.io/collector/client v1.52.0 h1:m/hNA4feow0nvTKVOAno/YejrtW1aYbEST3uaz0USBk=
+go.opentelemetry.io/collector/client v1.52.0/go.mod h1:0FcZ0RZS4IFkhfzLyqQhKV3a/L1c/WwTQ3bHDILsQ1Q=
 go.opentelemetry.io/collector/component v1.52.0 h1:RYk1KTz8g+tU9mcYGz2gXJJDS8A9NJv2lta3JoWSZXg=
 go.opentelemetry.io/collector/component v1.52.0/go.mod h1:7ZgH6qsvUDSIk3JuZfxPv2qHeeUz3Y6znAWGdtp1r78=
 go.opentelemetry.io/collector/component/componentstatus v0.146.1 h1:91kcSsNFFQh6SjAf5tfGqW+pmOe5Sjppyo3ixpMzBK0=

internal/attributesprocessorexecutor.go

@@ -0,0 +1,86 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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 internal
+
+import (
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor"
+)
+
+var attributesProcessorConfigExamples = []ConfigExample{
+	{
+		Name:   "Insert Attribute for All Except Certain Metric Name",
+		Signal: "metrics",
+		Config: "attributes: \n" +
+			"  exclude:\n" +
+			"    match_type: strict\n" +
+			"    metric_names: [\"my.gauge\"]\n" +
+			"  actions:\n" +
+			"    - key: my.insert.attr\n" +
+			"      action: insert\n" +
+			"      value: \"some inserted value\"",
+	},
+	{
+		Name:   "Upsert Attribute for All Logs Above Certain Severity",
+		Signal: "logs",
+		Config: "attributes: \n" +
+			"  include:\n" +
+			"    log_severity_number:\n" +
+			"      min: 10\n" +
+			"      match_undefined: false\n" +
+			"  actions:\n" +
+			"    - key: my.upsert.attr\n" +
+			"      action: upsert\n" +
+			"      value: true",
+	},
+	{
+		Name:   "Delete Attributes Matching Regex",
+		Signal: "metrics",
+		Config: "attributes: \n" +
+			"  actions:\n" +
+			"    - pattern: '^my\\.[a-z]+\\.attr$'\n" +
+			"      action: delete",
+	},
+	{
+		Name:   "Convert Attribute Type",
+		Signal: "traces",
+		Config: "attributes: \n" +
+			"  actions:\n" +
+			"    - key: http.response.status_code\n" +
+			"      action: convert\n" +
+			"      converted_type: string",
+		Payload: `{"resourceSpans":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"my.service"}}]},"scopeSpans":[{"scope":{"name":"my.library","version":"1.0.0","attributes":[{"key":"my.scope.attribute","value":{"stringValue":"some scope attribute"}}]},"spans":[{"traceId":"5b8efff798038103d269b633813fc60c","spanId":"eee19b7ec3c1b174","parentSpanId":"eee19b7ec3c1b173","name":"I'm a server span","startTimeUnixNano":"1544712660000000000","endTimeUnixNano":"1544712661000000000","kind":2,"attributes":[{"key":"http.response.status_code","value":{"intValue":"500"}}],"status":{}},{"traceId":"5b8efff798038103d269b633813fc60c","spanId":"eee19b7ec3c1b173","parentSpanId":"eee19b7ec3c1b173","name":"Me too","startTimeUnixNano":"1544712660000000000","endTimeUnixNano":"1544712661000000000","kind":1,"attributes":[{"key":"http.response.status_code","value":{"intValue":"500"}}],"status":{}}]}]}]}`,
+	},
+}
+
+// NewAttributesProcessorExecutor creates an internal.Executor that runs OTTL statements using
+// the [attributesprocessor].
+func NewAttributesProcessorExecutor() Executor {
+	return NewJSONExecutor[attributesprocessor.Config](
+		newProcessorConsumer[attributesprocessor.Config](attributesprocessor.NewFactory()),
+		newMetadata(
+			ComponentTypeProcessor,
+			"attributes_processor",
+			"Attributes",
+			"github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor",
+			"https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/processor/attributesprocessor",
+			withConfigExamples(attributesProcessorConfigExamples...),
+		),
+	)
+}

internal/attributesprocessorexecutor_test.go

@@ -0,0 +1,103 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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 internal
+
+import (
+	"testing"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+const (
+	attributesprocessorConfig = "attributesprocessor.yaml"
+)
+
+func Test_AttributeProcessorExecutor_ParseConfig(t *testing.T) {
+	yamlConfig := readTestData(t, attributesprocessorConfig)
+	cfgs, err := parseConfig[attributesprocessor.Config](
+		component.NewIDWithName(attributesprocessor.NewFactory().Type(), "test_attributes_processor"),
+		yamlConfig,
+		func() *attributesprocessor.Config {
+			return attributesprocessor.NewFactory().CreateDefaultConfig().(*attributesprocessor.Config)
+		},
+	)
+	require.NoError(t, err)
+
+	pc := cfgs[0].Value
+	require.NotNil(t, pc)
+	require.NotEmpty(t, pc.Actions)
+}
+
+func Test_AttributeProcessorExecutor_ExecuteLogs(t *testing.T) {
+	executor := NewAttributesProcessorExecutor()
+	config := readTestData(t, attributesprocessorConfig)
+	payload := readTestData(t, "logs.json")
+
+	output, err := executor.ExecuteLogs(config, payload)
+	require.NoError(t, err)
+
+	unmarshaler := &plog.JSONUnmarshaler{}
+	outputLogs, err := unmarshaler.UnmarshalLogs([]byte(output.Value))
+	require.NoError(t, err)
+	require.NotNil(t, outputLogs)
+}
+
+func Test_AttributeProcessorExecutor_ExecuteTraces(t *testing.T) {
+	executor := NewAttributesProcessorExecutor()
+	config := readTestData(t, attributesprocessorConfig)
+	payload := readTestData(t, "traces.json")
+
+	output, err := executor.ExecuteTraces(config, payload)
+	require.NoError(t, err)
+
+	unmarshaler := &ptrace.JSONUnmarshaler{}
+	outputTraces, err := unmarshaler.UnmarshalTraces([]byte(output.Value))
+	require.NoError(t, err)
+	require.NotNil(t, outputTraces)
+}
+
+func Test_AttributeProcessorExecutor_ExecuteMetrics(t *testing.T) {
+	executor := NewAttributesProcessorExecutor()
+	config := readTestData(t, attributesprocessorConfig)
+	payload := readTestData(t, "metrics.json")
+
+	output, err := executor.ExecuteMetrics(config, payload)
+	require.NoError(t, err)
+
+	unmarshaler := &pmetric.JSONUnmarshaler{}
+	outputMetrics, err := unmarshaler.UnmarshalMetrics([]byte(output.Value))
+	require.NoError(t, err)
+	require.NotNil(t, outputMetrics)
+}
+
+func Test_AttributeProcessorExecutor_ObservedLogs(t *testing.T) {
+	executor := NewAttributesProcessorExecutor().(*defaultExecutor[attributesprocessor.Config])
+	executor.consumer.TelemetrySettings().Logger.Sugar().Debug("this is a log")
+	logEntries := executor.ObservedLogs().TakeAll()
+
+	assert.Len(t, logEntries, 1)
+	assert.Contains(t, logEntries[0].ConsoleEncodedEntry(), "this is a log")
+}

internal/executors.go

@@ -24,5 +24,6 @@ func Executors() []Executor {
 	return []Executor{
 		NewTransformProcessorExecutor(),
 		NewFilterProcessorExecutor(),
+		NewAttributesProcessorExecutor(),
 	}
 }

testdata/attributesprocessor.yaml

@@ -0,0 +1,17 @@
+actions:
+  - key: my.gauge.attr
+    action: delete
+  - key: upsert.attr
+    value: true
+    action: upsert
+  - key: insert.attr
+    value: 2245
+    action: insert
+  - key: account_email
+    action: hash
+  - key: int.attribute
+    action: convert
+    converted_type: int
+  - key: my.span.attr
+    from_attribute: name
+    action: update