Change otlp attribute conversion to consist with prometheus

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

CHANGELOG.md

@@ -9,6 +9,7 @@
 * [FEATURE] Ruler: Minimize chances of missed rule group evaluations that can occur due to OOM kills, bad underlying nodes, or due to an unhealthy ruler that appears in the ring as healthy. This feature is enabled via `-ruler.enable-ha-evaluation` flag. #6129
 * [FEATURE] Store Gateway: Add an in-memory chunk cache. #6245
 * [FEATURE] Chunk Cache: Support multi level cache and add metrics. #6249
+* [ENHANCEMENT] OTLP: Change otlp handler to consist with the Prometheus otlp handler. #6272
 * [ENHANCEMENT] Ingester: Add `blocks-storage.tsdb.wal-compression-type` to support zstd wal compression type. #6232
 * [ENHANCEMENT] Query Frontend: Add info field to query response. #6207
 * [ENHANCEMENT] Query Frontend: Add peakSample in query stats response. #6188

docs/configuration/config-file-reference.md

@@ -2608,6 +2608,15 @@ instance_limits:
   # unlimited.
   # CLI flag: -distributor.instance-limits.max-inflight-push-requests
   [max_inflight_push_requests: <int> | default = 0]
+
+otlp:
+  # If enabled, all resource attributes are converted to labels.
+  # CLI flag: -distributor.otlp-config.convert-all-attributes
+  [convert_all_attributes: <boolean> | default = false]
+
+  # A list of resource attributes that should be converted to labels.
+  # CLI flag: -distributor.otlp-config.promote-resource-attributes
+  [promote_resource_attributes: <list of string> | default = []]
 ```
 
 ### `etcd_config`

pkg/api/api.go

@@ -277,7 +277,7 @@ func (a *API) RegisterDistributor(d *distributor.Distributor, pushConfig distrib
 	distributorpb.RegisterDistributorServer(a.server.GRPC, d)
 
 	a.RegisterRoute("/api/v1/push", push.Handler(pushConfig.MaxRecvMsgSize, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST")
-	a.RegisterRoute("/api/v1/otlp/v1/metrics", push.OTLPHandler(a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST")
+	a.RegisterRoute("/api/v1/otlp/v1/metrics", push.OTLPHandler(pushConfig.OTLPConfig, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST")
 
 	a.indexPage.AddLink(SectionAdminEndpoints, "/distributor/ring", "Distributor Ring Status")
 	a.indexPage.AddLink(SectionAdminEndpoints, "/distributor/all_user_stats", "Usage Statistics")

pkg/distributor/distributor.go

@@ -36,6 +36,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/extract"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/limiter"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	util_math "github.com/cortexproject/cortex/pkg/util/math"
@@ -164,13 +165,21 @@ type Config struct {
 
 	// Limits for distributor
 	InstanceLimits InstanceLimits `yaml:"instance_limits"`
+
+	// OTLPConfig
+	OTLPConfig OTLPConfig `yaml:"otlp"`
 }
 
 type InstanceLimits struct {
 	MaxIngestionRate        float64 `yaml:"max_ingestion_rate"`
 	MaxInflightPushRequests int     `yaml:"max_inflight_push_requests"`
 }
 
+type OTLPConfig struct {
+	ConvertAllAttributes      bool     `yaml:"convert_all_attributes"`
+	PromoteResourceAttributes []string `yaml:"promote_resource_attributes"`
+}
+
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.PoolConfig.RegisterFlags(f)
@@ -188,6 +197,9 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 
 	f.Float64Var(&cfg.InstanceLimits.MaxIngestionRate, "distributor.instance-limits.max-ingestion-rate", 0, "Max ingestion rate (samples/sec) that this distributor will accept. This limit is per-distributor, not per-tenant. Additional push requests will be rejected. Current ingestion rate is computed as exponentially weighted moving average, updated every second. 0 = unlimited.")
 	f.IntVar(&cfg.InstanceLimits.MaxInflightPushRequests, "distributor.instance-limits.max-inflight-push-requests", 0, "Max inflight push requests that this distributor can handle. This limit is per-distributor, not per-tenant. Additional requests will be rejected. 0 = unlimited.")
+
+	f.BoolVar(&cfg.OTLPConfig.ConvertAllAttributes, "distributor.otlp-config.convert-all-attributes", false, "If enabled, all resource attributes are converted to labels.")
+	f.Var((*flagext.StringSlice)(&cfg.OTLPConfig.PromoteResourceAttributes), "distributor.otlp-config.promote-resource-attributes", "A list of resource attributes that should be converted to labels.")
 }
 
 // Validate config and returns error on failure

pkg/util/push/otlp.go

@@ -1,33 +1,37 @@
 package push
 
 import (
+	"context"
 	"net/http"
 
+	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/prometheus/prometheus/storage/remote"
 	"github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheusremotewrite"
+	"github.com/prometheus/prometheus/util/annotations"
 	"github.com/weaveworks/common/httpgrpc"
 	"github.com/weaveworks/common/middleware"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
+	"github.com/cortexproject/cortex/pkg/distributor"
 	"github.com/cortexproject/cortex/pkg/util"
-	"github.com/cortexproject/cortex/pkg/util/log"
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
 
 // OTLPHandler is a http.Handler which accepts OTLP metrics.
-func OTLPHandler(sourceIPs *middleware.SourceIPExtractor, push Func) http.Handler {
+func OTLPHandler(cfg distributor.OTLPConfig, sourceIPs *middleware.SourceIPExtractor, push Func) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
-		logger := log.WithContext(ctx, log.Logger)
+		logger := util_log.WithContext(ctx, util_log.Logger)
 		if sourceIPs != nil {
 			source := sourceIPs.Get(r)
 			if source != "" {
 				ctx = util.AddSourceIPsToOutgoingContext(ctx, source)
-				logger = log.WithSourceIPs(source, logger)
+				logger = util_log.WithSourceIPs(source, logger)
 			}
 		}
 		req, err := remote.DecodeOTLPWriteRequest(r)
@@ -37,31 +41,22 @@ func OTLPHandler(sourceIPs *middleware.SourceIPExtractor, push Func) http.Handle
 			return
 		}
 
-		promConverter := prometheusremotewrite.NewPrometheusConverter()
-		setting := prometheusremotewrite.Settings{
-			AddMetricSuffixes: true,
-			DisableTargetInfo: true,
-		}
-		annots, err := promConverter.FromMetrics(ctx, convertToMetricsAttributes(req.Metrics()), setting)
-		ws, _ := annots.AsStrings("", 0, 0)
-		if len(ws) > 0 {
-			level.Warn(logger).Log("msg", "Warnings translating OTLP metrics to Prometheus write request", "warnings", ws)
+		prwReq := cortexpb.WriteRequest{
+			Source:                  cortexpb.API,
+			Metadata:                nil,
+			SkipLabelNameValidation: false,
 		}
 
+		// otlp to prompb TimeSeries
+		promTsList, err := convertToPromTS(r.Context(), req.Metrics(), cfg, logger)
 		if err != nil {
-			level.Error(logger).Log("msg", "Error translating OTLP metrics to Prometheus write request", "err", err)
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
 
-		prwReq := cortexpb.WriteRequest{
-			Source:                  cortexpb.API,
-			Metadata:                nil,
-			SkipLabelNameValidation: false,
-		}
-
+		// convert prompb to cortexpb TimeSeries
 		tsList := []cortexpb.PreallocTimeseries(nil)
-		for _, v := range promConverter.TimeSeries() {
+		for _, v := range promTsList {
 			tsList = append(tsList, cortexpb.PreallocTimeseries{TimeSeries: &cortexpb.TimeSeries{
 				Labels:     makeLabels(v.Labels),
 				Samples:    makeSamples(v.Samples),
@@ -87,6 +82,32 @@ func OTLPHandler(sourceIPs *middleware.SourceIPExtractor, push Func) http.Handle
 	})
 }
 
+func convertToPromTS(ctx context.Context, pmetrics pmetric.Metrics, cfg distributor.OTLPConfig, logger log.Logger) ([]prompb.TimeSeries, error) {
+	promConverter := prometheusremotewrite.NewPrometheusConverter()
+	settings := prometheusremotewrite.Settings{
+		AddMetricSuffixes:         true,
+		PromoteResourceAttributes: cfg.PromoteResourceAttributes,
+	}
+	var annots annotations.Annotations
+	var err error
+	if cfg.ConvertAllAttributes {
+		annots, err = promConverter.FromMetrics(ctx, convertToMetricsAttributes(pmetrics), settings)
+	} else {
+		annots, err = promConverter.FromMetrics(ctx, pmetrics, settings)
+	}
+
+	ws, _ := annots.AsStrings("", 0, 0)
+	if len(ws) > 0 {
+		level.Warn(logger).Log("msg", "Warnings translating OTLP metrics to Prometheus write request", "warnings", ws)
+	}
+
+	if err != nil {
+		level.Error(logger).Log("msg", "Error translating OTLP metrics to Prometheus write request", "err", err)
+		return nil, err
+	}
+	return promConverter.TimeSeries(), nil
+}
+
 func makeLabels(in []prompb.Label) []cortexpb.LabelAdapter {
 	out := make(labels.Labels, 0, len(in))
 	for _, l := range in {

pkg/util/push/otlp_test.go

@@ -8,16 +8,179 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-kit/log"
+	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
 
 	"github.com/cortexproject/cortex/pkg/cortexpb"
+	"github.com/cortexproject/cortex/pkg/distributor"
 )
 
+func TestOTLPConvertToPromTS(t *testing.T) {
+	logger := log.NewNopLogger()
+	ctx := context.Background()
+	d := pmetric.NewMetrics()
+	resourceMetric := d.ResourceMetrics().AppendEmpty()
+	resourceMetric.Resource().Attributes().PutStr("service.name", "test-service") // converted to job, service_name
+	resourceMetric.Resource().Attributes().PutStr("attr1", "value")
+	resourceMetric.Resource().Attributes().PutStr("attr2", "value")
+	resourceMetric.Resource().Attributes().PutStr("attr3", "value")
+
+	scopeMetric := resourceMetric.ScopeMetrics().AppendEmpty()
+
+	//Generate One Counter
+	timestamp := time.Now()
+	counterMetric := scopeMetric.Metrics().AppendEmpty()
+	counterMetric.SetName("test-counter")
+	counterMetric.SetDescription("test-counter-description")
+	counterMetric.SetEmptySum()
+	counterMetric.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative)
+	counterMetric.Sum().SetIsMonotonic(true)
+
+	counterDataPoint := counterMetric.Sum().DataPoints().AppendEmpty()
+	counterDataPoint.SetTimestamp(pcommon.NewTimestampFromTime(timestamp))
+	counterDataPoint.SetDoubleValue(10.0)
+
+	tests := []struct {
+		description    string
+		cfg            distributor.OTLPConfig
+		expectedLabels []prompb.Label
+	}{
+		{
+			description: "only attributes that exist in promote resource attributes should be converted",
+			cfg: distributor.OTLPConfig{
+				ConvertAllAttributes:      false,
+				PromoteResourceAttributes: []string{"attr1"},
+			},
+			expectedLabels: []prompb.Label{
+				{
+					Name:  "__name__",
+					Value: "test_counter_total",
+				},
+				{
+					Name:  "attr1",
+					Value: "value",
+				},
+				{
+					Name:  "job",
+					Value: "test-service",
+				},
+			},
+		},
+		{
+			description: "not exist attribute is ignored",
+			cfg: distributor.OTLPConfig{
+				ConvertAllAttributes:      false,
+				PromoteResourceAttributes: []string{"dummy"},
+			},
+			expectedLabels: []prompb.Label{
+				{
+					Name:  "__name__",
+					Value: "test_counter_total",
+				},
+				{
+					Name:  "job",
+					Value: "test-service",
+				},
+			},
+		},
+		{
+			description: "should convert all attribute",
+			cfg: distributor.OTLPConfig{
+				ConvertAllAttributes:      true,
+				PromoteResourceAttributes: nil,
+			},
+			expectedLabels: []prompb.Label{
+				{
+					Name:  "__name__",
+					Value: "test_counter_total",
+				},
+				{
+					Name:  "attr1",
+					Value: "value",
+				},
+				{
+					Name:  "attr2",
+					Value: "value",
+				},
+				{
+					Name:  "attr3",
+					Value: "value",
+				},
+				{
+					Name:  "job",
+					Value: "test-service",
+				},
+				{
+					Name:  "service_name",
+					Value: "test-service",
+				},
+			},
+		},
+		{
+			description: "should convert all attribute regardless of promote resource attributes",
+			cfg: distributor.OTLPConfig{
+				ConvertAllAttributes:      true,
+				PromoteResourceAttributes: []string{"attr1", "attr2"},
+			},
+			expectedLabels: []prompb.Label{
+				{
+					Name:  "__name__",
+					Value: "test_counter_total",
+				},
+				{
+					Name:  "attr1",
+					Value: "value",
+				},
+				{
+					Name:  "attr2",
+					Value: "value",
+				},
+				{
+					Name:  "attr3",
+					Value: "value",
+				},
+				{
+					Name:  "job",
+					Value: "test-service",
+				},
+				{
+					Name:  "service_name",
+					Value: "test-service",
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.description, func(t *testing.T) {
+			tsList, err := convertToPromTS(ctx, d, test.cfg, logger)
+			require.NoError(t, err)
+			require.Equal(t, 2, len(tsList)) // target_info + test_counter_total
+
+			var counterTs prompb.TimeSeries
+			for _, ts := range tsList {
+				for _, label := range ts.Labels {
+					if label.Name == "__name__" && label.Value == "test_counter_total" {
+						// get counter ts
+						counterTs = ts
+					}
+				}
+			}
+			require.ElementsMatch(t, test.expectedLabels, counterTs.Labels)
+		})
+	}
+}
+
 func TestOTLPWriteHandler(t *testing.T) {
+	cfg := distributor.OTLPConfig{
+		PromoteResourceAttributes: []string{},
+	}
+
 	exportRequest := generateOTLPWriteRequest(t)
 
 	buf, err := exportRequest.MarshalProto()
@@ -28,7 +191,7 @@ func TestOTLPWriteHandler(t *testing.T) {
 	req.Header.Set("Content-Type", "application/x-protobuf")
 
 	push := verifyOTLPWriteRequestHandler(t, cortexpb.API)
-	handler := OTLPHandler(nil, push)
+	handler := OTLPHandler(cfg, nil, push)
 
 	recorder := httptest.NewRecorder()
 	handler.ServeHTTP(recorder, req)
@@ -120,7 +283,7 @@ func generateOTLPWriteRequest(t *testing.T) pmetricotlp.ExportRequest {
 func verifyOTLPWriteRequestHandler(t *testing.T, expectSource cortexpb.WriteRequest_SourceEnum) func(ctx context.Context, request *cortexpb.WriteRequest) (response *cortexpb.WriteResponse, err error) {
 	t.Helper()
 	return func(ctx context.Context, request *cortexpb.WriteRequest) (response *cortexpb.WriteResponse, err error) {
-		assert.Len(t, request.Timeseries, 12) // 1 (counter) + 1 (gauge) + 7 (hist_bucket) + 2 (hist_sum, hist_count) + 1 (exponential histogram)
+		assert.Len(t, request.Timeseries, 13) // 1 (target_info) + 1 (counter) + 1 (gauge) + 7 (hist_bucket) + 2 (hist_sum, hist_count) + 1 (exponential histogram)
 		// TODO: test more things
 		assert.Equal(t, expectSource, request.Source)
 		assert.False(t, request.SkipLabelNameValidation)