Add ability to process traces (#38486)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
Adds ability to transform traces based on the schema target. 



<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes

<!--Describe what testing was performed and which tests were added.-->
#### Testing

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

Author: dineshg13

.chloggen/dinesh.gurumurthy_schema-processor-add-functioality-traces.yaml

@@ -0,0 +1,28 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: schemaprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds functionality to transform traces
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38486]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  - Adds functionality to transform traces using the target schema version.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

cmd/otelcontribcol/builder-config.yaml

@@ -127,6 +127,7 @@ processors:
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourceprocessor v0.121.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor v0.121.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/sumologicprocessor v0.121.0
+  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor v0.121.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanprocessor v0.121.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.121.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.121.0

processor/schemaprocessor/DESIGN.md

@@ -19,7 +19,7 @@ graph LR;
     end
 
 ``` 
-The [Transformer](transformer.go) is registered as a Processor in the Collector by the factory.
+The [Schema Processor](processor.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.
 

processor/schemaprocessor/README.md

@@ -1,4 +1,4 @@
-# Schema Transformer Processor
+# Schema Processor
 
 <!-- status autogenerated section -->
 | Status        |           |
@@ -59,4 +59,4 @@ 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.
+There's a rough design/overview of the processor in the [DESIGN.md](./DESIGN.md) file.

processor/schemaprocessor/factory.go

@@ -22,7 +22,7 @@ var processorCapabilities = consumer.Capabilities{MutatesData: true}
 // factory will store any of the precompiled schemas in future
 type factory struct{}
 
-// newDefaultConfiguration returns the configuration for schema transformer processor
+// newDefaultConfiguration returns the configuration for schema processor
 // with the default values being used throughout it
 func newDefaultConfiguration() component.Config {
 	return &Config{
@@ -47,7 +47,7 @@ func (f factory) createLogsProcessor(
 	cfg component.Config,
 	next consumer.Logs,
 ) (processor.Logs, error) {
-	transformer, err := newTransformer(ctx, cfg, set)
+	schemaProcessor, err := newSchemaProcessor(ctx, cfg, set)
 	if err != nil {
 		return nil, err
 	}
@@ -56,9 +56,9 @@ func (f factory) createLogsProcessor(
 		set,
 		cfg,
 		next,
-		transformer.processLogs,
+		schemaProcessor.processLogs,
 		processorhelper.WithCapabilities(processorCapabilities),
-		processorhelper.WithStart(transformer.start),
+		processorhelper.WithStart(schemaProcessor.start),
 	)
 }
 
@@ -68,7 +68,7 @@ func (f factory) createMetricsProcessor(
 	cfg component.Config,
 	next consumer.Metrics,
 ) (processor.Metrics, error) {
-	transformer, err := newTransformer(ctx, cfg, set)
+	schemaProcessor, err := newSchemaProcessor(ctx, cfg, set)
 	if err != nil {
 		return nil, err
 	}
@@ -77,9 +77,9 @@ func (f factory) createMetricsProcessor(
 		set,
 		cfg,
 		next,
-		transformer.processMetrics,
+		schemaProcessor.processMetrics,
 		processorhelper.WithCapabilities(processorCapabilities),
-		processorhelper.WithStart(transformer.start),
+		processorhelper.WithStart(schemaProcessor.start),
 	)
 }
 
@@ -89,7 +89,7 @@ func (f factory) createTracesProcessor(
 	cfg component.Config,
 	next consumer.Traces,
 ) (processor.Traces, error) {
-	transformer, err := newTransformer(ctx, cfg, set)
+	schemaProcessor, err := newSchemaProcessor(ctx, cfg, set)
 	if err != nil {
 		return nil, err
 	}
@@ -98,8 +98,8 @@ func (f factory) createTracesProcessor(
 		set,
 		cfg,
 		next,
-		transformer.processTraces,
+		schemaProcessor.processTraces,
 		processorhelper.WithCapabilities(processorCapabilities),
-		processorhelper.WithStart(transformer.start),
+		processorhelper.WithStart(schemaProcessor.start),
 	)
 }

processor/schemaprocessor/internal/translation/manager.go

@@ -70,6 +70,7 @@ func NewManager(targetSchemaURLS []string, log *zap.Logger, providers ...Provide
 }
 
 func (m *manager) RequestTranslation(ctx context.Context, schemaURL string) (Translation, error) {
+	m.log.Debug("Requesting translation for schemaURL", zap.String("schema-url", schemaURL))
 	family, version, err := GetFamilyAndVersion(schemaURL)
 	if err != nil {
 		m.log.Error("No valid schema url was provided",

processor/schemaprocessor/processor.go

@@ -0,0 +1,117 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package schemaprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor"
+
+import (
+	"context"
+	"errors"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/collector/processor"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/schemaprocessor/internal/translation"
+)
+
+type schemaProcessor struct {
+	telemetry component.TelemetrySettings
+	config    *Config
+
+	log *zap.Logger
+
+	manager translation.Manager
+}
+
+func newSchemaProcessor(_ context.Context, conf component.Config, set processor.Settings) (*schemaProcessor, error) {
+	cfg, ok := conf.(*Config)
+	if !ok {
+		return nil, errors.New("invalid configuration provided")
+	}
+
+	m, err := translation.NewManager(
+		cfg.Targets,
+		set.Logger.Named("schema-manager"),
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &schemaProcessor{
+		config:    cfg,
+		telemetry: set.TelemetrySettings,
+		log:       set.Logger,
+		manager:   m,
+	}, nil
+}
+
+func (t schemaProcessor) processLogs(_ context.Context, ld plog.Logs) (plog.Logs, error) {
+	return ld, nil
+}
+
+func (t schemaProcessor) processMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
+	return md, nil
+}
+
+func (t schemaProcessor) processTraces(ctx context.Context, td ptrace.Traces) (ptrace.Traces, error) {
+	for rt := 0; rt < td.ResourceSpans().Len(); rt++ {
+		rTrace := td.ResourceSpans().At(rt)
+		resourceSchemaURL := rTrace.SchemaUrl()
+
+		if resourceSchemaURL != "" {
+			t.log.Debug("requesting translation for resourceSchemaURL", zap.String("resourceSchemaURL", resourceSchemaURL))
+			tr, err := t.manager.
+				RequestTranslation(ctx, resourceSchemaURL)
+			if err != nil {
+				t.log.Error("failed to request translation", zap.Error(err))
+				return td, err
+			}
+			err = tr.ApplyAllResourceChanges(rTrace, resourceSchemaURL)
+			if err != nil {
+				t.log.Error("failed to apply resource changes", zap.Error(err))
+				return td, err
+			}
+		}
+		for ss := 0; ss < rTrace.ScopeSpans().Len(); ss++ {
+			span := rTrace.ScopeSpans().At(ss)
+			spanSchemaURL := span.SchemaUrl()
+			if spanSchemaURL == "" {
+				spanSchemaURL = resourceSchemaURL
+			}
+			if spanSchemaURL == "" {
+				continue
+			}
+			tr, err := t.manager.
+				RequestTranslation(ctx, spanSchemaURL)
+			if err != nil {
+				t.log.Error("failed to request translation", zap.Error(err))
+				continue
+			}
+			err = tr.ApplyScopeSpanChanges(span, spanSchemaURL)
+			if err != nil {
+				t.log.Error("failed to apply scope span changes", zap.Error(err))
+			}
+		}
+	}
+	return td, nil
+}
+
+// start will add HTTP provider to the manager and prefetch schemas
+func (t *schemaProcessor) start(ctx context.Context, host component.Host) error {
+	client, err := t.config.ToClient(ctx, host, t.telemetry)
+	if err != nil {
+		return err
+	}
+	t.manager.AddProvider(translation.NewHTTPProvider(client))
+
+	go func(ctx context.Context) {
+		for _, schemaURL := range t.config.Prefetch {
+			t.log.Info("prefetching schema", zap.String("url", schemaURL))
+			_, _ = t.manager.RequestTranslation(ctx, schemaURL)
+		}
+	}(ctx)
+
+	return nil
+}

processor/schemaprocessor/transformer_test.go processor/schemaprocessor/processor_test.go

@@ -6,6 +6,8 @@ package schemaprocessor
 import (
 	"context"
 	_ "embed"
+	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -18,27 +20,49 @@ import (
 	"go.uber.org/zap/zaptest"
 )
 
-func newTestTransformer(t *testing.T) *transformer {
-	trans, err := newTransformer(context.Background(), newDefaultConfiguration(), processor.Settings{
+type dummySchemaProvider struct {
+	transformations string
+}
+
+func (m *dummySchemaProvider) Retrieve(_ context.Context, _ string) (string, error) {
+	data := fmt.Sprintf(`
+file_format: 1.1.0
+schema_url: http://opentelemetry.io/schemas/1.9.0
+versions:
+  %s`, m.transformations)
+	data = strings.TrimSpace(data)
+	return data, nil
+}
+
+func newTestSchemaProcessor(t *testing.T, transformations string, targerVerion string) *schemaProcessor {
+	cfg := &Config{
+		Targets: []string{fmt.Sprintf("http://opentelemetry.io/schemas/%s", targerVerion)},
+	}
+	trans, err := newSchemaProcessor(context.Background(), cfg, processor.Settings{
 		TelemetrySettings: component.TelemetrySettings{
 			Logger: zaptest.NewLogger(t),
 		},
 	})
-	require.NoError(t, err, "Must not error when creating default transformer")
+	require.NoError(t, err, "Must not error when creating default schemaProcessor")
+	trans.manager.AddProvider(&dummySchemaProvider{
+		transformations: transformations,
+	})
 	return trans
 }
 
-func TestTransformerStart(t *testing.T) {
+func TestSchemaProcessorStart(t *testing.T) {
 	t.Parallel()
 
-	trans := newTestTransformer(t)
+	trans := newTestSchemaProcessor(t, "", "1.9.0")
 	assert.NoError(t, trans.start(context.Background(), nil))
 }
 
-func TestTransformerProcessing(t *testing.T) {
+func TestSchemaProcessorProcessing(t *testing.T) {
 	t.Parallel()
+	// these tests are just to ensure that the processor does not error out
+	// and that the data is not modified as dummyprovider has no transformations
 
-	trans := newTestTransformer(t)
+	trans := newTestSchemaProcessor(t, "", "1.9.0")
 	t.Run("metrics", func(t *testing.T) {
 		in := pmetric.NewMetrics()
 		in.ResourceMetrics().AppendEmpty()
@@ -52,7 +76,7 @@ func TestTransformerProcessing(t *testing.T) {
 
 		out, err := trans.processMetrics(context.Background(), in)
 		assert.NoError(t, err, "Must not error when processing metrics")
-		assert.Equal(t, in, out, "Must return the same data (subject to change)")
+		assert.Equal(t, in, out, "Must return the same data")
 	})
 
 	t.Run("traces", func(t *testing.T) {
@@ -67,8 +91,8 @@ func TestTransformerProcessing(t *testing.T) {
 		s.CopyTo(in.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0))
 
 		out, err := trans.processTraces(context.Background(), in)
-		assert.NoError(t, err, "Must not error when processing metrics")
-		assert.Equal(t, in, out, "Must return the same data (subject to change)")
+		assert.NoError(t, err, "Must not error when processing traces")
+		assert.Equal(t, in, out, "Must return the same data")
 	})
 
 	t.Run("logs", func(t *testing.T) {
@@ -83,6 +107,6 @@ func TestTransformerProcessing(t *testing.T) {
 
 		out, err := trans.processLogs(context.Background(), in)
 		assert.NoError(t, err, "Must not error when processing metrics")
-		assert.Equal(t, in, out, "Must return the same data (subject to change)")
+		assert.Equal(t, in, out, "Must return the same data")
 	})
 }