receiver/prometheusremotewrite: Implement counter metrics conversion.

This commit implements the addCounterDatapoints function to properly
convert Prometheus counter metrics to OTLP format.

Fixes part of #37277

Signed-off-by: sujal shah <[email protected]>

.chloggen/prometheusremotewrite-counter-metrics-conversion.yaml.yaml

@@ -0,0 +1,30 @@
+# 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: receiver/prometheusremotewrite
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Implement counter metrics conversion from Prometheus Remote Write to OTLP format"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37277]  # Replace with actual PR or issue number
+
+# (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: |
+  The implementation processes scope name and version from labels,
+  maintains proper scope metrics organization, and preserves all attributes
+  while setting appropriate counter-specific properties.
+
+# 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

@@ -193,7 +193,7 @@ func (prw *prometheusRemoteWriteReceiver) translateV2(_ context.Context, req *wr
 
 		switch ts.Metadata.Type {
 		case writev2.Metadata_METRIC_TYPE_COUNTER:
-			addCounterDatapoints(rm, ls, ts)
+			prw.addCounterDatapoints(rm, ls, ts)
 		case writev2.Metadata_METRIC_TYPE_GAUGE:
 			addGaugeDatapoints(rm, ls, ts)
 		case writev2.Metadata_METRIC_TYPE_SUMMARY:
@@ -225,8 +225,44 @@ func parseJobAndInstance(dest pcommon.Map, job, instance string) {
 	}
 }
 
-func addCounterDatapoints(_ pmetric.ResourceMetrics, _ labels.Labels, _ writev2.TimeSeries) {
-	// TODO: Implement this function
+func (prw *prometheusRemoteWriteReceiver) addCounterDatapoints(rm pmetric.ResourceMetrics, ls labels.Labels, ts writev2.TimeSeries) {
+	scopeName := prw.settings.BuildInfo.Description
+	scopeVersion := prw.settings.BuildInfo.Version
+	metricName := ls.Get(labels.MetricName)
+
+	if v := ls.Get("otel_scope_name"); v != "" {
+		scopeName = v
+	}
+	if v := ls.Get("otel_scope_version"); v != "" {
+		scopeVersion = v
+	}
+
+	var scope pmetric.ScopeMetrics
+	scopeFound := false
+	for i := 0; i < rm.ScopeMetrics().Len(); i++ {
+		sm := rm.ScopeMetrics().At(i)
+		if sm.Scope().Name() == scopeName && sm.Scope().Version() == scopeVersion {
+			scope = sm
+			scopeFound = true
+			break
+		}
+	}
+
+	if !scopeFound {
+		scope = rm.ScopeMetrics().AppendEmpty()
+		scope.Scope().SetName(scopeName)
+		scope.Scope().SetVersion(scopeVersion)
+	}
+
+	metric := scope.Metrics().AppendEmpty()
+	metric.SetName(metricName)
+	sum := metric.SetEmptySum()
+	sum.SetIsMonotonic(true)
+	sum.SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
+
+	// Add datapoints
+	datapoints := sum.DataPoints()
+	addDatapoints(datapoints, ls, ts)
 }
 
 func addGaugeDatapoints(rm pmetric.ResourceMetrics, ls labels.Labels, ts writev2.TimeSeries) {

receiver/prometheusremotewritereceiver/receiver_test.go

@@ -134,6 +134,61 @@ func TestTranslateV2(t *testing.T) {
 		expectedMetrics pmetric.Metrics
 		expectedStats   remote.WriteResponseStats
 	}{
+		{
+			name: "counter metric test",
+			request: &writev2.Request{
+				Symbols: []string{
+					"",
+					"__name__", "http_requests_total",
+					"job", "web-service",
+					"instance", "server-001",
+					"method", "GET",
+					"status", "200",
+					"otel_scope_name", "test-scope",
+					"otel_scope_version", "v1",
+				},
+				Timeseries: []writev2.TimeSeries{
+					{
+						Metadata:   writev2.Metadata{Type: writev2.Metadata_METRIC_TYPE_COUNTER},
+						LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14},
+						Samples: []writev2.Sample{
+							{Value: 100, Timestamp: 1613000000000},
+							{Value: 150, Timestamp: 1613001000000},
+						},
+					},
+				},
+			},
+			expectedMetrics: func() pmetric.Metrics {
+				expected := pmetric.NewMetrics()
+				rm := expected.ResourceMetrics().AppendEmpty()
+				rm.Resource().Attributes().PutStr("service.name", "web-service")
+				rm.Resource().Attributes().PutStr("service.instance.id", "server-001")
+
+				sm := rm.ScopeMetrics().AppendEmpty()
+				sm.Scope().SetName("test-scope")
+				sm.Scope().SetVersion("v1")
+
+				metric := sm.Metrics().AppendEmpty()
+				metric.SetName("http_requests_total")
+				metric.SetEmptySum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
+				metric.Sum().SetIsMonotonic(true)
+
+				dp1 := metric.Sum().DataPoints().AppendEmpty()
+				dp1.SetTimestamp(pcommon.Timestamp(1613000000000 * 1e6))
+				dp1.SetDoubleValue(100.0)
+				dp1.Attributes().PutStr("method", "GET")
+				dp1.Attributes().PutStr("status", "200")
+
+				dp2 := metric.Sum().DataPoints().AppendEmpty()
+				dp2.SetTimestamp(pcommon.Timestamp(1613001000000 * 1e6))
+				dp2.SetDoubleValue(150.0)
+				dp2.Attributes().PutStr("method", "GET")
+				dp2.Attributes().PutStr("status", "200")
+
+				return expected
+			}(),
+			expectedStats: remote.WriteResponseStats{},
+		},
 		{
 			name: "duplicated scope name and version",
 			request: &writev2.Request{