Add profiling support to the count connector (#39577)

Author: dmathieu

.chloggen/countconnector-profiler.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: countconnector
+
+# 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: [39577]
+
+# (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/countconnector/README.md

@@ -18,6 +18,7 @@
 | traces | metrics | [alpha] |
 | metrics | metrics | [alpha] |
 | logs | 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/countconnector/config.go

@@ -29,6 +29,9 @@ const (
 
 	defaultMetricNameLogs = "log.record.count"
 	defaultMetricDescLogs = "The number of log records observed."
+
+	defaultMetricNameProfiles = "profile.count"
+	defaultMetricDescProfiles = "The number of profiles observed."
 )
 
 // Config for the connector
@@ -38,6 +41,7 @@ type Config struct {
 	Metrics    map[string]MetricInfo `mapstructure:"metrics"`
 	DataPoints map[string]MetricInfo `mapstructure:"datapoints"`
 	Logs       map[string]MetricInfo `mapstructure:"logs"`
+	Profiles   map[string]MetricInfo `mapstructure:"profiles"`
 	// prevent unkeyed literal initialization
 	_ struct{}
 }
@@ -115,6 +119,17 @@ func (c *Config) Validate() error {
 			return fmt.Errorf("logs attributes: metric %q: %w", name, err)
 		}
 	}
+	for name, info := range c.Profiles {
+		if name == "" {
+			return errors.New("profiles: metric name missing")
+		}
+		if _, err := filterottl.NewBoolExprForProfile(info.Conditions, filterottl.StandardProfileFuncs(), ottl.PropagateError, component.TelemetrySettings{Logger: zap.NewNop()}); err != nil {
+			return fmt.Errorf("profiles condition: metric %q: %w", name, err)
+		}
+		if err := info.validateAttributes(); err != nil {
+			return fmt.Errorf("profiles attributes: metric %q: %w", name, err)
+		}
+	}
 	return nil
 }
 
@@ -155,6 +170,9 @@ func (c *Config) Unmarshal(componentParser *confmap.Conf) error {
 	if !componentParser.IsSet("logs") {
 		c.Logs = defaultLogsConfig()
 	}
+	if !componentParser.IsSet("profiles") {
+		c.Profiles = defaultProfilesConfig()
+	}
 	return nil
 }
 
@@ -197,3 +215,11 @@ func defaultLogsConfig() map[string]MetricInfo {
 		},
 	}
 }
+
+func defaultProfilesConfig() map[string]MetricInfo {
+	return map[string]MetricInfo{
+		defaultMetricNameProfiles: {
+			Description: defaultMetricDescProfiles,
+		},
+	}
+}

connector/countconnector/config_test.go

@@ -49,6 +49,11 @@ func TestLoadConfig(t *testing.T) {
 						Description: defaultMetricDescLogs,
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					defaultMetricNameProfiles: {
+						Description: defaultMetricDescProfiles,
+					},
+				},
 			},
 		},
 		{
@@ -79,6 +84,11 @@ func TestLoadConfig(t *testing.T) {
 						Description: "My description for default log count metric.",
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					defaultMetricNameProfiles: {
+						Description: "My description for default profile count metric.",
+					},
+				},
 			},
 		},
 		{
@@ -109,6 +119,11 @@ func TestLoadConfig(t *testing.T) {
 						Description: "My log record count.",
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					"my.profiling.count": {
+						Description: "My profile count.",
+					},
+				},
 			},
 		},
 		{
@@ -144,6 +159,12 @@ func TestLoadConfig(t *testing.T) {
 						Conditions:  []string{`IsMatch(resource.attributes["host.name"], "pod-l")`},
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					"my.profiling.count": {
+						Description: "My profile count.",
+						Conditions:  []string{`IsMatch(resource.attributes["host.name"], "pod-l")`},
+					},
+				},
 			},
 		},
 		{
@@ -194,6 +215,15 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					"my.profiling.count": {
+						Description: "My profile count.",
+						Conditions: []string{
+							`IsMatch(resource.attributes["host.name"], "pod-l")`,
+							`IsMatch(resource.attributes["foo"], "bar-l")`,
+						},
+					},
+				},
 			},
 		},
 		{
@@ -236,6 +266,14 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					"my.profiling.count": {
+						Description: "My profile count by environment.",
+						Attributes: []AttributeConfig{
+							{Key: "env"},
+						},
+					},
+				},
 			},
 		},
 		{
@@ -322,6 +360,24 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					"my.profiling.count": {
+						Description: "My profile count.",
+					},
+					"limited.profiling.count": {
+						Description: "Limited profile count.",
+						Conditions:  []string{`IsMatch(resource.attributes["host.name"], "pod-l")`},
+						Attributes: []AttributeConfig{
+							{
+								Key: "env",
+							},
+							{
+								Key:          "component",
+								DefaultValue: "other",
+							},
+						},
+					},
+				},
 			},
 		},
 		{
@@ -369,6 +425,11 @@ func TestLoadConfig(t *testing.T) {
 						},
 					},
 				},
+				Profiles: map[string]MetricInfo{
+					defaultMetricNameProfiles: {
+						Description: defaultMetricDescProfiles,
+					},
+				},
 			},
 		},
 	}
@@ -451,6 +512,17 @@ func TestConfigErrors(t *testing.T) {
 			},
 			expect: "logs: metric name missing",
 		},
+		{
+			name: "missing_metric_name_profile",
+			input: &Config{
+				Profiles: map[string]MetricInfo{
+					"": {
+						Description: defaultMetricDescSpans,
+					},
+				},
+			},
+			expect: "profiles: metric name missing",
+		},
 		{
 			name: "invalid_condition_span",
 			input: &Config{
@@ -511,6 +583,18 @@ func TestConfigErrors(t *testing.T) {
 			},
 			expect: fmt.Sprintf("logs condition: metric %q: unable to parse OTTL condition", defaultMetricNameLogs),
 		},
+		{
+			name: "invalid_condition_profile",
+			input: &Config{
+				Profiles: map[string]MetricInfo{
+					defaultMetricNameProfiles: {
+						Description: defaultMetricDescSpans,
+						Conditions:  []string{"invalid condition"},
+					},
+				},
+			},
+			expect: fmt.Sprintf("profiles condition: metric %q: unable to parse OTTL condition", defaultMetricNameProfiles),
+		},
 	}
 
 	for _, tc := range testCases {

connector/countconnector/connector.go

@@ -13,18 +13,20 @@ 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"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/countconnector/internal/metadata"
 	"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/ottlmetric"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlprofile"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspanevent"
 )
 
-// count can count spans, span event, metrics, data points, or log records
-// and emit the counts onto a metrics pipeline.
+// count can count spans, span event, metrics, data points, log records or
+// profiles and emit the counts onto a metrics pipeline.
 type count struct {
 	metricsConsumer consumer.Metrics
 	component.StartFunc
@@ -35,6 +37,7 @@ type count struct {
 	metricsMetricDefs    map[string]metricDef[ottlmetric.TransformContext]
 	dataPointsMetricDefs map[string]metricDef[ottldatapoint.TransformContext]
 	logsMetricDefs       map[string]metricDef[ottllog.TransformContext]
+	profilesMetricDefs   map[string]metricDef[ottlprofile.TransformContext]
 }
 
 func (c *count) Capabilities() consumer.Capabilities {
@@ -198,3 +201,42 @@ func (c *count) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 	}
 	return c.metricsConsumer.ConsumeMetrics(ctx, countMetrics)
 }
+
+func (c *count) ConsumeProfiles(ctx context.Context, ld pprofile.Profiles) error {
+	var multiError error
+	countMetrics := pmetric.NewMetrics()
+	countMetrics.ResourceMetrics().EnsureCapacity(ld.ResourceProfiles().Len())
+	for i := 0; i < ld.ResourceProfiles().Len(); i++ {
+		resourceProfile := ld.ResourceProfiles().At(i)
+		counter := newCounter[ottlprofile.TransformContext](c.profilesMetricDefs)
+
+		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)
+
+				pCtx := ottlprofile.NewTransformContext(profile, scopeProfile.Scope(), resourceProfile.Resource(), scopeProfile, resourceProfile)
+				attributes := pprofile.FromAttributeIndices(profile.AttributeTable(), profile)
+				multiError = errors.Join(multiError, counter.update(ctx, attributes, pCtx))
+			}
+		}
+
+		if len(counter.counts) == 0 {
+			continue // don't add an empty resource
+		}
+
+		countResource := countMetrics.ResourceMetrics().AppendEmpty()
+		resourceProfile.Resource().Attributes().CopyTo(countResource.Resource().Attributes())
+
+		countResource.ScopeMetrics().EnsureCapacity(resourceProfile.ScopeProfiles().Len())
+		countScope := countResource.ScopeMetrics().AppendEmpty()
+		countScope.Scope().SetName(metadata.ScopeName)
+
+		counter.appendMetricsTo(countScope.Metrics())
+	}
+	if multiError != nil {
+		return multiError
+	}
+	return c.metricsConsumer.ConsumeMetrics(ctx, countMetrics)
+}