[v2][adjuster] Implement otel attribute adjuster to operate on otlp d…

…ata model (#6358)

## Which problem is this PR solving?
- Towards #6344

## Description of the changes
- This PR implements the `OTelTag` adjuster to operate on the OTLP data
model. In the OTLP model, tags are dubbed as attributes so the adjuster
was renamed to `OTELAttribute`.

## How was this change tested?
- Added unit tests

## Checklist
- [x] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [x] I have signed all commits
- [x] I have added unit tests for the new functionality
- [x] I have run lint and test steps successfully
  - for `jaeger`: `make lint test`
  - for `jaeger-ui`: `npm run lint` and `npm run test`

---------

Signed-off-by: Mahad Zaryab <[email protected]>

cmd/query/app/querysvc/adjuster/resourceattributes.go

@@ -0,0 +1,66 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package adjuster
+
+import (
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/jaegertracing/jaeger/pkg/otelsemconv"
+)
+
+var libraryKeys = map[string]struct{}{
+	string(otelsemconv.TelemetrySDKLanguageKey):   {},
+	string(otelsemconv.TelemetrySDKNameKey):       {},
+	string(otelsemconv.TelemetrySDKVersionKey):    {},
+	string(otelsemconv.TelemetryDistroNameKey):    {},
+	string(otelsemconv.TelemetryDistroVersionKey): {},
+}
+
+// ResourceAttributes creates an adjuster that moves the OpenTelemetry library
+// attributes from spans to the parent resource so that the UI can
+// display them separately under Process.
+// https://github.com/jaegertracing/jaeger/issues/4534
+func ResourceAttributes() ResourceAttributesAdjuster {
+	return ResourceAttributesAdjuster{}
+}
+
+type ResourceAttributesAdjuster struct{}
+
+func (o ResourceAttributesAdjuster) Adjust(traces ptrace.Traces) (ptrace.Traces, error) {
+	resourceSpans := traces.ResourceSpans()
+	for i := 0; i < resourceSpans.Len(); i++ {
+		rs := resourceSpans.At(i)
+		resource := rs.Resource()
+		scopeSpans := rs.ScopeSpans()
+		for j := 0; j < scopeSpans.Len(); j++ {
+			ss := scopeSpans.At(j)
+			spans := ss.Spans()
+			for k := 0; k < spans.Len(); k++ {
+				span := spans.At(k)
+				o.moveAttributes(span, resource)
+			}
+		}
+	}
+	return traces, nil
+}
+
+func (ResourceAttributesAdjuster) moveAttributes(span ptrace.Span, resource pcommon.Resource) {
+	replace := make(map[string]pcommon.Value)
+	span.Attributes().Range(func(k string, v pcommon.Value) bool {
+		if _, ok := libraryKeys[k]; ok {
+			replace[k] = v
+		}
+		return true
+	})
+	for k, v := range replace {
+		existing, ok := resource.Attributes().Get(k)
+		if ok && existing.AsRaw() != v.AsRaw() {
+			addWarning(span, "conflicting values between Span and Resource for attribute "+k)
+			continue
+		}
+		v.CopyTo(resource.Attributes().PutEmpty(k))
+		span.Attributes().Remove(k)
+	}
+}

cmd/query/app/querysvc/adjuster/resourceattributes_test.go

@@ -0,0 +1,138 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package adjuster
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/jaegertracing/jaeger/pkg/otelsemconv"
+)
+
+func TestResourceAttributesAdjuster_SpanWithLibraryAttributes(t *testing.T) {
+	traces := ptrace.NewTraces()
+	rs := traces.ResourceSpans().AppendEmpty()
+	span := rs.ScopeSpans().AppendEmpty().Spans().AppendEmpty()
+	span.Attributes().PutStr("random_key", "random_value")
+	span.Attributes().PutStr(string(otelsemconv.TelemetrySDKLanguageKey), "Go")
+	span.Attributes().PutStr(string(otelsemconv.TelemetrySDKNameKey), "opentelemetry")
+	span.Attributes().PutStr(string(otelsemconv.TelemetrySDKVersionKey), "1.27.0")
+	span.Attributes().PutStr(string(otelsemconv.TelemetryDistroNameKey), "opentelemetry")
+	span.Attributes().PutStr(string(otelsemconv.TelemetryDistroVersionKey), "blah")
+	span.Attributes().PutStr("another_key", "another_value")
+
+	adjuster := ResourceAttributes()
+	result, err := adjuster.Adjust(traces)
+	require.NoError(t, err)
+
+	resultSpanAttributes := result.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes()
+	require.Equal(t, 2, resultSpanAttributes.Len())
+	val, ok := resultSpanAttributes.Get("random_key")
+	require.True(t, ok)
+	require.Equal(t, "random_value", val.Str())
+
+	val, ok = resultSpanAttributes.Get("another_key")
+	require.True(t, ok)
+	require.Equal(t, "another_value", val.Str())
+
+	resultResourceAttributes := result.ResourceSpans().At(0).Resource().Attributes()
+
+	val, ok = resultResourceAttributes.Get(string(otelsemconv.TelemetrySDKLanguageKey))
+	require.True(t, ok)
+	require.Equal(t, "Go", val.Str())
+
+	val, ok = resultResourceAttributes.Get(string(otelsemconv.TelemetrySDKNameKey))
+	require.True(t, ok)
+	require.Equal(t, "opentelemetry", val.Str())
+
+	val, ok = resultResourceAttributes.Get(string(otelsemconv.TelemetrySDKVersionKey))
+	require.True(t, ok)
+	require.Equal(t, "1.27.0", val.Str())
+
+	val, ok = resultResourceAttributes.Get(string(otelsemconv.TelemetryDistroNameKey))
+	require.True(t, ok)
+	require.Equal(t, "opentelemetry", val.Str())
+
+	val, ok = resultResourceAttributes.Get(string(otelsemconv.TelemetryDistroVersionKey))
+	require.True(t, ok)
+	require.Equal(t, "blah", val.Str())
+}
+
+func TestResourceAttributesAdjuster_SpanWithoutLibraryAttributes(t *testing.T) {
+	traces := ptrace.NewTraces()
+	rs := traces.ResourceSpans().AppendEmpty()
+	span := rs.ScopeSpans().AppendEmpty().Spans().AppendEmpty()
+	span.Attributes().PutStr("random_key", "random_value")
+
+	adjuster := ResourceAttributes()
+	result, err := adjuster.Adjust(traces)
+	require.NoError(t, err)
+
+	resultSpanAttributes := result.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes()
+	require.Equal(t, 1, resultSpanAttributes.Len())
+	val, ok := resultSpanAttributes.Get("random_key")
+	require.True(t, ok)
+	require.Equal(t, "random_value", val.Str())
+}
+
+func TestResourceAttributesAdjuster_SpanWithConflictingLibraryAttributes(t *testing.T) {
+	traces := ptrace.NewTraces()
+	rs := traces.ResourceSpans().AppendEmpty()
+	rs.Resource().Attributes().PutStr(string(otelsemconv.TelemetrySDKLanguageKey), "Go")
+	span := rs.ScopeSpans().AppendEmpty().Spans().AppendEmpty()
+	span.Attributes().PutStr("random_key", "random_value")
+	span.Attributes().PutStr(string(otelsemconv.TelemetrySDKLanguageKey), "Java")
+
+	adjuster := ResourceAttributes()
+	result, err := adjuster.Adjust(traces)
+	require.NoError(t, err)
+
+	resultSpanAttributes := result.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes()
+	require.Equal(t, 3, resultSpanAttributes.Len())
+	val, ok := resultSpanAttributes.Get("random_key")
+	require.True(t, ok)
+	require.Equal(t, "random_value", val.Str())
+
+	// value remains in the span
+	val, ok = resultSpanAttributes.Get(string(otelsemconv.TelemetrySDKLanguageKey))
+	require.True(t, ok)
+	require.Equal(t, "Java", val.Str())
+
+	val, ok = resultSpanAttributes.Get("jaeger.adjuster.warning")
+	require.True(t, ok)
+	warnings := val.Slice()
+	require.Equal(t, 1, warnings.Len())
+	require.Equal(t, "conflicting values between Span and Resource for attribute telemetry.sdk.language", warnings.At(0).Str())
+
+	resultResourceAttributes := result.ResourceSpans().At(0).Resource().Attributes()
+	val, ok = resultResourceAttributes.Get(string(otelsemconv.TelemetrySDKLanguageKey))
+	require.True(t, ok)
+	require.Equal(t, "Go", val.Str())
+}
+
+func TestResourceAttributesAdjuster_SpanWithNonConflictingLibraryAttributes(t *testing.T) {
+	traces := ptrace.NewTraces()
+	rs := traces.ResourceSpans().AppendEmpty()
+	rs.Resource().Attributes().PutStr(string(otelsemconv.TelemetrySDKLanguageKey), "Go")
+	span := rs.ScopeSpans().AppendEmpty().Spans().AppendEmpty()
+	span.Attributes().PutStr("random_key", "random_value")
+	span.Attributes().PutStr(string(otelsemconv.TelemetrySDKLanguageKey), "Go")
+
+	adjuster := ResourceAttributes()
+	result, err := adjuster.Adjust(traces)
+	require.NoError(t, err)
+
+	resultSpanAttributes := result.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes()
+	require.Equal(t, 1, resultSpanAttributes.Len())
+	val, ok := resultSpanAttributes.Get("random_key")
+	require.True(t, ok)
+	require.Equal(t, "random_value", val.Str())
+
+	resultResourceAttributes := result.ResourceSpans().At(0).Resource().Attributes()
+	val, ok = resultResourceAttributes.Get(string(otelsemconv.TelemetrySDKLanguageKey))
+	require.True(t, ok)
+	require.Equal(t, "Go", val.Str())
+}

cmd/query/app/querysvc/adjuster/warning.go

@@ -0,0 +1,23 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package adjuster
+
+import (
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+const (
+	adjusterWarningAttribute = "jaeger.adjuster.warning"
+)
+
+func addWarning(span ptrace.Span, warning string) {
+	var warnings pcommon.Slice
+	if currWarnings, ok := span.Attributes().Get(adjusterWarningAttribute); ok {
+		warnings = currWarnings.Slice()
+	} else {
+		warnings = span.Attributes().PutEmptySlice(adjusterWarningAttribute)
+	}
+	warnings.AppendEmpty().SetStr(warning)
+}

cmd/query/app/querysvc/adjuster/warning_test.go

@@ -0,0 +1,58 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package adjuster
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func TestAddWarning(t *testing.T) {
+	tests := []struct {
+		name     string
+		existing []string
+		newWarn  string
+		expected []string
+	}{
+		{
+			name:     "add to nil warnings",
+			existing: nil,
+			newWarn:  "new warning",
+			expected: []string{"new warning"},
+		},
+		{
+			name:     "add to empty warnings",
+			existing: []string{},
+			newWarn:  "new warning",
+			expected: []string{"new warning"},
+		},
+		{
+			name:     "add to existing warnings",
+			existing: []string{"existing warning 1", "existing warning 2"},
+			newWarn:  "new warning",
+			expected: []string{"existing warning 1", "existing warning 2", "new warning"},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			span := ptrace.NewSpan()
+			attrs := span.Attributes()
+			if test.existing != nil {
+				warnings := attrs.PutEmptySlice("jaeger.adjuster.warning")
+				for _, warn := range test.existing {
+					warnings.AppendEmpty().SetStr(warn)
+				}
+			}
+			addWarning(span, test.newWarn)
+			warnings, ok := attrs.Get("jaeger.adjuster.warning")
+			assert.True(t, ok)
+			assert.Equal(t, len(test.expected), warnings.Slice().Len())
+			for i, expectedWarn := range test.expected {
+				assert.Equal(t, expectedWarn, warnings.Slice().At(i).Str())
+			}
+		})
+	}
+}