-
Notifications
You must be signed in to change notification settings - Fork 2.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[chore] Schema Processor Revamp [Part 2] - ChangeList and Revision (#…
…35267) **Description:** <Describe what has changed.> This is a slice of changes from #35248 This PR details how operators are used to build the execution pipeline for a given schemafile. Changed files from the [previous PR](#35214) are: processor/schemaprocessor/internal/changelist/changelist.go processor/schemaprocessor/internal/translation/revision_v1.go processor/schemaprocessor/internal/translation/revision_v1_test.go processor/schemaprocessor/go.mod I'm asking a maintainer if they would be willing to push a copy of the previous PR's branch to the core repo so I can switch the base of this PR to the previous PR - thus only the stacked changes would be shown. Edit: this is apparently not easily supported - so asking reviewers to just focus on the changed files listed above. Sorry about that! **Testing:** <Describe what testing was performed and which tests were added.> Unit tests --------- Co-authored-by: Pablo Baeyens <[email protected]>
- Loading branch information
1 parent
6119d51
commit 9ae6a48
Showing
26 changed files
with
1,299 additions
and
487 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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()`). |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
84 changes: 84 additions & 0 deletions
84
processor/schemaprocessor/internal/changelist/changelist.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| // 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/pcommon" | ||
| "go.opentelemetry.io/collector/pdata/plog" | ||
| "go.opentelemetry.io/collector/pdata/pmetric" | ||
| "go.opentelemetry.io/collector/pdata/ptrace" | ||
|
|
||
| "github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/migrate" | ||
| "github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/transformer" | ||
| ) | ||
|
|
||
| // 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 transformer types - what do the transformers act on? | ||
| switch thisMigrator := migrator.(type) { | ||
| // this one acts on both spans and span events! | ||
| case transformer.Transformer[ptrace.Span]: | ||
| if span, ok := signal.(ptrace.Span); ok { | ||
| if err := thisMigrator.Do(ss, span); err != nil { | ||
| return err | ||
| } | ||
| } else { | ||
| return fmt.Errorf("span Transformer %T can't act on %T", thisMigrator, signal) | ||
| } | ||
| case transformer.Transformer[pmetric.Metric]: | ||
| if metric, ok := signal.(pmetric.Metric); ok { | ||
| if err := thisMigrator.Do(ss, metric); err != nil { | ||
| return err | ||
| } | ||
| } else { | ||
| return fmt.Errorf("metric Transformer %T can't act on %T", thisMigrator, signal) | ||
| } | ||
| case transformer.Transformer[plog.LogRecord]: | ||
| if log, ok := signal.(plog.LogRecord); ok { | ||
| if err := thisMigrator.Do(ss, log); err != nil { | ||
| return err | ||
| } | ||
| } else { | ||
| return fmt.Errorf("log Transformer %T can't act on %T", thisMigrator, signal) | ||
| } | ||
| case transformer.Transformer[pcommon.Resource]: | ||
| if resource, ok := signal.(pcommon.Resource); ok { | ||
| if err := thisMigrator.Do(ss, resource); err != nil { | ||
| return err | ||
| } | ||
| } else { | ||
| return fmt.Errorf("resource Transformer %T can't act on %T", thisMigrator, signal) | ||
| } | ||
| case transformer.AllAttributes: | ||
| if err := thisMigrator.Do(ss, signal); err != nil { | ||
| return err | ||
| } | ||
| 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) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.