Schema Processor Revamp implementation

processor/schemaprocessor/go.mod

@@ -1,15 +1,17 @@
 module github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor
 
-go 1.21.0
+go 1.22.0
 
 require (
+	github.com/google/go-cmp v0.6.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.108.0
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/component v0.106.1
 	go.opentelemetry.io/collector/config/confighttp v0.106.1
 	go.opentelemetry.io/collector/confmap v0.106.1
 	go.opentelemetry.io/collector/consumer v0.106.1
 	go.opentelemetry.io/collector/consumer/consumertest v0.106.1
-	go.opentelemetry.io/collector/pdata v1.12.0
+	go.opentelemetry.io/collector/pdata v1.14.1
 	go.opentelemetry.io/collector/processor v0.106.1
 	go.opentelemetry.io/otel/metric v1.28.0
 	go.opentelemetry.io/otel/schema v0.0.8
@@ -20,6 +22,7 @@ require (
 )
 
 require (
+	github.com/Masterminds/semver/v3 v3.2.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -42,6 +45,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.108.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.19.1 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect

processor/schemaprocessor/internal/alias/alias.go

@@ -1,7 +1,7 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-// Package Alias is a subset of the interfaces defined by pdata and family
+// Package alias is a subset of the interfaces defined by pdata and family
 // package to allow for higher code reuse without using generics.
 package alias // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/alias"
 
@@ -30,6 +30,10 @@ type NamedSignal interface {
 	SetName(name string)
 }
 
+type Attributed interface {
+	Attributes() pcommon.Map
+}
+
 var (
 	_ Resource = (*plog.ResourceLogs)(nil)
 	_ Resource = (*pmetric.ResourceMetrics)(nil)

processor/schemaprocessor/internal/changelist/changelist.go

@@ -0,0 +1,75 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package changelist // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/changelist"
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/alias"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/migrate"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/operator"
+)
+
+// ChangeList represents a list of changes within a section of the schema processor.  It can take in a list of different migrators for a specific section and will apply them in order, based on whether Apply or Rollback is called
+type ChangeList struct {
+	Migrators []migrate.Migrator
+}
+
+func (c ChangeList) Do(ss migrate.StateSelector, signal any) error {
+	for i := 0; i < len(c.Migrators); i++ {
+		var migrator migrate.Migrator
+		// todo(ankit) in go1.23 switch to reversed iterators for this
+		if ss == migrate.StateSelectorApply {
+			migrator = c.Migrators[i]
+		} else {
+			migrator = c.Migrators[len(c.Migrators)-1-i]
+		}
+		// switch between operator types - what do the operators act on?
+		switch thisMigrator := migrator.(type) {
+		// this one acts on both spans and span events!
+		case operator.SpanOperator:
+			if span, ok := signal.(ptrace.Span); ok {
+				if err := thisMigrator.Do(ss, span); err != nil {
+					return err
+				}
+			} else {
+				return fmt.Errorf("SpanOperator %T can't act on %T", thisMigrator, signal)
+			}
+		case operator.MetricOperator:
+			if metric, ok := signal.(pmetric.Metric); ok {
+				if err := thisMigrator.Do(ss, metric); err != nil {
+					return err
+				}
+			} else {
+				return fmt.Errorf("MetricOperator %T can't act on %T", thisMigrator, signal)
+			}
+		// no log operator because the only log operation is an attribute changeset
+		// this block is for the `all` block, the `resource` block, and the `log` block
+		// todo(ankit) switch these to specific typed ones?
+		case migrate.AttributeChangeSet:
+			switch attributeSignal := signal.(type) {
+			case alias.Attributed:
+				if err := thisMigrator.Do(ss, attributeSignal.Attributes()); err != nil {
+					return err
+				}
+			default:
+				return fmt.Errorf("unsupported signal type %T for AttributeChangeSet", attributeSignal)
+			}
+		default:
+			return fmt.Errorf("unsupported migrator type %T", thisMigrator)
+		}
+	}
+	return nil
+}
+
+func (c ChangeList) Apply(signal any) error {
+	return c.Do(migrate.StateSelectorApply, signal)
+}
+
+func (c ChangeList) Rollback(signal any) error {
+	return c.Do(migrate.StateSelectorRollback, signal)
+}

processor/schemaprocessor/internal/migrate/attributes.go

@@ -11,24 +11,21 @@ import (
 	"go.uber.org/multierr"
 )
 
-// AttributeChangeSet represents an unscoped entry that can be applied.
-//
+// AttributeChangeSet represents a rename_attributes type operation.
 // The listed changes are duplicated twice
 // to allow for simplified means of transition to or from a revision.
 type AttributeChangeSet struct {
-	updates  ast.AttributeMap
+	// The keys are the old attribute name used in the previous version, the values are the
+	// new attribute name starting from this version (comment from ast.AttributeMap)
+	updates ast.AttributeMap
+	// the inverse of the updates map
 	rollback ast.AttributeMap
 }
 
-// AttributeChangeSetSlice allows for `AttributeChangeSet`
-// to be chained together as they are defined within the schema
-// and be applied sequentially to ensure deterministic behavior.
-type AttributeChangeSetSlice []*AttributeChangeSet
-
 // NewAttributeChangeSet allows for typed strings to be used as part
 // of the invocation that will be converted into the default string type.
-func NewAttributeChangeSet(mappings ast.AttributeMap) *AttributeChangeSet {
-	attr := &AttributeChangeSet{
+func NewAttributeChangeSet(mappings ast.AttributeMap) AttributeChangeSet {
+	attr := AttributeChangeSet{
 		updates:  make(map[string]string, len(mappings)),
 		rollback: make(map[string]string, len(mappings)),
 	}
@@ -39,15 +36,17 @@ func NewAttributeChangeSet(mappings ast.AttributeMap) *AttributeChangeSet {
 	return attr
 }
 
+func (a AttributeChangeSet) IsMigrator() {}
+
 func (a *AttributeChangeSet) Apply(attrs pcommon.Map) error {
-	return a.do(StateSelectorApply, attrs)
+	return a.Do(StateSelectorApply, attrs)
 }
 
 func (a *AttributeChangeSet) Rollback(attrs pcommon.Map) error {
-	return a.do(StateSelectorRollback, attrs)
+	return a.Do(StateSelectorRollback, attrs)
 }
 
-func (a *AttributeChangeSet) do(ss StateSelector, attrs pcommon.Map) (errs error) {
+func (a *AttributeChangeSet) Do(ss StateSelector, attrs pcommon.Map) (errs error) {
 	var (
 		updated = make(map[string]struct{})
 		results = pcommon.NewMap()
@@ -81,33 +80,3 @@ func (a *AttributeChangeSet) do(ss StateSelector, attrs pcommon.Map) (errs error
 	results.CopyTo(attrs)
 	return errs
 }
-
-// NewAttributeChangeSetSlice combines all the provided `AttributeChangeSets`
-// and allows them to be executed in the provided order.
-func NewAttributeChangeSetSlice(changes ...*AttributeChangeSet) *AttributeChangeSetSlice {
-	values := new(AttributeChangeSetSlice)
-	for _, c := range changes {
-		(*values) = append((*values), c)
-	}
-	return values
-}
-
-func (slice *AttributeChangeSetSlice) Apply(attrs pcommon.Map) error {
-	return slice.do(StateSelectorApply, attrs)
-}
-
-func (slice *AttributeChangeSetSlice) Rollback(attrs pcommon.Map) error {
-	return slice.do(StateSelectorRollback, attrs)
-}
-
-func (slice *AttributeChangeSetSlice) do(ss StateSelector, attrs pcommon.Map) (errs error) {
-	for i := 0; i < len(*slice); i++ {
-		switch ss {
-		case StateSelectorApply:
-			errs = multierr.Append(errs, (*slice)[i].Apply(attrs))
-		case StateSelectorRollback:
-			errs = multierr.Append(errs, (*slice)[len(*slice)-1-i].Rollback(attrs))
-		}
-	}
-	return errs
-}

processor/schemaprocessor/internal/migrate/attributes_test.go

@@ -27,7 +27,7 @@ func TestNewAttributeChangeSet(t *testing.T) {
 			"hello": "world",
 		})
 
-		expect := &AttributeChangeSet{
+		expect := AttributeChangeSet{
 			updates: map[string]string{
 				"hello": "world",
 			},
@@ -45,7 +45,7 @@ func TestAttributeChangeSetApply(t *testing.T) {
 
 	for _, tc := range []struct {
 		name   string
-		acs    *AttributeChangeSet
+		acs    AttributeChangeSet
 		attrs  pcommon.Map
 		expect pcommon.Map
 		errVal string
@@ -120,7 +120,7 @@ func TestAttributeChangeSetRollback(t *testing.T) {
 
 	for _, tc := range []struct {
 		name   string
-		acs    *AttributeChangeSet
+		acs    AttributeChangeSet
 		attrs  pcommon.Map
 		expect pcommon.Map
 		errVal string
@@ -189,98 +189,3 @@ func TestAttributeChangeSetRollback(t *testing.T) {
 		})
 	}
 }
-
-func TestNewAttributeChangeSetSliceApply(t *testing.T) {
-	t.Parallel()
-
-	for _, tc := range []struct {
-		name    string
-		changes *AttributeChangeSetSlice
-		attr    pcommon.Map
-		expect  pcommon.Map
-	}{
-		{
-			name:    "no changes listed",
-			changes: NewAttributeChangeSetSlice(),
-			attr: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("service.version", "v0.0.1")
-			}),
-			expect: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("service.version", "v0.0.1")
-			}),
-		},
-		{
-			name: "changes defined",
-			changes: NewAttributeChangeSetSlice(
-				NewAttributeChangeSet(map[string]string{
-					"service_version": "service.version",
-				}),
-				NewAttributeChangeSet(map[string]string{
-					"service.version": "application.service.version",
-				}),
-			),
-			attr: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("service_version", "v0.0.1")
-			}),
-			expect: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("application.service.version", "v0.0.1")
-			}),
-		},
-	} {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			assert.NoError(t, tc.changes.Apply(tc.attr))
-			assert.Equal(t, tc.expect.AsRaw(), tc.attr.AsRaw(), "Must match the expected attributes")
-		})
-	}
-}
-
-func TestNewAttributeChangeSetSliceApplyRollback(t *testing.T) {
-	t.Parallel()
-
-	for _, tc := range []struct {
-		name    string
-		changes *AttributeChangeSetSlice
-		attr    pcommon.Map
-		expect  pcommon.Map
-	}{
-		{
-			name:    "no changes listed",
-			changes: NewAttributeChangeSetSlice(),
-			attr: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("service.version", "v0.0.1")
-			}),
-			expect: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("service.version", "v0.0.1")
-			}),
-		},
-		{
-			name: "changes defined",
-			changes: NewAttributeChangeSetSlice(
-				NewAttributeChangeSet(map[string]string{
-					"service_version": "service.version",
-				}),
-				NewAttributeChangeSet(map[string]string{
-					"service.version": "application.service.version",
-				}),
-			),
-			attr: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("application.service.version", "v0.0.1")
-
-			}),
-			expect: testHelperBuildMap(func(m pcommon.Map) {
-				m.PutStr("service_version", "v0.0.1")
-			}),
-		},
-	} {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			assert.NoError(t, tc.changes.Rollback(tc.attr))
-			assert.Equal(t, tc.expect.AsRaw(), tc.attr.AsRaw(), "Must match the expected attributes")
-		})
-	}
-}