Include changelist and revision parts

ankitpatel96 · ankitpatel96 · commit 47ebae236321 · 2024-09-30T14:45:40.000-04:00
diff --git a/processor/schemaprocessor/internal/changelist/changelist.go b/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)
+}
diff --git a/processor/schemaprocessor/internal/translation/revision_v1.go b/processor/schemaprocessor/internal/translation/revision_v1.go
@@ -6,109 +6,141 @@ package translation // import "github.com/open-telemetry/opentelemetry-collector
 import (
 	"go.opentelemetry.io/otel/schema/v1.0/ast"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/changelist"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/migrate"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/operator"
 )
 
 // RevisionV1 represents all changes that are to be
 // applied to a signal at a given version.
+// todo(ankit) implement split and rest of otel schema
 type RevisionV1 struct {
-	ver              *Version
-	all              *migrate.AttributeChangeSetSlice
-	resource         *migrate.AttributeChangeSetSlice
-	spans            *migrate.ConditionalAttributeSetSlice
-	eventNames       *migrate.SignalNameChangeSlice
-	eventAttrsOnSpan *migrate.ConditionalAttributeSetSlice
-	eventAttrsOnName *migrate.ConditionalAttributeSetSlice
-	metricsAttrs     *migrate.ConditionalAttributeSetSlice
-	metricNames      *migrate.SignalNameChangeSlice
+	ver        *Version
+	all        *changelist.ChangeList
+	resources  *changelist.ChangeList
+	spans      *changelist.ChangeList
+	spanEvents *changelist.ChangeList
+	metrics    *changelist.ChangeList
+	logs       *changelist.ChangeList
 }
 
 // NewRevision processes the VersionDef and assigns the version to this revision
 // to allow sorting within a slice.
 // Since VersionDef uses custom types for various definitions, it isn't possible
 // to cast those values into the primitives so each has to be processed together.
 // Generics would be handy here.
+// todo(ankit) investigate using generics
 func NewRevision(ver *Version, def ast.VersionDef) *RevisionV1 {
+	// todo(ankit) change logs to be an ast.Attributes type so I dont have to change this
+	var logChanges ast.Attributes
+	for _, change := range def.Logs.Changes {
+		//nolint:gosimple
+		logChanges.Changes = append(logChanges.Changes, ast.AttributeChange{RenameAttributes: change.RenameAttributes})
+
+	}
 	return &RevisionV1{
-		ver:              ver,
-		all:              newAttributeChangeSetSliceFromChanges(def.All),
-		resource:         newAttributeChangeSetSliceFromChanges(def.Resources),
-		spans:            newSpanConditionalAttributeSlice(def.Spans),
-		eventNames:       newSpanEventSignalSlice(def.SpanEvents),
-		eventAttrsOnSpan: newSpanEventConditionalSpans(def.SpanEvents),
-		eventAttrsOnName: newSpanEventConditionalNames(def.SpanEvents),
-		metricsAttrs:     newMetricConditionalSlice(def.Metrics),
-		metricNames:      newMetricNameSignalSlice(def.Metrics),
+		ver:        ver,
+		all:        newAllChangeList(def.All),
+		resources:  newResourceChangeList(def.Resources),
+		spans:      newSpanChangeList(def.Spans),
+		spanEvents: newSpanEventChangeList(def.SpanEvents),
+		metrics:    newMetricChangeList(def.Metrics),
+		logs:       newLogsChangelist(def.Logs),
 	}
+
 }
 
-func newAttributeChangeSetSliceFromChanges(attrs ast.Attributes) *migrate.AttributeChangeSetSlice {
-	values := make([]*migrate.AttributeChangeSet, 0, 10)
-	for _, at := range attrs.Changes {
-		if renamed := at.RenameAttributes; renamed != nil {
-			values = append(values, migrate.NewAttributeChangeSet(renamed.AttributeMap))
-		}
-	}
-	return migrate.NewAttributeChangeSetSlice(values...)
+func (r RevisionV1) Version() *Version {
+	return r.ver
 }
 
-func newSpanConditionalAttributeSlice(spans ast.Spans) *migrate.ConditionalAttributeSetSlice {
-	values := make([]*migrate.ConditionalAttributeSet, 0, 10)
-	for _, ch := range spans.Changes {
-		if renamed := ch.RenameAttributes; renamed != nil {
-			values = append(values, migrate.NewConditionalAttributeSet(
-				renamed.AttributeMap,
-				renamed.ApplyToSpans...,
-			))
+func newAllChangeList(all ast.Attributes) *changelist.ChangeList {
+	values := make([]migrate.Migrator, 0)
+	for _, at := range all.Changes {
+		if renamed := at.RenameAttributes; renamed != nil {
+			attributeChangeSet := migrate.NewAttributeChangeSet(renamed.AttributeMap)
+			values = append(values, attributeChangeSet)
 		}
 	}
-	return migrate.NewConditionalAttributeSetSlice(values...)
+	return &changelist.ChangeList{Migrators: values}
 }
 
-func newSpanEventSignalSlice(events ast.SpanEvents) *migrate.SignalNameChangeSlice {
-	values := make([]*migrate.SignalNameChange, 0, 10)
-	for _, ch := range events.Changes {
-		if renamed := ch.RenameEvents; renamed != nil {
-			values = append(values, migrate.NewSignalNameChange(renamed.EventNameMap))
+func newResourceChangeList(resource ast.Attributes) *changelist.ChangeList {
+	values := make([]migrate.Migrator, 0)
+	for _, at := range resource.Changes {
+		if renamed := at.RenameAttributes; renamed != nil {
+			attributeChangeSet := migrate.NewAttributeChangeSet(renamed.AttributeMap)
+			values = append(values, attributeChangeSet)
 		}
 	}
-	return migrate.NewSignalNameChangeSlice(values...)
+	return &changelist.ChangeList{Migrators: values}
 }
 
-func newSpanEventConditionalSpans(events ast.SpanEvents) *migrate.ConditionalAttributeSetSlice {
-	values := make([]*migrate.ConditionalAttributeSet, 0, 10)
-	for _, ch := range events.Changes {
-		if rename := ch.RenameAttributes; rename != nil {
-			values = append(values, migrate.NewConditionalAttributeSet(rename.AttributeMap, rename.ApplyToSpans...))
+func newSpanChangeList(spans ast.Spans) *changelist.ChangeList {
+	values := make([]migrate.Migrator, 0)
+	for _, at := range spans.Changes {
+		if renamed := at.RenameAttributes; renamed != nil {
+			conditionalAttributeChangeSet := operator.SpanConditionalAttributeOperator{Migrator: migrate.NewConditionalAttributeSet(renamed.AttributeMap, renamed.ApplyToSpans...)}
+			values = append(values, conditionalAttributeChangeSet)
 		}
 	}
-	return migrate.NewConditionalAttributeSetSlice(values...)
+	return &changelist.ChangeList{Migrators: values}
+
 }
 
-func newSpanEventConditionalNames(events ast.SpanEvents) *migrate.ConditionalAttributeSetSlice {
-	values := make([]*migrate.ConditionalAttributeSet, 0, 10)
-	for _, ch := range events.Changes {
-		if rename := ch.RenameAttributes; rename != nil {
-			values = append(values, migrate.NewConditionalAttributeSet(rename.AttributeMap, rename.ApplyToEvents...))
+func newMetricChangeList(metrics ast.Metrics) *changelist.ChangeList {
+	values := make([]migrate.Migrator, 0)
+	for _, at := range metrics.Changes {
+		if renameAttributes := at.RenameAttributes; renameAttributes != nil {
+			attributeChangeSet := operator.MetricDataPointAttributeOperator{
+				ConditionalAttributeChange: migrate.NewConditionalAttributeSet(renameAttributes.AttributeMap, renameAttributes.ApplyToMetrics...),
+			}
+			values = append(values, attributeChangeSet)
+		} else if renamedMetrics := at.RenameMetrics; renamedMetrics != nil {
+			signalNameChange := operator.MetricSignalNameChange{SignalNameChange: migrate.NewSignalNameChange(renamedMetrics)}
+			values = append(values, signalNameChange)
 		}
 	}
-	return migrate.NewConditionalAttributeSetSlice(values...)
+	return &changelist.ChangeList{Migrators: values}
 }
 
-func newMetricConditionalSlice(metrics ast.Metrics) *migrate.ConditionalAttributeSetSlice {
-	values := make([]*migrate.ConditionalAttributeSet, 0, 10)
-	for _, ch := range metrics.Changes {
-		if rename := ch.RenameAttributes; rename != nil {
-			values = append(values, migrate.NewConditionalAttributeSet(rename.AttributeMap, rename.ApplyToMetrics...))
+func newSpanEventChangeList(spanEvents ast.SpanEvents) *changelist.ChangeList {
+	values := make([]migrate.Migrator, 0)
+	for _, at := range spanEvents.Changes {
+		if renamedEvent := at.RenameEvents; renamedEvent != nil {
+			signalNameChange := migrate.NewSignalNameChange(renamedEvent.EventNameMap)
+			spanEventSignalNameChange := operator.SpanEventSignalNameChange{SignalNameChange: signalNameChange}
+			values = append(values, spanEventSignalNameChange)
+		} else if renamedAttribute := at.RenameAttributes; renamedAttribute != nil {
+			acceptableSpanNames := make([]string, 0)
+			for _, spanName := range renamedAttribute.ApplyToSpans {
+				acceptableSpanNames = append(acceptableSpanNames, string(spanName))
+			}
+			acceptableEventNames := make([]string, 0)
+			for _, eventName := range renamedAttribute.ApplyToEvents {
+				acceptableEventNames = append(acceptableEventNames, string(eventName))
+			}
+
+			attributeChangeSet := migrate.NewMultiConditionalAttributeSet(renamedAttribute.AttributeMap, map[string][]string{
+				"span.name":  acceptableSpanNames,
+				"event.name": acceptableEventNames,
+			})
+			spanEventAttributeChangeSet := operator.NewSpanEventConditionalAttributeOperator(attributeChangeSet)
+			values = append(values, spanEventAttributeChangeSet)
+		} else {
+			panic("spanEvents change must have either RenameEvents or RenameAttributes")
 		}
 	}
-	return migrate.NewConditionalAttributeSetSlice(values...)
+	return &changelist.ChangeList{Migrators: values}
 }
 
-func newMetricNameSignalSlice(metrics ast.Metrics) *migrate.SignalNameChangeSlice {
-	values := make([]*migrate.SignalNameChange, 0, 10)
-	for _, ch := range metrics.Changes {
-		values = append(values, migrate.NewSignalNameChange(ch.RenameMetrics))
+func newLogsChangelist(logs ast.Logs) *changelist.ChangeList {
+	values := make([]migrate.Migrator, 0)
+	for _, at := range logs.Changes {
+		if renamed := at.RenameAttributes; renamed != nil {
+			attributeChangeSet := migrate.NewAttributeChangeSet(renamed.AttributeMap)
+			values = append(values, attributeChangeSet)
+		}
 	}
-	return migrate.NewSignalNameChangeSlice(values...)
+	return &changelist.ChangeList{Migrators: values}
 }
diff --git a/processor/schemaprocessor/internal/translation/revision_v1_test.go b/processor/schemaprocessor/internal/translation/revision_v1_test.go
