Schema Processor Revamp [Part 1] - Operators and Migrators

processor/schemaprocessor/DESIGN.md

@@ -0,0 +1,28 @@
+# Design
+
+The Schema Processor is split into several different components.
+
+Here's a general structure diagram:
+
+```mermaid
+graph LR;
+    A[Previous Collector Component] --> B[Transformer]
+    B -- Schema URL --> C[Translation Manager]
+    C -- Translation --> B
+    B --> H[Translator]
+    H --> E[Revision]
+    E --> I[ChangeList]
+    subgraph Interpreter
+        direction RL
+        I --> F[Transformer]
+        F --> G[Migrator]
+    end
+
+``` 
+The [Transformer](transformer.go) is registered as a Processor in the Collector by the factory.
+Data flows into the Transformer, which uses the Schema URL to fetch the translation from the Translation Manager.
+The Translation Manager (at internal/translation/manager.go in a future PR) is responsible for fetching and caching the translations.  It takes in a schema URL and returns a Translator struct.
+
+The Translator struct contains the target schema URL, the target schema version, and a list of Revisions.  The Translator figures out what the version of the incoming data is and what Revisions to apply to the incoming data to get it to the target schema version. The Translator is also responsible for applying the Revisions to the incoming data - it iterates through these Revisions and applies them to the incoming data.   
+
+Each Revision represents all the changes within a specific version.  It consists of several ChangeLists (at internal/changelist/changelist.go in a future PR) - one for each type of change block (at the time of writing - `all`, `resources`, `spans`, `spanEvents`, `metrics`, `logs`).  Each ChangeList is similar to a program in an interpreter - in this case the programming language is the schema file!  They iterate through whatever changes they are constructed with, and call a [Transformer](internal/transformer) for each type of change.  The Transformer accepts a typed value - a log, a metric, etc.  It then, under the hood, calls one of a few Migrators.  The Migrators do the fundamental work of changing attributes, changing names, etc.  The Migrators generally operate on lower levels than the Transformers - they operate on `Attributes`, or an `alias.NamedSignal` (a signal that implements `Name()` and `SetName()`).

processor/schemaprocessor/README.md

@@ -6,7 +6,7 @@
 | Stability     | [development]: traces, metrics, logs   |
 | Distributions | [] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aprocessor%2Fschema%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aprocessor%2Fschema) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aprocessor%2Fschema%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aprocessor%2Fschema) |
-| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@MovieStoreGuy](https://www.github.com/MovieStoreGuy) |
+| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@MovieStoreGuy](https://www.github.com/MovieStoreGuy), [@ankitpatel96](https://www.github.com/ankitpatel96) |
 
 [development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development
 <!-- end autogenerated section -->
@@ -59,3 +59,5 @@ processors:
 ```
 
 For more complete examples, please refer to [config.yml](./testdata/config.yml).
+
+There's a rough design/overview of the processor in the [DESIGN.md](./DESIGN.md) file.

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/migrate/attributes.go

@@ -11,24 +11,22 @@ 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 {
+	// for ambiguous rollbacks (if updates contains entries with multiple keys that have the same value), rollback contains the last key iterated over in mappings
+	attr := AttributeChangeSet{
 		updates:  make(map[string]string, len(mappings)),
 		rollback: make(map[string]string, len(mappings)),
 	}
@@ -39,15 +37,9 @@ func NewAttributeChangeSet(mappings ast.AttributeMap) *AttributeChangeSet {
 	return attr
 }
 
-func (a *AttributeChangeSet) Apply(attrs pcommon.Map) error {
-	return a.do(StateSelectorApply, attrs)
-}
+func (a AttributeChangeSet) IsMigrator() {}
 
-func (a *AttributeChangeSet) Rollback(attrs pcommon.Map) error {
-	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 +73,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
@@ -104,7 +104,7 @@ func TestAttributeChangeSetApply(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			err := tc.acs.Apply(tc.attrs)
+			err := tc.acs.Do(StateSelectorApply, tc.attrs)
 			if tc.errVal == "" {
 				assert.NoError(t, err, "Must not return an error")
 			} else {
@@ -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
@@ -179,7 +179,7 @@ func TestAttributeChangeSetRollback(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			err := tc.acs.Rollback(tc.attrs)
+			err := tc.acs.Do(StateSelectorRollback, tc.attrs)
 			if tc.errVal == "" {
 				assert.NoError(t, err, "Must not return an error")
 			} else {
@@ -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")
-		})
-	}
-}

processor/schemaprocessor/internal/migrate/conditional.go

@@ -6,7 +6,6 @@ package migrate // import "github.com/open-telemetry/opentelemetry-collector-con
 import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/otel/schema/v1.0/ast"
-	"go.uber.org/multierr"
 )
 
 // ValueMatch defines the expected match type
@@ -15,74 +14,43 @@ type ValueMatch interface {
 	~string
 }
 
+// ConditionalAttributeSet represents a rename_attribute that will happen only if the passed in value matches the `on` set.
 type ConditionalAttributeSet struct {
-	on    *map[string]struct{}
-	attrs *AttributeChangeSet
+	on    map[string]struct{}
+	attrs AttributeChangeSet
 }
 
 type ConditionalAttributeSetSlice []*ConditionalAttributeSet
 
-func NewConditionalAttributeSet[Match ValueMatch](mappings ast.AttributeMap, matches ...Match) *ConditionalAttributeSet {
+func NewConditionalAttributeSet[Match ValueMatch](mappings ast.AttributeMap, matches ...Match) ConditionalAttributeSet {
 	on := make(map[string]struct{})
 	for _, m := range matches {
 		on[string(m)] = struct{}{}
 	}
-	return &ConditionalAttributeSet{
-		on:    &on,
+	return ConditionalAttributeSet{
+		on:    on,
 		attrs: NewAttributeChangeSet(mappings),
 	}
 }
 
-func (ca *ConditionalAttributeSet) Apply(attrs pcommon.Map, values ...string) (errs error) {
-	if ca.check(values...) {
-		errs = ca.attrs.Apply(attrs)
-	}
-	return errs
-}
+func (ca ConditionalAttributeSet) IsMigrator() {}
 
-func (ca *ConditionalAttributeSet) Rollback(attrs pcommon.Map, values ...string) (errs error) {
+// Do applies the attribute changes specified in the constructor if any of the values in values matches the matches specified in the constructor.
+func (ca *ConditionalAttributeSet) Do(ss StateSelector, attrs pcommon.Map, values ...string) (errs error) {
 	if ca.check(values...) {
-		errs = ca.attrs.Rollback(attrs)
+		errs = ca.attrs.Do(ss, attrs)
 	}
 	return errs
 }
 
 func (ca *ConditionalAttributeSet) check(values ...string) bool {
-	if len(*ca.on) == 0 {
+	if len(ca.on) == 0 {
 		return true
 	}
 	for _, v := range values {
-		if _, ok := (*ca.on)[v]; !ok {
+		if _, ok := (ca.on)[v]; !ok {
 			return false
 		}
 	}
 	return true
 }
-
-func NewConditionalAttributeSetSlice(conditions ...*ConditionalAttributeSet) *ConditionalAttributeSetSlice {
-	values := new(ConditionalAttributeSetSlice)
-	for _, c := range conditions {
-		(*values) = append((*values), c)
-	}
-	return values
-}
-
-func (slice *ConditionalAttributeSetSlice) Apply(attrs pcommon.Map, values ...string) error {
-	return slice.do(StateSelectorApply, attrs, values)
-}
-
-func (slice *ConditionalAttributeSetSlice) Rollback(attrs pcommon.Map, values ...string) error {
-	return slice.do(StateSelectorRollback, attrs, values)
-}
-
-func (slice *ConditionalAttributeSetSlice) do(ss StateSelector, attrs pcommon.Map, values []string) (errs error) {
-	for i := 0; i < len((*slice)); i++ {
-		switch ss {
-		case StateSelectorApply:
-			errs = multierr.Append(errs, (*slice)[i].Apply(attrs, values...))
-		case StateSelectorRollback:
-			errs = multierr.Append(errs, (*slice)[len((*slice))-i-1].Rollback(attrs, values...))
-		}
-	}
-	return errs
-}