[pkg/ottl] add profilesample

.chloggen/ottl-profilesample.yaml

@@ -0,0 +1,27 @@
+# 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: pkg/ottl
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add OTTL support for sample submessage of OTel Profiling signal.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40161]
+
+# (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:
+
+# 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: [api]

pkg/ottl/contexts/internal/ctxprofilesample/context.go

@@ -0,0 +1,16 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ctxprofilesample // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/internal/ctxprofilesample"
+
+import "go.opentelemetry.io/collector/pdata/pprofile"
+
+const (
+	Name   = "profilesample"
+	DocRef = "https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/pkg/ottl/contexts/ottlprofilesample"
+)
+
+type Context interface {
+	GetProfileSample() pprofile.Sample
+	GetProfilesDictionary() pprofile.ProfilesDictionary
+}

pkg/ottl/contexts/internal/ctxprofilesample/profilesample.go

@@ -0,0 +1,218 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ctxprofilesample // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/internal/ctxprofilesample"
+
+import (
+	"context"
+	"errors"
+	"math"
+	"time"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pprofile"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/internal/ctxerror"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/internal/ctxutil"
+)
+
+var (
+	errMaxValueExceed   = errors.New("exceeded max value")
+	errInvalidValueType = errors.New("invalid value type")
+)
+
+func PathGetSetter[K Context](path ottl.Path[K]) (ottl.GetSetter[K], error) {
+	if path == nil {
+		return nil, ctxerror.New("nil", "nil", Name, DocRef)
+	}
+	switch path.Name() {
+	case "locations_start_index":
+		return accessLocationsStartIndex[K](), nil
+	case "locations_length":
+		return accessLocationsLength[K](), nil
+	case "values":
+		return accessValues[K](), nil
+	case "attribute_indices":
+		return accessAttributeIndices[K](), nil
+	case "link_index":
+		return accessLinkIndex[K](), nil
+	case "timestamps_unix_nano":
+		return accessTimestampsUnixNano[K](), nil
+	case "timestamps":
+		return accessTimestamps[K](), nil
+	case "attributes":
+		if path.Keys() == nil {
+			return accessAttributes[K](), nil
+		}
+		return accessAttributesKey(path.Keys()), nil
+	default:
+		return nil, ctxerror.New(path.Name(), path.String(), Name, DocRef)
+	}
+}
+
+func accessLocationsStartIndex[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			return int64(tCtx.GetProfileSample().LocationsStartIndex()), nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			if v, ok := val.(int64); ok {
+				if v >= math.MaxInt32 {
+					return errMaxValueExceed
+				}
+				tCtx.GetProfileSample().SetLocationsStartIndex(int32(v))
+				return nil
+			}
+			return errInvalidValueType
+		},
+	}
+}
+
+func accessLocationsLength[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			return int64(tCtx.GetProfileSample().LocationsLength()), nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			if v, ok := val.(int64); ok {
+				if v >= math.MaxInt32 {
+					return errMaxValueExceed
+				}
+				tCtx.GetProfileSample().SetLocationsLength(int32(v))
+				return nil
+			}
+			return errInvalidValueType
+		},
+	}
+}
+
+func accessValues[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			return ctxutil.GetCommonIntSliceValues[int64](tCtx.GetProfileSample().Value()), nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			return ctxutil.SetCommonIntSliceValues[int64](tCtx.GetProfileSample().Value(), val)
+		},
+	}
+}
+
+func accessAttributeIndices[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			return ctxutil.GetCommonIntSliceValues[int32](tCtx.GetProfileSample().AttributeIndices()), nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			return ctxutil.SetCommonIntSliceValues[int32](tCtx.GetProfileSample().AttributeIndices(), val)
+		},
+	}
+}
+
+func accessLinkIndex[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			return int64(tCtx.GetProfileSample().LinkIndex()), nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			if v, ok := val.(int64); ok {
+				if v >= math.MaxInt32 {
+					return errMaxValueExceed
+				}
+				tCtx.GetProfileSample().SetLinkIndex(int32(v))
+				return nil
+			}
+			return errInvalidValueType
+		},
+	}
+}
+
+func accessTimestampsUnixNano[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			return ctxutil.GetCommonIntSliceValues[uint64](tCtx.GetProfileSample().TimestampsUnixNano()), nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			return ctxutil.SetCommonIntSliceValues[uint64](tCtx.GetProfileSample().TimestampsUnixNano(), val)
+		},
+	}
+}
+
+func accessTimestamps[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			var ts []time.Time
+			for _, t := range tCtx.GetProfileSample().TimestampsUnixNano().All() {
+				ts = append(ts, time.Unix(0, int64(t)).UTC())
+			}
+			return ts, nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			if ts, ok := val.([]time.Time); ok {
+				tCtx.GetProfileSample().TimestampsUnixNano().FromRaw([]uint64{})
+				for _, t := range ts {
+					tCtx.GetProfileSample().TimestampsUnixNano().Append(uint64(t.UTC().UnixNano()))
+				}
+			}
+			return nil
+		},
+	}
+}
+
+func accessAttributes[K Context]() ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(_ context.Context, tCtx K) (any, error) {
+			return pprofile.FromAttributeIndices(tCtx.GetProfilesDictionary().AttributeTable(), tCtx.GetProfileSample()), nil
+		},
+		Setter: func(_ context.Context, tCtx K, val any) error {
+			m, err := ctxutil.GetMap(val)
+			if err != nil {
+				return err
+			}
+			tCtx.GetProfileSample().AttributeIndices().FromRaw([]int32{})
+			for k, v := range m.All() {
+				if err := pprofile.PutAttribute(tCtx.GetProfilesDictionary().AttributeTable(), tCtx.GetProfileSample(), k, v); err != nil {
+					return err
+				}
+			}
+			return nil
+		},
+	}
+}
+
+func accessAttributesKey[K Context](key []ottl.Key[K]) ottl.StandardGetSetter[K] {
+	return ottl.StandardGetSetter[K]{
+		Getter: func(ctx context.Context, tCtx K) (any, error) {
+			return ctxutil.GetMapValue[K](ctx, tCtx, pprofile.FromAttributeIndices(tCtx.GetProfilesDictionary().AttributeTable(), tCtx.GetProfileSample()), key)
+		},
+		Setter: func(ctx context.Context, tCtx K, val any) error {
+			newKey, err := ctxutil.GetMapKeyName(ctx, tCtx, key[0])
+			if err != nil {
+				return err
+			}
+			v := getAttributeValue(tCtx, *newKey)
+			if err := ctxutil.SetIndexableValue[K](ctx, tCtx, v, val, key[1:]); err != nil {
+				return err
+			}
+			return pprofile.PutAttribute(tCtx.GetProfilesDictionary().AttributeTable(), tCtx.GetProfileSample(), *newKey, v)
+		},
+	}
+}
+
+func getAttributeValue[K Context](tCtx K, key string) pcommon.Value {
+	// Find the index of the attribute in the profile's attribute indices
+	// and return the corresponding value from the attribute table.
+	table := tCtx.GetProfilesDictionary().AttributeTable()
+	indices := tCtx.GetProfileSample().AttributeIndices().AsRaw()
+
+	for _, tableIndex := range indices {
+		attr := table.At(int(tableIndex))
+		if attr.Key() == key {
+			v := pcommon.NewValueEmpty()
+			attr.Value().CopyTo(v)
+			return v
+		}
+	}
+
+	return pcommon.NewValueEmpty()
+}

pkg/ottl/contexts/internal/ctxprofilesample/profilesample_test.go

@@ -0,0 +1,100 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ctxprofilesample // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/internal/ctxprofilesample"
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pprofile"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/internal/pathtest"
+)
+
+func TestPathGetSetter(t *testing.T) {
+	tsNow := time.Now().UTC()
+	tests := []struct {
+		path string
+		val  any
+		keys []ottl.Key[*profileSampleContext]
+	}{
+		{
+			path: "locations_start_index",
+			val:  int64(42),
+		},
+		{
+			path: "locations_length",
+			val:  int64(43),
+		},
+		{
+			path: "values",
+			val:  []int64{73, 74, 75},
+		},
+		{
+			path: "attribute_indices",
+			val:  []int64{97, 98, 99},
+		},
+		{
+			path: "link_index",
+			val:  int64(44),
+		},
+		{
+			path: "timestamps_unix_nano",
+			val:  []int64{tsNow.Unix(), 2, 3},
+		},
+		{
+			path: "timestamps",
+			val:  []time.Time{tsNow},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.path, func(t *testing.T) {
+			pathParts := strings.Split(tt.path, " ")
+			path := &pathtest.Path[*profileSampleContext]{N: pathParts[0]}
+			if tt.keys != nil {
+				path.KeySlice = tt.keys
+			}
+			if len(pathParts) > 1 {
+				path.NextPath = &pathtest.Path[*profileSampleContext]{N: pathParts[1]}
+			}
+
+			sample := pprofile.NewSample()
+			dictionary := pprofile.NewProfilesDictionary()
+
+			accessor, err := PathGetSetter(path)
+			require.NoError(t, err)
+
+			err = accessor.Set(context.Background(), newProfileSampleContext(sample, dictionary), tt.val)
+			require.NoError(t, err)
+
+			got, err := accessor.Get(context.Background(), newProfileSampleContext(sample, dictionary))
+			require.NoError(t, err)
+
+			assert.Equal(t, tt.val, got)
+		})
+	}
+}
+
+type profileSampleContext struct {
+	sample     pprofile.Sample
+	dictionary pprofile.ProfilesDictionary
+}
+
+func (p *profileSampleContext) GetProfilesDictionary() pprofile.ProfilesDictionary {
+	return p.dictionary
+}
+
+func (p *profileSampleContext) GetProfileSample() pprofile.Sample {
+	return p.sample
+}
+
+func newProfileSampleContext(sample pprofile.Sample, dictionary pprofile.ProfilesDictionary) *profileSampleContext {
+	return &profileSampleContext{sample: sample, dictionary: dictionary}
+}