Add ability to process metrics in schema processor

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

@@ -0,0 +1,29 @@
+# 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: |
+  Add functionality to transform metrics for the schema processor.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38628]
+
+# (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 metrics 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: []

processor/schemaprocessor/processor.go

@@ -51,7 +51,45 @@ func (t schemaProcessor) processLogs(_ context.Context, ld plog.Logs) (plog.Logs
 	return ld, nil
 }
 
-func (t schemaProcessor) processMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
+func (t schemaProcessor) processMetrics(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
+	for mt := 0; mt < md.ResourceMetrics().Len(); mt++ {
+		rMetric := md.ResourceMetrics().At(mt)
+		resourceSchemaURL := rMetric.SchemaUrl()
+		if resourceSchemaURL != "" {
+			t.log.Debug("requesting translation for resourceSchemaURL", zap.String("resourceSchemaURL", resourceSchemaURL))
+			tr, err := t.manager.RequestTranslation(context.Background(), resourceSchemaURL)
+			if err != nil {
+				t.log.Error("failed to request translation", zap.Error(err))
+				return md, err
+			}
+			err = tr.ApplyAllResourceChanges(rMetric, resourceSchemaURL)
+			if err != nil {
+				t.log.Error("failed to apply resource changes", zap.Error(err))
+				return md, err
+			}
+		}
+		for sm := 0; sm < rMetric.ScopeMetrics().Len(); sm++ {
+			metric := rMetric.ScopeMetrics().At(sm)
+			metricSchemaURL := metric.SchemaUrl()
+			if metricSchemaURL == "" {
+				metricSchemaURL = resourceSchemaURL
+			}
+			if metricSchemaURL == "" {
+				continue
+			}
+			tr, err := t.manager.
+				RequestTranslation(ctx, metricSchemaURL)
+			if err != nil {
+				t.log.Error("failed to request translation", zap.Error(err))
+				continue
+			}
+			err = tr.ApplyScopeMetricChanges(metric, metricSchemaURL)
+			if err != nil {
+				t.log.Error("failed to apply scope metric changes", zap.Error(err))
+			}
+		}
+	}
+
 	return md, nil
 }
 

processor/schemaprocessor/processor_metrics_test.go

@@ -0,0 +1,235 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package schemaprocessor
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+func TestMetrics_RenameAttributes(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name            string
+		in              pmetric.Metrics
+		out             pmetric.Metrics
+		transformations string
+		targetVersion   string
+	}{
+		{
+			name: "one_version_downgrade",
+			in: func() pmetric.Metrics {
+				in := pmetric.NewMetrics()
+				in.ResourceMetrics().AppendEmpty()
+				in.ResourceMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.9.0")
+				in.ResourceMetrics().At(0).Resource().Attributes().PutStr("new.resource.name", "test-cluster")
+				in.ResourceMetrics().At(0).ScopeMetrics().AppendEmpty()
+
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.9.0")
+				// Sum metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetName("new.sum.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetEmptySum()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				// Gauge metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetName("new.gauge.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetEmptyGauge()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				// Histogram metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetName("new.histogram.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetEmptyHistogram()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				// Summary metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetName("new.summary.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetEmptySummary()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				return in
+			}(),
+			out: func() pmetric.Metrics {
+				out := pmetric.NewMetrics()
+				out.ResourceMetrics().AppendEmpty()
+				out.ResourceMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.8.0")
+				out.ResourceMetrics().At(0).Resource().Attributes().PutStr("old.resource.name", "test-cluster")
+
+				out.ResourceMetrics().At(0).ScopeMetrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.8.0")
+				// Sum metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetName("old.sum.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetEmptySum()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				// Gauge metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetName("old.gauge.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetEmptyGauge()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				// Histogram metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetName("old.histogram.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetEmptyHistogram()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				// Summary metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetName("old.summary.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetEmptySummary()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				return out
+			}(),
+			transformations: `
+	1.9.0:
+	  all:
+		changes:
+		  - rename_attributes:
+			  attribute_map:
+				old.resource.name: new.resource.name
+	  metrics:
+		changes:
+			- rename_attributes:
+				  attribute_map:
+					old.attr.name: new.attr.name
+	    	- rename_metrics:
+				  old.sum.metric: new.sum.metric
+				  old.gauge.metric: new.gauge.metric
+			- rename_metrics:
+				  old.histogram.metric: new.histogram.metric
+				  old.summary.metric: new.summary.metric
+	1.8.0:`,
+			targetVersion: "1.8.0",
+		},
+		{
+			name: "one_version_upgrade",
+			in: func() pmetric.Metrics {
+				in := pmetric.NewMetrics()
+				in.ResourceMetrics().AppendEmpty()
+				in.ResourceMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.8.0")
+				in.ResourceMetrics().At(0).Resource().Attributes().PutStr("old.resource.name", "test-cluster")
+				in.ResourceMetrics().At(0).ScopeMetrics().AppendEmpty()
+
+				// Sum metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.8.0")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetName("old.sum.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetEmptySum()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				// Gauge metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetName("old.gauge.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetEmptyGauge()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				// Histogram metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetName("old.histogram.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetEmptyHistogram()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				// Summary metric
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetName("old.summary.metric")
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetEmptySummary()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().AppendEmpty()
+				in.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().At(0).Attributes().PutStr("old.attr.name", "test-cluster")
+
+				return in
+			}(),
+			out: func() pmetric.Metrics {
+				out := pmetric.NewMetrics()
+				out.ResourceMetrics().AppendEmpty()
+				out.ResourceMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.9.0")
+				out.ResourceMetrics().At(0).Resource().Attributes().PutStr("new.resource.name", "test-cluster")
+				out.ResourceMetrics().At(0).ScopeMetrics().AppendEmpty()
+
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).SetSchemaUrl("http://opentelemetry.io/schemas/1.9.0")
+				// Sum metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetName("new.sum.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).SetEmptySum()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(0).Sum().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				// Gauge metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetName("new.gauge.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).SetEmptyGauge()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(1).Gauge().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				// Histogram metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetName("new.histogram.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).SetEmptyHistogram()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(2).Histogram().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				// Summary metric
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetName("new.summary.metric")
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).SetEmptySummary()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().AppendEmpty()
+				out.ResourceMetrics().At(0).ScopeMetrics().At(0).Metrics().At(3).Summary().DataPoints().At(0).Attributes().PutStr("new.attr.name", "test-cluster")
+
+				return out
+			}(),
+			transformations: `
+	1.9.0:
+	  all:
+		changes:
+		  - rename_attributes:
+			  attribute_map:
+				old.resource.name: new.resource.name
+	  metrics:
+		changes:
+		  - rename_attributes:
+			  attribute_map:
+				old.attr.name: new.attr.name
+		  - rename_metrics:
+				old.sum.metric: new.sum.metric
+				old.gauge.metric: new.gauge.metric
+		  - rename_metrics:
+				old.histogram.metric: new.histogram.metric
+				old.summary.metric: new.summary.metric
+	1.8.0:`,
+			targetVersion: "1.9.0",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pr := newTestSchemaProcessor(t, tt.transformations, tt.targetVersion)
+			ctx := context.Background()
+			out, err := pr.processMetrics(ctx, tt.in)
+			if err != nil {
+				t.Errorf("Error while processing metrics: %v", err)
+			}
+			assert.Equal(t, tt.out, out, "Metrics transformation failed")
+		})
+	}
+}

processor/schemaprocessor/processor_test.go

@@ -25,11 +25,12 @@ type dummySchemaProvider struct {
 }
 
 func (m *dummySchemaProvider) Retrieve(_ context.Context, _ string) (string, error) {
+	transformations := strings.ReplaceAll(m.transformations, "\t", "    ")
 	data := fmt.Sprintf(`
 file_format: 1.1.0
 schema_url: http://opentelemetry.io/schemas/1.9.0
-versions:
-  %s`, m.transformations)
+versions:%s`, transformations)
+
 	data = strings.TrimSpace(data)
 	return data, nil
 }