Add otelcol.processor.groupbyattrs component

CHANGELOG.md

@@ -117,6 +117,9 @@ Main (unreleased)
   - `otelcol.connector.spanmetrics`: Produce delta temporality span metrics with StartTimeUnixNano and TimeUnixNano values representing an uninterrupted series.
     https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/31780
 
+- Add a `otelcol.processor.groupbyattrs` component to reassociate collected metrics that match specified attributes
+  from opentelemetry. (@kehindesalaam)
+
 - Upgrade Beyla component v1.6.3 to v1.7.0
   - Reporting application process metrics
   - New supported protocols: SQL, Redis, Kafka

docs/sources/reference/compatibility/_index.md

@@ -300,6 +300,7 @@ The following components, grouped by namespace, _export_ OpenTelemetry `otelcol.
 - [otelcol.processor.deltatocumulative](../components/otelcol/otelcol.processor.deltatocumulative)
 - [otelcol.processor.discovery](../components/otelcol/otelcol.processor.discovery)
 - [otelcol.processor.filter](../components/otelcol/otelcol.processor.filter)
+- [otelcol.processor.groupbyattrs](../components/otelcol/otelcol.processor.groupbyattrs)
 - [otelcol.processor.k8sattributes](../components/otelcol/otelcol.processor.k8sattributes)
 - [otelcol.processor.memory_limiter](../components/otelcol/otelcol.processor.memory_limiter)
 - [otelcol.processor.probabilistic_sampler](../components/otelcol/otelcol.processor.probabilistic_sampler)
@@ -336,6 +337,7 @@ The following components, grouped by namespace, _consume_ OpenTelemetry `otelcol
 - [otelcol.processor.deltatocumulative](../components/otelcol/otelcol.processor.deltatocumulative)
 - [otelcol.processor.discovery](../components/otelcol/otelcol.processor.discovery)
 - [otelcol.processor.filter](../components/otelcol/otelcol.processor.filter)
+- [otelcol.processor.groupbyattrs](../components/otelcol/otelcol.processor.groupbyattrs)
 - [otelcol.processor.k8sattributes](../components/otelcol/otelcol.processor.k8sattributes)
 - [otelcol.processor.memory_limiter](../components/otelcol/otelcol.processor.memory_limiter)
 - [otelcol.processor.probabilistic_sampler](../components/otelcol/otelcol.processor.probabilistic_sampler)

docs/sources/reference/components/otelcol/otelcol.processor.groupbyattrs.md

@@ -0,0 +1,175 @@
+---
+canonical: https://grafana.com/docs/alloy/latest/reference/components/otelcol/otelcol.processor.groupbyattrs/
+description: Learn about otelcol.processor.groupbyattrs
+title: otelcol.processor.groupbyattrs
+---
+
+# otelcol.processor.groupbyattrs
+
+`otelcol.processor.groupbyattrs` accepts spans, metrics, and traces from other `otelcol`
+components and groups them under the same resource.
+
+{{% admonition type="note" %}}
+`otelcol.processor.groupbyattrs` is a wrapper over the upstream OpenTelemetry
+Collector `groupbyattrs` processor. If necessary, bug reports or feature requests
+will be redirected to the upstream repository.
+{{% /admonition %}}
+
+We recommend you use the groupbyattrs processor together with [otelcol.processor.batch][], as a consecutive step. This will reduce the fragmentation of data by grouping records together under the matching Resource/Instrumentation Library.
+
+You can specify multiple `otelcol.processor.groupbyattrs` components by giving them
+different labels.
+
+## Usage
+
+```alloy
+otelcol.processor.groupbyattrs "LABEL" {
+  output {
+    metrics = [...]
+    logs    = [...]
+    traces  = [...]
+  }
+}
+```
+
+## Arguments
+
+The following arguments are supported:
+
+| Name            | Type              | Description                                                                           | Default | Required |
+|-----------------|-------------------|---------------------------------------------------------------------------------------|---------|----------|
+| `keys`          | `list(string)`    | Keys that will be used to group the spans, log records or metric data points together |         | no       |
+| `output`        | [output][]        | Configures where to send received telemetry data.                                     | yes     |          |
+| `debug_metrics` | [debug_metrics][] | Configures the metrics that this component generates to monitor its state.            | no      |          |
+
+[output]: #output-block
+[debug_metrics]: #debug_metrics-block
+
+`keys` is a string array that is used for grouping the data. If it is empty, the processor performs compaction and reassociates all spans with matching Resource and InstrumentationLibrary.
+
+
+### output block
+
+{{< docs/shared lookup="reference/components/output-block.md" source="alloy" version="<ALLOY_VERSION>" >}}
+
+### debug_metrics block
+
+{{< docs/shared lookup="reference/components/otelcol-debug-metrics-block.md" source="alloy" version="<ALLOY_VERSION>" >}}
+
+## Exported fields
+
+The following fields are exported and can be referenced by other components:
+
+| Name    | Type               | Description                                                   |
+|---------|--------------------|---------------------------------------------------------------|
+| `input` | `otelcol.Consumer` | Accepts `otelcol.Consumer` data for metrics, logs, or traces. |
+
+`input` accepts `otelcol.Consumer` data for any telemetry signal (metrics,
+logs, or traces).
+
+## Component health
+
+`otelcol.processor.groupbyattrs` is only reported as unhealthy if given an invalid
+configuration.
+
+## Debug information
+
+`otelcol.processor.groupbyattrs` doesn't expose any component-specific debug
+information.
+
+## Debug metrics
+
+`otelcol.processor.groupbyattrs` doesn't expose any component-specific debug metrics.
+
+## Examples
+
+### Grouping metrics
+
+Consider the following metrics, all originally associated to the same Resource:
+
+```
+Resource {host.name="localhost",source="prom"}
+  Metric "gauge-1" (GAUGE)
+    DataPoint {host.name="host-A",id="eth0"}
+    DataPoint {host.name="host-A",id="eth0"}
+    DataPoint {host.name="host-B",id="eth0"}
+  Metric "gauge-1" (GAUGE) // Identical to previous Metric
+    DataPoint {host.name="host-A",id="eth0"}
+    DataPoint {host.name="host-A",id="eth0"}
+    DataPoint {host.name="host-B",id="eth0"}
+  Metric "mixed-type" (GAUGE)
+    DataPoint {host.name="host-A",id="eth0"}
+    DataPoint {host.name="host-A",id="eth0"}
+    DataPoint {host.name="host-B",id="eth0"}
+  Metric "mixed-type" (SUM)
+    DataPoint {host.name="host-A",id="eth0"}
+    DataPoint {host.name="host-A",id="eth0"}
+  Metric "dont-move" (Gauge)
+    DataPoint {id="eth0"}
+```
+
+With the following configuration, the groupbyattrs will re-associate the metrics with either `host-A` or `host-B`, based on the value of the `host.name` attribute.
+
+```alloy
+otelcol.processor.groupbyattrs "default" {
+  keys = [ "host.name" ]
+  output {
+    metrics = [otelcol.exporter.otlp.default.input]
+  }
+}
+```
+
+The output of the processor will therefore be:
+
+```
+Resource {host.name="localhost",source="prom"}
+  Metric "dont-move" (Gauge)
+    DataPoint {id="eth0"}
+Resource {host.name="host-A",source="prom"}
+  Metric "gauge-1"
+    DataPoint {id="eth0"}
+    DataPoint {id="eth0"}
+    DataPoint {id="eth0"}
+    DataPoint {id="eth0"}
+  Metric "mixed-type" (GAUGE)
+    DataPoint {id="eth0"}
+    DataPoint {id="eth0"}
+  Metric "mixed-type" (SUM)
+    DataPoint {id="eth0"}
+    DataPoint {id="eth0"}
+Resource {host.name="host-B",source="prom"}
+  Metric "gauge-1"
+    DataPoint {id="eth0"}
+    DataPoint {id="eth0"}
+  Metric "mixed-type" (GAUGE)
+    DataPoint {id="eth0"}
+```
+
+This output demonstrates how `otelcol.processor.groupbyattrs` works in various situations:
+
+- The DataPoints for the `gauge-1` (GAUGE) metric were originally split under 2 Metric instances and have been merged in the output.
+- The DataPoints of the `mixed-type` (GAUGE) and `mixed-type` (SUM) metrics have not been merged under the same Metric, because their DataType is different.
+- The `dont-move` metric DataPoints don't have a `host.name` attribute and therefore remained under the original Resource.
+- The new Resources inherited the attributes from the original Resource (`source="prom"`), plus the specified attributes from the processed metrics (`host.name="host-A"` or `host.name="host-B"`).
+- The specified "grouping" attributes that are set on the new Resources are also removed from the metric DataPoints.
+- While not shown in the above example, the processor also merges collections of records under matching InstrumentationLibrary.
+
+[otelcol.processor.batch]: https://grafana.com/docs/alloy/latest/reference/components/otelcol/otelcol.processor.batch/
+<!-- START GENERATED COMPATIBLE COMPONENTS -->
+
+## Compatible components
+
+`otelcol.processor.groupbyattrs` can accept arguments from the following components:
+
+- Components that export [OpenTelemetry `otelcol.Consumer`](../../../compatibility/#opentelemetry-otelcolconsumer-exporters)
+
+`otelcol.processor.groupbyattrs` has exports that can be consumed by the following components:
+
+- Components that consume [OpenTelemetry `otelcol.Consumer`](../../../compatibility/#opentelemetry-otelcolconsumer-consumers)
+
+{{< admonition type="note" >}}
+Connecting some components may not be sensible or components may require further configuration to make the connection work correctly.
+Refer to the linked documentation for more details.
+{{< /admonition >}}
+
+<!-- END GENERATED COMPATIBLE COMPONENTS -->

go.mod

@@ -121,6 +121,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.105.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.105.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.105.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor v0.105.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor v0.105.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor v0.105.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/resourcedetectionprocessor v0.105.0

go.sum

@@ -1770,6 +1770,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumul
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor v0.105.0/go.mod h1:PKox+dLnO2bWc1qUN6WZnyHPV0MpWZ10arqGV5v69kI=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.105.0 h1:oRa+acTM4f5rjTT3+hjOVM1LYrlwrm6CSNG4o/RIqcA=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/filterprocessor v0.105.0/go.mod h1:66cZFd4X8vQBTmvm1hPHxrSNHS474iUEsAVbYk9xQBU=
+github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor v0.105.0 h1:OYsGaSC9G7pAVYKTd1+D0f7HTHcxuQfoEHyQy+a1NKk=
+github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor v0.105.0/go.mod h1:WCesGEakYveZYZH4o3cUTLt3UB7JxE+yDiiphRHoJoc=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor v0.105.0 h1:ScIwuYg6l79Ta+deOyZIADXrBlXSdeAZ7sp3MXhm7JY=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/k8sattributesprocessor v0.105.0/go.mod h1:pranRmnWRkzDsn9a16BzSqX6HJ6XjjVVFmMhyZPEzt0=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/probabilisticsamplerprocessor v0.105.0 h1:mFAlBmDFELQJS8uj1M8csB/vQqjpq6W9/9k9izh9Hr4=

internal/component/all/all.go

@@ -82,6 +82,7 @@ import (
 	_ "github.com/grafana/alloy/internal/component/otelcol/processor/deltatocumulative"      // Import otelcol.processor.deltatocumulative
 	_ "github.com/grafana/alloy/internal/component/otelcol/processor/discovery"              // Import otelcol.processor.discovery
 	_ "github.com/grafana/alloy/internal/component/otelcol/processor/filter"                 // Import otelcol.processor.filter
+	_ "github.com/grafana/alloy/internal/component/otelcol/processor/groupbyattrs"           // Import otelcol.processor.groupbyattrs
 	_ "github.com/grafana/alloy/internal/component/otelcol/processor/k8sattributes"          // Import otelcol.processor.k8sattributes
 	_ "github.com/grafana/alloy/internal/component/otelcol/processor/memorylimiter"          // Import otelcol.processor.memory_limiter
 	_ "github.com/grafana/alloy/internal/component/otelcol/processor/probabilistic_sampler"  // Import otelcol.processor.probabilistic_sampler

internal/component/otelcol/processor/groupbyattrs/groupbyattrs.go

@@ -0,0 +1,83 @@
+package groupbyattrs
+
+import (
+	"github.com/grafana/alloy/internal/component"
+	"github.com/grafana/alloy/internal/component/otelcol"
+	otelcolCfg "github.com/grafana/alloy/internal/component/otelcol/config"
+	"github.com/grafana/alloy/internal/component/otelcol/processor"
+	"github.com/grafana/alloy/internal/featuregate"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor"
+	otelcomponent "go.opentelemetry.io/collector/component"
+	otelextension "go.opentelemetry.io/collector/extension"
+)
+
+func init() {
+	component.Register(component.Registration{
+		Name:      "otelcol.processor.groupbyattrs",
+		Stability: featuregate.StabilityGenerallyAvailable,
+		Exports:   otelcol.ConsumerExports{},
+		Args:      Arguments{},
+
+		Build: func(opts component.Options, args component.Arguments) (component.Component, error) {
+			fact := groupbyattrsprocessor.NewFactory()
+			return processor.New(opts, fact, args.(Arguments))
+		},
+	})
+}
+
+type Arguments struct {
+
+	// Keys is a list of attributes to group metrics by.
+	Keys []string `alloy:"keys,attr,optional"`
+
+	// Output configures where to send processed data. Required.
+	Output *otelcol.ConsumerArguments `alloy:"output,block"`
+
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcolCfg.DebugMetricsArguments `alloy:"debug_metrics,block,optional"`
+}
+
+var (
+	_ processor.Arguments = Arguments{}
+)
+
+// SetToDefault implements syntax.Defaulter.
+func (args *Arguments) SetToDefault() {
+	*args = Arguments{
+		Keys: []string{},
+	}
+	args.DebugMetrics.SetToDefault()
+}
+
+// Validate implements syntax.Validator.
+func (args *Arguments) Validate() error {
+	return nil
+}
+
+// Convert implements processor.Arguments.
+func (args Arguments) Convert() (otelcomponent.Config, error) {
+	return &groupbyattrsprocessor.Config{
+		GroupByKeys: args.Keys,
+	}, nil
+}
+
+// Extensions implements processor.Arguments.
+func (args Arguments) Extensions() map[otelcomponent.ID]otelextension.Extension {
+	return nil
+}
+
+// Exporters implements processor.Arguments.
+func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.ID]otelcomponent.Component {
+	return nil
+}
+
+// NextConsumers implements processor.Arguments.
+func (args Arguments) NextConsumers() *otelcol.ConsumerArguments {
+	return args.Output
+}
+
+// DebugMetricsConfig implements processor.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcolCfg.DebugMetricsArguments {
+	return args.DebugMetrics
+}

internal/component/otelcol/processor/groupbyattrs/groupbyattrs_test.go

@@ -0,0 +1,80 @@
+package groupbyattrs_test
+
+import (
+	"testing"
+
+	"github.com/grafana/alloy/internal/component/otelcol/processor/groupbyattrs"
+	"github.com/grafana/alloy/syntax"
+
+	"github.com/mitchellh/mapstructure"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor"
+	"github.com/stretchr/testify/require"
+)
+
+func TestArguments_UnmarshalAlloy(t *testing.T) {
+	tests := []struct {
+		testName string
+		cfg      string
+		expected map[string]interface{}
+		errMsg   string
+	}{
+		{
+			testName: "Default",
+			cfg: `
+			output {}
+			`,
+			expected: map[string]interface{}{
+				"keys": []string{},
+			},
+		},
+		{
+			testName: "SingleKey",
+			cfg: `
+			keys = ["key1"]
+			output {}
+			`,
+			expected: map[string]interface{}{
+				"keys": []string{
+					"key1",
+				},
+			},
+		},
+		{
+			testName: "MultipleKeys",
+			cfg: `
+			keys = ["key1", "key2"]
+			output {}
+			`,
+			expected: map[string]interface{}{
+				"keys": []string{
+					"key1",
+					"key2",
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.testName, func(t *testing.T) {
+			var args groupbyattrs.Arguments
+			err := syntax.Unmarshal([]byte(tt.cfg), &args)
+			if tt.errMsg != "" {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tt.errMsg)
+				return
+			}
+			require.NoError(t, err)
+
+			actualPtr, err := args.Convert()
+			require.NoError(t, err)
+
+			actual := actualPtr.(*groupbyattrsprocessor.Config)
+
+			var expectedCfg groupbyattrsprocessor.Config
+			err = mapstructure.Decode(tt.expected, &expectedCfg)
+			require.NoError(t, err)
+
+			require.Equal(t, expectedCfg, *actual)
+		})
+	}
+}