[receiver/prometheusremotewrite]: handle all otel_scope_* labels per compatibility spec

Author: himanshu130700

.chloggen/prw-receiver-otel-scope-handling.yaml

@@ -0,0 +1,28 @@
+# 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. receiver/filelog)
+component: receiver/prometheus_remote_write
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Handle all `otel_scope_*` prefixed labels per the Prometheus/OTLP compatibility spec."
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [47726]
+
+# (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: |
+  `otel_scope_schema_url` is now set as the instrumentation scope schema URL, and other `otel_scope_<attr>` labels become scope attributes (with the `otel_scope_` prefix stripped), instead of being incorrectly added as metric data point attributes.
+
+# 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]

receiver/prometheusremotewritereceiver/receiver.go

@@ -69,38 +69,50 @@ type prometheusRemoteWriteReceiver struct {
 	bodyBufferPool *sync.Pool
 }
 
-// metricIdentity contains all the components that uniquely identify a metric
-// according to the OpenTelemetry Protocol data model.
-// The definition of the metric uniqueness is based on the following document. Ref: https://opentelemetry.io/docs/specs/otel/metrics/data-model/#opentelemetry-protocol-data-model
+// scopeInfo holds instrumentation scope fields extracted from otel_scope_* labels.
+type scopeInfo struct {
+	Name       string
+	Version    string
+	SchemaURL  string
+	extraAttrs [][2]string // scope attributes with the "otel_scope_" prefix stripped
+}
+
+func (si scopeInfo) key() string {
+	const sep = "\xff"
+	parts := make([]string, 0, 3+len(si.extraAttrs))
+	parts = append(parts, si.Name, si.Version, si.SchemaURL)
+	for _, kv := range si.extraAttrs {
+		parts = append(parts, kv[0]+sep+kv[1])
+	}
+	return strings.Join(parts, sep)
+}
+
+// metricIdentity uniquely identifies a metric per the OTLP data model.
+// Ref: https://opentelemetry.io/docs/specs/otel/metrics/data-model/#opentelemetry-protocol-data-model
 type metricIdentity struct {
-	ResourceID   string
-	ScopeName    string
-	ScopeVersion string
-	MetricName   string
-	Unit         string
-	Type         writev2.Metadata_MetricType
+	ResourceID string
+	ScopeKey   string
+	MetricName string
+	Unit       string
+	Type       writev2.Metadata_MetricType
 }
 
-// createMetricIdentity creates a metricIdentity struct from the required components
-func createMetricIdentity(resourceID, scopeName, scopeVersion, metricName, unit string, metricType writev2.Metadata_MetricType) metricIdentity {
+func createMetricIdentity(resourceID string, si scopeInfo, metricName, unit string, metricType writev2.Metadata_MetricType) metricIdentity {
 	return metricIdentity{
-		ResourceID:   resourceID,
-		ScopeName:    scopeName,
-		ScopeVersion: scopeVersion,
-		MetricName:   metricName,
-		Unit:         unit,
-		Type:         metricType,
+		ResourceID: resourceID,
+		ScopeKey:   si.key(),
+		MetricName: metricName,
+		Unit:       unit,
+		Type:       metricType,
 	}
 }
 
-// Hash generates a unique hash for the metric identity
 func (mi metricIdentity) Hash() uint64 {
 	const separator = "\xff"
 
 	combined := strings.Join([]string{
 		mi.ResourceID,
-		mi.ScopeName,
-		mi.ScopeVersion,
+		mi.ScopeKey,
 		mi.MetricName,
 		mi.Unit,
 		fmt.Sprintf("%d", mi.Type),
@@ -348,7 +360,7 @@ func (prw *prometheusRemoteWriteReceiver) translateV2(_ context.Context, req *wr
 			continue
 		}
 
-		scopeName, scopeVersion := prw.extractScopeInfo(ls)
+		si := prw.extractScopeInfo(ls)
 		metricName := metadata.Name
 		if ts.Metadata.UnitRef >= uint32(len(req.Symbols)) {
 			badRequestErrors = errors.Join(badRequestErrors, fmt.Errorf("unit ref %d is out of bounds of symbolsTable", ts.Metadata.UnitRef))
@@ -366,40 +378,38 @@ func (prw *prometheusRemoteWriteReceiver) translateV2(_ context.Context, req *wr
 		// Handle histograms separately due to their complex mixed-schema processing
 		if ts.Metadata.Type == writev2.Metadata_METRIC_TYPE_HISTOGRAM ||
 			ts.Metadata.Type == writev2.Metadata_METRIC_TYPE_UNSPECIFIED && len(ts.Histograms) > 0 {
-			prw.processHistogramTimeSeries(otelMetrics, ls, ts, scopeName, scopeVersion, metricName, unit, description, metricCache, &stats, modifiedResourceMetric, exemplarMap)
+			prw.processHistogramTimeSeries(otelMetrics, ls, ts, si, metricName, unit, description, metricCache, &stats, modifiedResourceMetric, exemplarMap)
 			continue
 		}
 
 		// Handle regular metrics (gauge, counter, summary)
 		rm, _ := prw.getOrCreateRM(ls, otelMetrics, modifiedResourceMetric)
 
 		resourceID := identity.OfResource(rm.Resource())
-		metricIdentity := createMetricIdentity(
+		metricID := createMetricIdentity(
 			resourceID.String(), // Resource identity
-			scopeName,           // Scope name
-			scopeVersion,        // Scope version
+			si,                  // Scope info
 			metricName,          // Metric name
 			unit,                // Unit
 			ts.Metadata.Type,    // Metric type
 		)
 
-		metricKey := metricIdentity.Hash()
+		metricKey := metricID.Hash()
 
 		// Find or create scope
 		var scope pmetric.ScopeMetrics
 		var foundScope bool
 		for i := 0; i < rm.ScopeMetrics().Len(); i++ {
 			s := rm.ScopeMetrics().At(i)
-			if s.Scope().Name() == scopeName && s.Scope().Version() == scopeVersion {
+			if scopeMatchesInfo(s, si) {
 				scope = s
 				foundScope = true
 				break
 			}
 		}
 		if !foundScope {
 			scope = rm.ScopeMetrics().AppendEmpty()
-			scope.Scope().SetName(scopeName)
-			scope.Scope().SetVersion(scopeVersion)
+			applyScopeInfo(scope, si)
 		}
 
 		// Get or create metric
@@ -440,8 +450,8 @@ func (prw *prometheusRemoteWriteReceiver) translateV2(_ context.Context, req *wr
 		case writev2.Metadata_METRIC_TYPE_COUNTER:
 			addNumberDatapoints(metric.Sum().DataPoints(), ls, ts, &stats)
 			key := exemplarKey{
-				ScopeName:    scopeName,
-				ScopeVersion: scopeVersion,
+				ScopeName:    si.Name,
+				ScopeVersion: si.Version,
 				MetricName:   metricName,
 				MetricType:   ts.Metadata.Type,
 			}
@@ -466,7 +476,8 @@ func (prw *prometheusRemoteWriteReceiver) processHistogramTimeSeries(
 	otelMetrics pmetric.Metrics,
 	ls labels.Labels,
 	ts *writev2.TimeSeries,
-	scopeName, scopeVersion, metricName, unit, description string,
+	si scopeInfo,
+	metricName, unit, description string,
 	metricCache map[uint64]pmetric.Metric,
 	stats *promremote.WriteResponseStats,
 	modifiedRM map[uint64]pmetric.ResourceMetrics,
@@ -516,22 +527,20 @@ func (prw *prometheusRemoteWriteReceiver) processHistogramTimeSeries(
 		var foundScope bool
 		for i := 0; i < rm.ScopeMetrics().Len(); i++ {
 			s := rm.ScopeMetrics().At(i)
-			if s.Scope().Name() == scopeName && s.Scope().Version() == scopeVersion {
+			if scopeMatchesInfo(s, si) {
 				scope = s
 				foundScope = true
 				break
 			}
 		}
 		if !foundScope {
 			scope = rm.ScopeMetrics().AppendEmpty()
-			scope.Scope().SetName(scopeName)
-			scope.Scope().SetVersion(scopeVersion)
+			applyScopeInfo(scope, si)
 		}
 
-		metricID := fmt.Sprintf("%s:%s:%s:%s:%s:%s:%s",
+		metricID := fmt.Sprintf("%s:%s:%s:%s:%s:%s",
 			resourceID.String(),
-			scopeName,
-			scopeVersion,
+			si.key(),
 			metricName,
 			unit,
 			fmt.Sprintf("%d", ts.Metadata.Type),
@@ -581,8 +590,8 @@ func (prw *prometheusRemoteWriteReceiver) processHistogramTimeSeries(
 		}
 
 		key := exemplarKey{
-			ScopeName:    scopeName,
-			ScopeVersion: scopeVersion,
+			ScopeName:    si.Name,
+			ScopeVersion: si.Version,
 			MetricName:   metricName,
 			MetricType:   ts.Metadata.Type,
 		}
@@ -786,35 +795,74 @@ func convertAbsoluteBuckets(spans []writev2.BucketSpan, counts []float64, bucket
 	}
 }
 
-// extractAttributes return all attributes different from job, instance, metric name and scope name/version
+// extractAttributes returns metric data point attributes, excluding job, instance, metric name, and all otel_scope_* labels.
 func extractAttributes(ls labels.Labels) pcommon.Map {
 	attrs := pcommon.NewMap()
-	// job, instance and metric name will always become labels
 	attrs.EnsureCapacity(ls.Len() - 3)
 	ls.Range(func(l labels.Label) {
 		if l.Name != "instance" && l.Name != "job" && // Become resource attributes
 			l.Name != model.MetricNameLabel && // Becomes metric name
-			l.Name != "otel_scope_name" && l.Name != "otel_scope_version" { // Becomes scope name and version
+			!strings.HasPrefix(l.Name, "otel_scope_") { // Become instrumentation scope fields
 			attrs.PutStr(l.Name, l.Value)
 		}
 	})
 	return attrs
 }
 
-// extractScopeInfo extracts the scope name and version from the labels. If the labels do not contain the scope name/version,
-// it will use the default values from the settings.
-func (prw *prometheusRemoteWriteReceiver) extractScopeInfo(ls labels.Labels) (string, string) {
-	scopeName := prw.settings.BuildInfo.Description
-	scopeVersion := prw.settings.BuildInfo.Version
+// extractScopeInfo extracts all otel_scope_* labels into a scopeInfo per the Prometheus/OTLP compatibility spec.
+// Falls back to receiver build info when otel_scope_name is absent.
+func (prw *prometheusRemoteWriteReceiver) extractScopeInfo(ls labels.Labels) scopeInfo {
+	si := scopeInfo{
+		Name:    prw.settings.BuildInfo.Description,
+		Version: prw.settings.BuildInfo.Version,
+	}
+
+	ls.Range(func(l labels.Label) {
+		switch l.Name {
+		case "otel_scope_name":
+			if l.Value != "" {
+				si.Name = l.Value
+			}
+		case "otel_scope_version":
+			if l.Value != "" {
+				si.Version = l.Value
+			}
+		case "otel_scope_schema_url":
+			si.SchemaURL = l.Value
+		default:
+			if strings.HasPrefix(l.Name, "otel_scope_") {
+				attrKey := strings.TrimPrefix(l.Name, "otel_scope_")
+				si.extraAttrs = append(si.extraAttrs, [2]string{attrKey, l.Value})
+			}
+		}
+	})
+
+	return si
+}
 
-	if sName := ls.Get("otel_scope_name"); sName != "" {
-		scopeName = sName
+func scopeMatchesInfo(sm pmetric.ScopeMetrics, si scopeInfo) bool {
+	if sm.Scope().Name() != si.Name || sm.Scope().Version() != si.Version || sm.SchemaUrl() != si.SchemaURL {
+		return false
+	}
+	if sm.Scope().Attributes().Len() != len(si.extraAttrs) {
+		return false
 	}
+	for _, kv := range si.extraAttrs {
+		v, ok := sm.Scope().Attributes().Get(kv[0])
+		if !ok || v.Str() != kv[1] {
+			return false
+		}
+	}
+	return true
+}
 
-	if sVersion := ls.Get("otel_scope_version"); sVersion != "" {
-		scopeVersion = sVersion
+func applyScopeInfo(sm pmetric.ScopeMetrics, si scopeInfo) {
+	sm.Scope().SetName(si.Name)
+	sm.Scope().SetVersion(si.Version)
+	sm.SetSchemaUrl(si.SchemaURL)
+	for _, kv := range si.extraAttrs {
+		sm.Scope().Attributes().PutStr(kv[0], kv[1])
 	}
-	return scopeName, scopeVersion
 }
 
 // addNHCBDatapoint converts a single Native Histogram Custom Buckets (NHCB) to OpenTelemetry histogram datapoints