Add profiles support to signaltometrics connector

.chloggen/signaltometrics-profiles.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: signaltometricsconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add profiles support
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39609]
+
+# (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: [user]

connector/signaltometricsconnector/README.md

@@ -20,6 +20,7 @@ logs, or metrics).
 | traces | metrics | [alpha] |
 | logs | metrics | [alpha] |
 | metrics | metrics | [alpha] |
+| profiles | metrics | [alpha] |
 
 [Exporter Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#exporter-pipeline-type
 [Receiver Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#receiver-pipeline-type

connector/signaltometricsconnector/config/config.go

@@ -17,6 +17,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlprofile"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
 )
 
@@ -42,12 +43,13 @@ type Config struct {
 	Spans      []MetricInfo `mapstructure:"spans"`
 	Datapoints []MetricInfo `mapstructure:"datapoints"`
 	Logs       []MetricInfo `mapstructure:"logs"`
+	Profiles   []MetricInfo `mapstructure:"profiles"`
 	// prevent unkeyed literal initialization
 	_ struct{}
 }
 
 func (c *Config) Validate() error {
-	if len(c.Spans) == 0 && len(c.Datapoints) == 0 && len(c.Logs) == 0 {
+	if len(c.Spans) == 0 && len(c.Datapoints) == 0 && len(c.Logs) == 0 && len(c.Profiles) == 0 {
 		return errors.New("no configuration provided, at least one should be specified")
 	}
 	var multiError error // collect all errors at once
@@ -93,6 +95,20 @@ func (c *Config) Validate() error {
 			}
 		}
 	}
+	if len(c.Profiles) > 0 {
+		parser, err := ottlprofile.NewParser(
+			customottl.ProfileFuncs(),
+			component.TelemetrySettings{Logger: zap.NewNop()},
+		)
+		if err != nil {
+			return fmt.Errorf("failed to create parser for OTTL profiles: %w", err)
+		}
+		for _, profile := range c.Profiles {
+			if err := validateMetricInfo(profile, parser); err != nil {
+				multiError = errors.Join(multiError, fmt.Errorf("failed to validate profiles configuration: %w", err))
+			}
+		}
+	}
 	return multiError
 }
 
@@ -118,6 +134,10 @@ func (c *Config) Unmarshal(collectorCfg *confmap.Conf) error {
 		info.ensureDefaults()
 		c.Logs[i] = info
 	}
+	for i, info := range c.Profiles {
+		info.ensureDefaults()
+		c.Profiles[i] = info
+	}
 	return nil
 }
 

connector/signaltometricsconnector/config/config_test.go

@@ -33,6 +33,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "missing required metric name"),
 				fullErrorForSignal(t, "datapoints", "missing required metric name"),
 				fullErrorForSignal(t, "logs", "missing required metric name"),
+				fullErrorForSignal(t, "profiles", "missing required metric name"),
 			},
 		},
 		{
@@ -41,6 +42,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "attributes validation failed"),
 				fullErrorForSignal(t, "datapoints", "attributes validation failed"),
 				fullErrorForSignal(t, "logs", "attributes validation failed"),
+				fullErrorForSignal(t, "profiles", "attributes validation failed"),
 			},
 		},
 		{
@@ -49,6 +51,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "attributes validation failed"),
 				fullErrorForSignal(t, "datapoints", "attributes validation failed"),
 				fullErrorForSignal(t, "logs", "attributes validation failed"),
+				fullErrorForSignal(t, "profiles", "attributes validation failed"),
 			},
 		},
 		{
@@ -57,6 +60,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "attributes validation failed: only one of default_value or optional should be set"),
 				fullErrorForSignal(t, "datapoints", "attributes validation failed: only one of default_value or optional should be set"),
 				fullErrorForSignal(t, "logs", "attributes validation failed: only one of default_value or optional should be set"),
+				fullErrorForSignal(t, "profiles", "attributes validation failed: only one of default_value or optional should be set"),
 			},
 		},
 		{
@@ -65,6 +69,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "histogram validation failed"),
 				fullErrorForSignal(t, "datapoints", "histogram validation failed"),
 				fullErrorForSignal(t, "logs", "histogram validation failed"),
+				fullErrorForSignal(t, "profiles", "histogram validation failed"),
 			},
 		},
 		{
@@ -73,6 +78,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "histogram validation failed"),
 				fullErrorForSignal(t, "datapoints", "histogram validation failed"),
 				fullErrorForSignal(t, "logs", "histogram validation failed"),
+				fullErrorForSignal(t, "profiles", "histogram validation failed"),
 			},
 		},
 		{
@@ -81,6 +87,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "sum validation failed"),
 				fullErrorForSignal(t, "datapoints", "sum validation failed"),
 				fullErrorForSignal(t, "logs", "sum validation failed"),
+				fullErrorForSignal(t, "profiles", "sum validation failed"),
 			},
 		},
 		{
@@ -89,6 +96,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "exactly one of the metrics must be defined"),
 				fullErrorForSignal(t, "datapoints", "exactly one of the metrics must be defined"),
 				fullErrorForSignal(t, "logs", "exactly one of the metrics must be defined"),
+				fullErrorForSignal(t, "profiles", "exactly one of the metrics must be defined"),
 			},
 		},
 		{
@@ -97,6 +105,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "failed to parse value OTTL expression"),
 				fullErrorForSignal(t, "datapoints", "failed to parse value OTTL expression"),
 				fullErrorForSignal(t, "logs", "failed to parse value OTTL expression"),
+				fullErrorForSignal(t, "profiles", "failed to parse value OTTL expression"),
 			},
 		},
 		{
@@ -105,6 +114,7 @@ func TestConfig(t *testing.T) {
 				fullErrorForSignal(t, "spans", "failed to parse OTTL conditions"),
 				fullErrorForSignal(t, "datapoints", "failed to parse OTTL conditions"),
 				fullErrorForSignal(t, "logs", "failed to parse OTTL conditions"),
+				fullErrorForSignal(t, "profiles", "failed to parse OTTL conditions"),
 			},
 		},
 		{
@@ -187,6 +197,24 @@ func TestConfig(t *testing.T) {
 						},
 					},
 				},
+				Profiles: []MetricInfo{
+					{
+						Name:                      "profile.sum",
+						Description:               "Sum",
+						Unit:                      "1",
+						IncludeResourceAttributes: []Attribute{{Key: "key.1", DefaultValue: "foo"}},
+						Attributes: []Attribute{
+							{Key: "key.2", DefaultValue: "bar"},
+							{Key: "key.3", Optional: true},
+						},
+						Conditions: []string{
+							`duration_unix_nano > 0`,
+						},
+						Sum: &Sum{
+							Value: "1",
+						},
+					},
+				},
 			},
 		},
 	} {
@@ -220,7 +248,7 @@ func fullErrorForSignal(t *testing.T, signal, errMsg string) string {
 	t.Helper()
 
 	switch signal {
-	case "spans", "datapoints", "logs":
+	case "spans", "datapoints", "logs", "profiles":
 		return fmt.Sprintf(validationMsgFormat, signal, errMsg)
 	default:
 		panic("unhandled signal type")

connector/signaltometricsconnector/connector.go

@@ -12,13 +12,15 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/pprofile"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/signaltometricsconnector/internal/aggregator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/signaltometricsconnector/internal/model"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlprofile"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
 )
 
@@ -27,9 +29,10 @@ type signalToMetrics struct {
 	collectorInstanceInfo model.CollectorInstanceInfo
 	logger                *zap.Logger
 
-	spanMetricDefs []model.MetricDef[ottlspan.TransformContext]
-	dpMetricDefs   []model.MetricDef[ottldatapoint.TransformContext]
-	logMetricDefs  []model.MetricDef[ottllog.TransformContext]
+	spanMetricDefs    []model.MetricDef[ottlspan.TransformContext]
+	dpMetricDefs      []model.MetricDef[ottldatapoint.TransformContext]
+	logMetricDefs     []model.MetricDef[ottllog.TransformContext]
+	profileMetricDefs []model.MetricDef[ottlprofile.TransformContext]
 
 	component.StartFunc
 	component.ShutdownFunc
@@ -242,3 +245,54 @@ func (sm *signalToMetrics) ConsumeLogs(ctx context.Context, logs plog.Logs) erro
 	aggregator.Finalize(sm.logMetricDefs)
 	return sm.next.ConsumeMetrics(ctx, processedMetrics)
 }
+
+func (sm *signalToMetrics) ConsumeProfiles(ctx context.Context, profiles pprofile.Profiles) error {
+	if len(sm.profileMetricDefs) == 0 {
+		return nil
+	}
+
+	processedMetrics := pmetric.NewMetrics()
+	processedMetrics.ResourceMetrics().EnsureCapacity(profiles.ResourceProfiles().Len())
+	aggregator := aggregator.NewAggregator[ottlprofile.TransformContext](processedMetrics)
+
+	for i := 0; i < profiles.ResourceProfiles().Len(); i++ {
+		resourceProfile := profiles.ResourceProfiles().At(i)
+		resourceAttrs := resourceProfile.Resource().Attributes()
+
+		for j := 0; j < resourceProfile.ScopeProfiles().Len(); j++ {
+			scopeProfile := resourceProfile.ScopeProfiles().At(j)
+
+			for k := 0; k < scopeProfile.Profiles().Len(); k++ {
+				profile := scopeProfile.Profiles().At(k)
+				profileAttrs := pprofile.FromAttributeIndices(profile.AttributeTable(), profile)
+
+				for _, md := range sm.profileMetricDefs {
+					filteredProfileAttrs, ok := md.FilterAttributes(profileAttrs)
+					if !ok {
+						continue
+					}
+
+					// The transform context is created from original attributes so that the
+					// OTTL expressions are also applied on the original attributes.
+					tCtx := ottlprofile.NewTransformContext(profile, scopeProfile.Scope(), resourceProfile.Resource(), scopeProfile, resourceProfile)
+					if md.Conditions != nil {
+						match, err := md.Conditions.Eval(ctx, tCtx)
+						if err != nil {
+							return fmt.Errorf("failed to evaluate conditions: %w", err)
+						}
+						if !match {
+							sm.logger.Debug("condition not matched, skipping", zap.String("name", md.Key.Name))
+							continue
+						}
+					}
+					filteredResAttrs := md.FilterResourceAttributes(resourceAttrs, sm.collectorInstanceInfo)
+					if err := aggregator.Aggregate(ctx, tCtx, md, filteredResAttrs, filteredProfileAttrs, 1); err != nil {
+						return err
+					}
+				}
+			}
+		}
+	}
+	aggregator.Finalize(sm.profileMetricDefs)
+	return sm.next.ConsumeMetrics(ctx, processedMetrics)
+}

connector/signaltometricsconnector/connector_test.go

@@ -16,6 +16,7 @@ import (
 	"go.opentelemetry.io/collector/confmap/xconfmap"
 	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/connector/connectortest"
+	"go.opentelemetry.io/collector/connector/xconnector"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -128,6 +129,40 @@ func TestConnectorWithLogs(t *testing.T) {
 	}
 }
 
+func TestConnectorWithProfiles(t *testing.T) {
+	testCases := []string{
+		"sum",
+		"histograms",
+		"exponential_histograms",
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	for _, tc := range testCases {
+		t.Run(tc, func(t *testing.T) {
+			profileTestDataDir := filepath.Join(testDataDir, "profiles")
+			inputProfiles, err := golden.ReadProfiles(filepath.Join(profileTestDataDir, "profiles.yaml"))
+			require.NoError(t, err)
+
+			next := &consumertest.MetricsSink{}
+			tcTestDataDir := filepath.Join(profileTestDataDir, tc)
+			factory, settings, cfg := setupConnector(t, tcTestDataDir)
+			connector, err := factory.CreateProfilesToMetrics(ctx, settings, cfg, next)
+			require.NoError(t, err)
+			require.IsType(t, &signalToMetrics{}, connector)
+
+			require.NoError(t, connector.ConsumeProfiles(ctx, inputProfiles))
+			require.Len(t, next.AllMetrics(), 1)
+
+			expectedMetrics, err := golden.ReadMetrics(filepath.Join(tcTestDataDir, "output.yaml"))
+			require.NoError(t, err)
+
+			assertAggregatedMetrics(t, expectedMetrics, next.AllMetrics()[0])
+		})
+	}
+}
+
 func BenchmarkConnectorWithTraces(b *testing.B) {
 	factory := NewFactory()
 	settings := connectortest.NewNopSettings(metadata.Type)
@@ -285,7 +320,7 @@ func testMetricInfo(b *testing.B) []config.MetricInfo {
 
 func setupConnector(
 	t *testing.T, testFilePath string,
-) (connector.Factory, connector.Settings, component.Config) {
+) (xconnector.Factory, connector.Settings, component.Config) {
 	t.Helper()
 	factory := NewFactory()
 	settings := connectortest.NewNopSettings(metadata.Type)
@@ -300,7 +335,7 @@ func setupConnector(
 	require.NoError(t, sub.Unmarshal(&cfg))
 	require.NoError(t, xconfmap.Validate(cfg))
 
-	return factory, settings, cfg
+	return factory.(xconnector.Factory), settings, cfg
 }
 
 func telemetryResource(t *testing.T) pcommon.Resource {