[processor/spanmetrics] Resource attributes support

CHANGELOG.md

@@ -25,6 +25,7 @@
 - `jaeger` receiver/exporter: Parse/set Jaeger status with OTel spec values (#6682)
 - `awsecscontainermetricsreceiver`: remove tag from `container.image.name` (#6436)
 - `k8sclusterreceiver`: remove tag from `container.image.name` (#6436)
+- `spanmetricproccessor`: service.name attribute added as a default resource attribute instead of attribute in generated metrics (#6717)
 
 ## 🚀 New components 🚀
 
@@ -48,6 +49,7 @@
 - `skywalkingexporter`: add skywalking metrics exporter (#6528)
 - `deltatorateprocessor`: add int counter support (#6982)
 - `filestorageextension`: document default values (#7022)
+- `spanmetricproccessor`: Support specifying resource attributes to attach to metrics generated from traces (#6717)
 - `redisreceiver`: Migrate the scraper to the mdatagen metrics builder (#6938)  
 
 ## v0.41.0

processor/spanmetricsprocessor/README.md

@@ -31,11 +31,13 @@ latency_bucket{http_method="GET",http_status_code="200",label1="value1",operatio
 ```
 
 Each metric will have _at least_ the following dimensions because they are common across all spans:
-- Service name
 - Operation
 - Span kind
 - Status code
 
+Each metric will have _at least_ the following resource attributes because they are common across all spans:
+- Service name
+
 This processor lets traces to continue through the pipeline unmodified.
 
 The following settings are required:
@@ -48,8 +50,7 @@ The following settings can be optionally configured:
   - Default: `[2ms, 4ms, 6ms, 8ms, 10ms, 50ms, 100ms, 200ms, 400ms, 800ms, 1s, 1400ms, 2s, 5s, 10s, 15s]`
 - `dimensions`: the list of dimensions to add together with the default dimensions defined above.
 
-  Each additional dimension is defined with a `name` which is looked up in the span's collection of attributes or
-  resource attributes (AKA process tags) such as `ip`, `host.name` or `region`.
+  Each additional dimension is defined with a `name` which is looked up in the span's collection of attributes.
 
   If the `name`d attribute is missing in the span, the optional provided `default` is used.
 
@@ -60,6 +61,15 @@ The following settings can be optionally configured:
   One of either `AGGREGATION_TEMPORALITY_CUMULATIVE` or `AGGREGATION_TEMPORALITY_DELTA`.
   - Default: `AGGREGATION_TEMPORALITY_CUMULATIVE`
 
+- `resource_attributes`: the list of resource attributes to add together with the default resource attributes defined 
+  above. Each additional resource attribute is defined with a `name` which is looked up in the span's collection of 
+  resource attributes. If the `name`d resource attribute is missing in the span, the optional provided `default` is 
+  used. If no `default` is provided, this resource attribute will be **omitted** from the metric.
+
+  `service.name` will be automatically added as a resource attribute to all the generated metrics.
+
+- `resource_attributes_cache_size`: the max number of items in the `resouce_key_to_dimensions_cache`. If not provided,
+   will use default value size `1000`.
 ## Examples
 
 The following is a simple example usage of the spanmetrics processor.
@@ -96,6 +106,11 @@ processors:
         default: GET
       - name: http.status_code
     dimensions_cache_size: 1000
+    resource_attributes:
+      - name: region
+        default: us-east-1
+      - name: host_id
+    resource_attributes_cache_size: 1000
     aggregation_temporality: "AGGREGATION_TEMPORALITY_DELTA"     
 
 exporters:

processor/spanmetricsprocessor/config.go

@@ -26,7 +26,7 @@ const (
 	cumulative = "AGGREGATION_TEMPORALITY_CUMULATIVE"
 )
 
-// Dimension defines the dimension name and optional default value if the Dimension is missing from a span attribute.
+// Dimension defines the key and optional default value if the key is missing from a span attribute.
 type Dimension struct {
 	Name    string  `mapstructure:"name"`
 	Default *string `mapstructure:"default"`
@@ -44,7 +44,6 @@ type Config struct {
 	LatencyHistogramBuckets []time.Duration `mapstructure:"latency_histogram_buckets"`
 
 	// Dimensions defines the list of additional dimensions on top of the provided:
-	// - service.name
 	// - operation
 	// - span.kind
 	// - status.code
@@ -58,6 +57,18 @@ type Config struct {
 	DimensionsCacheSize int `mapstructure:"dimensions_cache_size"`
 
 	AggregationTemporality string `mapstructure:"aggregation_temporality"`
+
+	// ResourceAttributes defines the list of additional resource attributes to attach to metrics on top of the provided:
+	// - service.name
+	// These will be fetched from the span's resource attributes. For more details, see:
+	// https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/resource/sdk.md
+	// and https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/resource/semantic_conventions/README.md.
+	ResourceAttributes []Dimension `mapstructure:"resource_attributes"`
+
+	// ResourceAttributesCacheSize defines the size of cache for storing ResourceAttributes, which helps to avoid cache
+	// memory growing indefinitely over the lifetime of the collector.
+	// Optional. See defaultResourceAttributesCacheSize in processor.go for the default value.
+	ResourceAttributesCacheSize int `mapstructure:"resource_attributes_cache_size"`
 }
 
 // GetAggregationTemporality converts the string value given in the config into a MetricAggregationTemporality.

processor/spanmetricsprocessor/config_test.go

@@ -36,25 +36,30 @@ import (
 
 func TestLoadConfig(t *testing.T) {
 	defaultMethod := "GET"
+	defaultRegion := "us-east-1"
 	testcases := []struct {
-		configFile                  string
-		wantMetricsExporter         string
-		wantLatencyHistogramBuckets []time.Duration
-		wantDimensions              []Dimension
-		wantDimensionsCacheSize     int
-		wantAggregationTemporality  string
+		configFile                      string
+		wantMetricsExporter             string
+		wantLatencyHistogramBuckets     []time.Duration
+		wantDimensions                  []Dimension
+		wantDimensionsCacheSize         int
+		wantResourceAttributes          []Dimension
+		wantResourceAttributesCacheSize int
+		wantAggregationTemporality      string
 	}{
 		{
-			configFile:                 "config-2-pipelines.yaml",
-			wantMetricsExporter:        "prometheus",
-			wantAggregationTemporality: cumulative,
-			wantDimensionsCacheSize:    500,
+			configFile:                      "config-2-pipelines.yaml",
+			wantMetricsExporter:             "prometheus",
+			wantAggregationTemporality:      cumulative,
+			wantDimensionsCacheSize:         500,
+			wantResourceAttributesCacheSize: 300,
 		},
 		{
-			configFile:                 "config-3-pipelines.yaml",
-			wantMetricsExporter:        "otlp/spanmetrics",
-			wantAggregationTemporality: cumulative,
-			wantDimensionsCacheSize:    defaultDimensionsCacheSize,
+			configFile:                      "config-3-pipelines.yaml",
+			wantMetricsExporter:             "otlp/spanmetrics",
+			wantAggregationTemporality:      cumulative,
+			wantDimensionsCacheSize:         defaultDimensionsCacheSize,
+			wantResourceAttributesCacheSize: defaultResourceAttributesCacheSize,
 		},
 		{
 			configFile:          "config-full.yaml",
@@ -72,12 +77,19 @@ func TestLoadConfig(t *testing.T) {
 				{"http.method", &defaultMethod},
 				{"http.status_code", nil},
 			},
-			wantDimensionsCacheSize:    1500,
-			wantAggregationTemporality: delta,
+			wantDimensionsCacheSize: 1500,
+			wantResourceAttributes: []Dimension{
+				{"region", &defaultRegion},
+				{"host_id", nil},
+			},
+			wantResourceAttributesCacheSize: 3000,
+			wantAggregationTemporality:      delta,
 		},
 	}
 	for _, tc := range testcases {
+		tc := tc
 		t.Run(tc.configFile, func(t *testing.T) {
+			t.Parallel()
 			// Prepare
 			factories, err := componenttest.NopFactories()
 			require.NoError(t, err)
@@ -100,12 +112,14 @@ func TestLoadConfig(t *testing.T) {
 			require.NotNil(t, cfg)
 			assert.Equal(t,
 				&Config{
-					ProcessorSettings:       config.NewProcessorSettings(config.NewComponentID(typeStr)),
-					MetricsExporter:         tc.wantMetricsExporter,
-					LatencyHistogramBuckets: tc.wantLatencyHistogramBuckets,
-					Dimensions:              tc.wantDimensions,
-					DimensionsCacheSize:     tc.wantDimensionsCacheSize,
-					AggregationTemporality:  tc.wantAggregationTemporality,
+					ProcessorSettings:           config.NewProcessorSettings(config.NewComponentID(typeStr)),
+					MetricsExporter:             tc.wantMetricsExporter,
+					LatencyHistogramBuckets:     tc.wantLatencyHistogramBuckets,
+					Dimensions:                  tc.wantDimensions,
+					DimensionsCacheSize:         tc.wantDimensionsCacheSize,
+					ResourceAttributes:          tc.wantResourceAttributes,
+					ResourceAttributesCacheSize: tc.wantResourceAttributesCacheSize,
+					AggregationTemporality:      tc.wantAggregationTemporality,
 				},
 				cfg.Processors[config.NewComponentID(typeStr)],
 			)

processor/spanmetricsprocessor/factory.go

@@ -39,9 +39,10 @@ func NewFactory() component.ProcessorFactory {
 
 func createDefaultConfig() config.Processor {
 	return &Config{
-		ProcessorSettings:      config.NewProcessorSettings(config.NewComponentID(typeStr)),
-		AggregationTemporality: "AGGREGATION_TEMPORALITY_CUMULATIVE",
-		DimensionsCacheSize:    defaultDimensionsCacheSize,
+		ProcessorSettings:           config.NewProcessorSettings(config.NewComponentID(typeStr)),
+		AggregationTemporality:      "AGGREGATION_TEMPORALITY_CUMULATIVE",
+		DimensionsCacheSize:         defaultDimensionsCacheSize,
+		ResourceAttributesCacheSize: defaultResourceAttributesCacheSize,
 	}
 }
 

processor/spanmetricsprocessor/keybuilder/keybuilder.go

@@ -0,0 +1,59 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package keybuilder // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanmetricsprocessor/keybuilder"
+
+import "strings"
+
+const (
+	separator       = string(byte(0))
+	defaultCapacity = 1024
+)
+
+type KeyBuilder interface {
+	Append(value ...string)
+	String() string
+}
+
+type keyBuilder struct {
+	sb        strings.Builder
+	separator string
+}
+
+var _ KeyBuilder = (*keyBuilder)(nil)
+
+func New() KeyBuilder {
+	b := keyBuilder{
+		sb:        strings.Builder{},
+		separator: separator,
+	}
+	b.sb.Grow(defaultCapacity)
+	return &b
+}
+
+func (mkb *keyBuilder) Append(values ...string) {
+	for _, value := range values {
+		if len(value) == 0 {
+			continue
+		}
+		if mkb.sb.Len() != 0 {
+			mkb.sb.WriteString(mkb.separator)
+		}
+		mkb.sb.WriteString(value)
+	}
+}
+
+func (mkb *keyBuilder) String() string {
+	return mkb.sb.String()
+}

processor/spanmetricsprocessor/keybuilder/keybuilder_test.go

@@ -0,0 +1,60 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package keybuilder
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNew(t *testing.T) {
+	t.Run("should create new keybuilder", func(t *testing.T) {
+		assert.NotPanics(t, func() {
+			assert.NotNil(t, New())
+		})
+	})
+}
+
+func TestMetricKeyBuilderAppend(t *testing.T) {
+	tests := []struct {
+		name   string
+		args   [][]string
+		result string
+	}{
+		{
+			name:   "should skip empty string",
+			args:   [][]string{{"", "abc", "", "def"}},
+			result: fmt.Sprintf("abc%sdef", separator),
+		},
+		{
+			name:   "should concat multiple append",
+			args:   [][]string{{"abc", "def"}, {"", "hij"}},
+			result: fmt.Sprintf("abc%sdef%shij", separator, separator),
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			kb := New()
+			for _, arg := range tt.args {
+				kb.Append(arg...)
+			}
+			assert.Equal(t, tt.result, kb.String())
+		})
+	}
+}