[connector/signaltometrics]Add otelcol instance info as res attribute (#178)

* [connector/signaltometrics]Add collector instance info as res attribute

Metric data streams MUST obey single-writer. `signaltometrics`
the component produces metrics from non-metric signals as well as alters
resource attributes from the source signals. To keep the single-writer
valid, the component adds collector information as resource attributes
with the component name (`signaltometrics`) as the prefix for all
added resource attributes.
---------

Co-authored-by: Andrew Wilkins <[email protected]>

Author: lahsivjar

connector/signaltometricsconnector/config/config.go

@@ -143,8 +143,11 @@ type MetricInfo struct {
 	// IncludeResourceAttributes is a list of resource attributes that
 	// needs to be included in the generated metric. If no resource
 	// attribute is included in the list then all attributes are included.
-	// Note that configuring this setting might cause the produced metric
-	// to lose its identity or cause identity conflict.
+	// Metric data streams MUST obey single-writer. The component produces
+	// metrics from non-metric signals as well as alters resource attributes
+	// from the source signals. To keep the single-writer valid, the
+	// component adds collector information as resource attribute with the
+	// component name as the prefix of the resource attributes.
 	IncludeResourceAttributes []Attribute `mapstructure:"include_resource_attributes"`
 	Attributes                []Attribute `mapstructure:"attributes"`
 	// Conditions are a set of OTTL condtions which are ORd. Data is

connector/signaltometricsconnector/connector.go

@@ -41,7 +41,8 @@ type signalToMetrics struct {
 	component.StartFunc
 	component.ShutdownFunc
 
-	logger *zap.Logger
+	collectorInstanceInfo *model.CollectorInstanceInfo
+	logger                *zap.Logger
 
 	next           consumer.Metrics
 	spanMetricDefs []model.MetricDef[ottlspan.TransformContext]
@@ -73,10 +74,8 @@ func (sm *signalToMetrics) ConsumeTraces(ctx context.Context, td ptrace.Traces)
 				spanAttrs := span.Attributes()
 				adjustedCount := int64(trace.CalculateAdjustedCount(span.TraceState().AsRaw()))
 				for _, md := range sm.spanMetricDefs {
-					filteredSpanAttrs := getFilteredAttributes(spanAttrs, md.Attributes)
-					if filteredSpanAttrs.Len() != len(md.Attributes) {
-						// If any of the configured attributes is not present in
-						// source metric then don't count them.
+					filteredSpanAttrs, ok := md.FilterAttributes(spanAttrs)
+					if !ok {
 						continue
 					}
 
@@ -95,14 +94,7 @@ func (sm *signalToMetrics) ConsumeTraces(ctx context.Context, td ptrace.Traces)
 						}
 					}
 
-					var filteredResAttrs pcommon.Map
-					if len(md.IncludeResourceAttributes) > 0 {
-						filteredResAttrs = getFilteredAttributes(resourceAttrs, md.IncludeResourceAttributes)
-					} else {
-						// Copy resource attrs to avoid mutating data
-						filteredResAttrs = pcommon.NewMap()
-						resourceAttrs.CopyTo(filteredResAttrs)
-					}
+					filteredResAttrs := md.FilterResourceAttributes(resourceAttrs, sm.collectorInstanceInfo)
 					multiError = errors.Join(multiError, aggregator.Aggregate(ctx, tCtx, md, filteredResAttrs, filteredSpanAttrs, adjustedCount))
 				}
 			}
@@ -130,15 +122,7 @@ func (sm *signalToMetrics) ConsumeMetrics(ctx context.Context, m pmetric.Metrics
 				metrics := scopeMetric.Metrics()
 				metric := metrics.At(k)
 				for _, md := range sm.dpMetricDefs {
-					var filteredResAttrs pcommon.Map
-					if len(md.IncludeResourceAttributes) > 0 {
-						filteredResAttrs = getFilteredAttributes(resourceAttrs, md.IncludeResourceAttributes)
-					} else {
-						// Copy resource attrs to avoid mutating data
-						filteredResAttrs = pcommon.NewMap()
-						resourceAttrs.CopyTo(filteredResAttrs)
-					}
-
+					filteredResAttrs := md.FilterResourceAttributes(resourceAttrs, sm.collectorInstanceInfo)
 					aggregate := func(dp any, dpAttrs pcommon.Map) error {
 						// The transform context is created from orginal attributes so that the
 						// OTTL expressions are also applied on the original attributes.
@@ -163,10 +147,8 @@ func (sm *signalToMetrics) ConsumeMetrics(ctx context.Context, m pmetric.Metrics
 						dps := metric.Gauge().DataPoints()
 						for l := 0; l < dps.Len(); l++ {
 							dp := dps.At(l)
-							filteredDPAttrs := getFilteredAttributes(dp.Attributes(), md.Attributes)
-							if filteredDPAttrs.Len() != len(md.Attributes) {
-								// If all the configured attributes are not present in
-								// source metric then don't count them.
+							filteredDPAttrs, ok := md.FilterAttributes(dp.Attributes())
+							if !ok {
 								continue
 							}
 							multiError = errors.Join(multiError, aggregate(dp, filteredDPAttrs))
@@ -175,10 +157,8 @@ func (sm *signalToMetrics) ConsumeMetrics(ctx context.Context, m pmetric.Metrics
 						dps := metric.Sum().DataPoints()
 						for l := 0; l < dps.Len(); l++ {
 							dp := dps.At(l)
-							filteredDPAttrs := getFilteredAttributes(dp.Attributes(), md.Attributes)
-							if filteredDPAttrs.Len() != len(md.Attributes) {
-								// If all the configured attributes are not present in
-								// source metric then don't count them.
+							filteredDPAttrs, ok := md.FilterAttributes(dp.Attributes())
+							if !ok {
 								continue
 							}
 							multiError = errors.Join(multiError, aggregate(dp, filteredDPAttrs))
@@ -187,10 +167,8 @@ func (sm *signalToMetrics) ConsumeMetrics(ctx context.Context, m pmetric.Metrics
 						dps := metric.Summary().DataPoints()
 						for l := 0; l < dps.Len(); l++ {
 							dp := dps.At(l)
-							filteredDPAttrs := getFilteredAttributes(dp.Attributes(), md.Attributes)
-							if filteredDPAttrs.Len() != len(md.Attributes) {
-								// If all the configured attributes are not present in
-								// source metric then don't count them.
+							filteredDPAttrs, ok := md.FilterAttributes(dp.Attributes())
+							if !ok {
 								continue
 							}
 							multiError = errors.Join(multiError, aggregate(dp, filteredDPAttrs))
@@ -199,10 +177,8 @@ func (sm *signalToMetrics) ConsumeMetrics(ctx context.Context, m pmetric.Metrics
 						dps := metric.Histogram().DataPoints()
 						for l := 0; l < dps.Len(); l++ {
 							dp := dps.At(l)
-							filteredDPAttrs := getFilteredAttributes(dp.Attributes(), md.Attributes)
-							if filteredDPAttrs.Len() != len(md.Attributes) {
-								// If all the configured attributes are not present in
-								// source metric then don't count them.
+							filteredDPAttrs, ok := md.FilterAttributes(dp.Attributes())
+							if !ok {
 								continue
 							}
 							multiError = errors.Join(multiError, aggregate(dp, filteredDPAttrs))
@@ -211,10 +187,8 @@ func (sm *signalToMetrics) ConsumeMetrics(ctx context.Context, m pmetric.Metrics
 						dps := metric.ExponentialHistogram().DataPoints()
 						for l := 0; l < dps.Len(); l++ {
 							dp := dps.At(l)
-							filteredDPAttrs := getFilteredAttributes(dp.Attributes(), md.Attributes)
-							if filteredDPAttrs.Len() != len(md.Attributes) {
-								// If all the configured attributes are not present in
-								// source metric then don't count them.
+							filteredDPAttrs, ok := md.FilterAttributes(dp.Attributes())
+							if !ok {
 								continue
 							}
 							multiError = errors.Join(multiError, aggregate(dp, filteredDPAttrs))
@@ -248,10 +222,8 @@ func (sm *signalToMetrics) ConsumeLogs(ctx context.Context, logs plog.Logs) erro
 				log := scopeLog.LogRecords().At(k)
 				logAttrs := log.Attributes()
 				for _, md := range sm.logMetricDefs {
-					filteredLogAttrs := getFilteredAttributes(logAttrs, md.Attributes)
-					if filteredLogAttrs.Len() != len(md.Attributes) {
-						// If all the configured attributes are not present in
-						// source metric then don't count them.
+					filteredLogAttrs, ok := md.FilterAttributes(logAttrs)
+					if !ok {
 						continue
 					}
 
@@ -270,14 +242,7 @@ func (sm *signalToMetrics) ConsumeLogs(ctx context.Context, logs plog.Logs) erro
 						}
 					}
 
-					var filteredResAttrs pcommon.Map
-					if len(md.IncludeResourceAttributes) > 0 {
-						filteredResAttrs = getFilteredAttributes(resourceAttrs, md.IncludeResourceAttributes)
-					} else {
-						// Copy resource attrs to avoid mutating data
-						filteredResAttrs = pcommon.NewMap()
-						resourceAttrs.CopyTo(filteredResAttrs)
-					}
+					filteredResAttrs := md.FilterResourceAttributes(resourceAttrs, sm.collectorInstanceInfo)
 					multiError = errors.Join(multiError, aggregator.Aggregate(ctx, tCtx, md, filteredResAttrs, filteredLogAttrs, 1))
 				}
 			}
@@ -306,17 +271,3 @@ func (sm *signalToMetrics) processNext(ctx context.Context, m pmetric.Metrics, e
 	}
 	return sm.next.ConsumeMetrics(ctx, m)
 }
-
-func getFilteredAttributes(attrs pcommon.Map, filters []model.AttributeKeyValue) pcommon.Map {
-	filteredAttrs := pcommon.NewMap()
-	for _, filter := range filters {
-		if attr, ok := attrs.Get(filter.Key); ok {
-			attr.CopyTo(filteredAttrs.PutEmpty(filter.Key))
-			continue
-		}
-		if filter.DefaultValue.Type() != pcommon.ValueTypeEmpty {
-			filter.DefaultValue.CopyTo(filteredAttrs.PutEmpty(filter.Key))
-		}
-	}
-	return filteredAttrs
-}

connector/signaltometricsconnector/connector_test.go

@@ -35,7 +35,9 @@ import (
 	"go.opentelemetry.io/collector/connector/connectortest"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	semconv "go.opentelemetry.io/collector/semconv/v1.26.0"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest"
 )
@@ -299,6 +301,7 @@ func setupConnector(
 	t.Helper()
 	factory := NewFactory()
 	settings := connectortest.NewNopSettings()
+	telemetryResource(t).CopyTo(settings.TelemetrySettings.Resource)
 	settings.TelemetrySettings.Logger = zaptest.NewLogger(t, zaptest.Level(zapcore.DebugLevel))
 
 	cfg := createDefaultConfig()
@@ -312,6 +315,16 @@ func setupConnector(
 	return factory, settings, cfg
 }
 
+func telemetryResource(t *testing.T) pcommon.Resource {
+	t.Helper()
+
+	r := pcommon.NewResource()
+	r.Attributes().PutStr(semconv.AttributeServiceInstanceID, "627cc493-f310-47de-96bd-71410b7dec09")
+	r.Attributes().PutStr(semconv.AttributeServiceName, "signaltometrics")
+	r.Attributes().PutStr(semconv.AttributeServiceNamespace, "test")
+	return r
+}
+
 func assertAggregatedMetrics(t *testing.T, expected, actual pmetric.Metrics) bool {
 	t.Helper()
 	return assert.NoError(t, pmetrictest.CompareMetrics(

connector/signaltometricsconnector/factory.go

@@ -73,7 +73,10 @@ func createTracesToMetrics(
 	}
 
 	return &signalToMetrics{
-		logger:         set.Logger,
+		logger: set.Logger,
+		collectorInstanceInfo: model.NewCollectorInstanceInfo(
+			set.TelemetrySettings,
+		),
 		next:           nextConsumer,
 		spanMetricDefs: metricDefs,
 	}, nil
@@ -101,7 +104,10 @@ func createMetricsToMetrics(
 	}
 
 	return &signalToMetrics{
-		logger:       set.Logger,
+		logger: set.Logger,
+		collectorInstanceInfo: model.NewCollectorInstanceInfo(
+			set.TelemetrySettings,
+		),
 		next:         nextConsumer,
 		dpMetricDefs: metricDefs,
 	}, nil
@@ -129,7 +135,10 @@ func createLogsToMetrics(
 	}
 
 	return &signalToMetrics{
-		logger:        set.Logger,
+		logger: set.Logger,
+		collectorInstanceInfo: model.NewCollectorInstanceInfo(
+			set.TelemetrySettings,
+		),
 		next:          nextConsumer,
 		logMetricDefs: metricDefs,
 	}, nil

connector/signaltometricsconnector/go.mod

@@ -17,6 +17,7 @@ require (
 	go.opentelemetry.io/collector/consumer/consumertest v0.111.0
 	go.opentelemetry.io/collector/pdata v1.17.0
 	go.opentelemetry.io/collector/pipeline v0.111.0
+	go.opentelemetry.io/collector/semconv v0.111.0
 	go.uber.org/goleak v1.3.0
 	go.uber.org/zap v1.27.0
 )
@@ -58,7 +59,6 @@ require (
 	go.opentelemetry.io/collector/consumer/consumerprofiles v0.111.0 // indirect
 	go.opentelemetry.io/collector/internal/globalsignal v0.111.0 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.111.0 // indirect
-	go.opentelemetry.io/collector/semconv v0.111.0 // indirect
 	go.opentelemetry.io/otel v1.30.0 // indirect
 	go.opentelemetry.io/otel/metric v1.30.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.30.0 // indirect

connector/signaltometricsconnector/internal/model/collectorinstanceinfo.go

@@ -0,0 +1,88 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 model // import "github.com/elastic/opentelemetry-collector-components/connector/signaltometricsconnector/internal/model"
+
+import (
+	"github.com/elastic/opentelemetry-collector-components/connector/signaltometricsconnector/internal/metadata"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	semconv "go.opentelemetry.io/collector/semconv/v1.26.0"
+)
+
+var prefix = metadata.Type.String()
+
+// CollectorInstanceInfo holds the attributes that could uniquely identify
+// the current collector instance. These attributes are initialized from the
+// telemetry settings. The CollectorInstanceInfo can copy these attributes,
+// with a given prefix, to a provided map.
+type CollectorInstanceInfo struct {
+	size              int
+	serviceInstanceID string
+	serviceName       string
+	serviceNamespace  string
+}
+
+func NewCollectorInstanceInfo(
+	set component.TelemetrySettings,
+) *CollectorInstanceInfo {
+	var info CollectorInstanceInfo
+	set.Resource.Attributes().Range(func(k string, v pcommon.Value) bool {
+		switch k {
+		case semconv.AttributeServiceInstanceID:
+			if str := v.Str(); str != "" {
+				info.serviceInstanceID = str
+				info.size++
+			}
+		case semconv.AttributeServiceName:
+			if str := v.Str(); str != "" {
+				info.serviceName = str
+				info.size++
+			}
+		case semconv.AttributeServiceNamespace:
+			if str := v.Str(); str != "" {
+				info.serviceNamespace = str
+				info.size++
+			}
+		}
+		return true
+	})
+	return &info
+}
+
+// Size returns the max number of attributes that defines a collector's
+// instance information. Can be used to presize the attributes.
+func (info CollectorInstanceInfo) Size() int {
+	return info.size
+}
+
+func (info CollectorInstanceInfo) Copy(to pcommon.Map) {
+	to.EnsureCapacity(info.Size())
+	if info.serviceInstanceID != "" {
+		to.PutStr(keyWithPrefix(semconv.AttributeServiceInstanceID), info.serviceInstanceID)
+	}
+	if info.serviceName != "" {
+		to.PutStr(keyWithPrefix(semconv.AttributeServiceName), info.serviceName)
+	}
+	if info.serviceNamespace != "" {
+		to.PutStr(keyWithPrefix(semconv.AttributeServiceNamespace), info.serviceNamespace)
+	}
+}
+
+func keyWithPrefix(key string) string {
+	return prefix + "." + key
+}