Add EnableTypeAndUnitLabels flag to add type and unit label

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

Author: SungJin1212

CHANGELOG.md

@@ -4,6 +4,7 @@
 * [CHANGE] StoreGateway/Alertmanager: Add default 5s connection timeout on client. #6603
 * [CHANGE] Ingester: Remove EnableNativeHistograms config flag and instead gate keep through new per-tenant limit at ingestion. #6718
 * [CHANGE] Validate a tenantID when to use a single tenant resolver. #6727
+* [FEATURE] Distributor: Add an experimental `-distributor.otlp.enable-type-and-unit-labels` flag to add `__type__` and `__unit__` labels for OTLP metrics. #6969 
 * [FEATURE] Distributor: Add an experimental `-distributor.otlp.allow-delta-temporality` flag to ingest delta temporality otlp metrics. #6934
 * [FEATURE] Query Frontend: Add dynamic interval size for query splitting. This is enabled by configuring experimental flags `querier.max-shards-per-query` and/or `querier.max-fetched-data-duration-per-query`. The split interval size is dynamically increased to maintain a number of shards and total duration fetched below the configured values. #6458
 * [FEATURE] Querier/Ruler: Add `query_partial_data` and `rules_partial_data` limits to allow queries/rules to be evaluated with data from a single zone, if other zones are not available. #6526

docs/configuration/config-file-reference.md

@@ -3265,6 +3265,11 @@ otlp:
   # EXPERIMENTAL: If true, delta temporality otlp metrics to be ingested.
   # CLI flag: -distributor.otlp.allow-delta-temporality
   [allow_delta_temporality: <boolean> | default = false]
+
+  # EXPERIMENTAL: If true, the '__type__' and '__unit__' labels are added for
+  # the OTLP metrics.
+  # CLI flag: -distributor.otlp.enable-type-and-unit-labels
+  [enable_type_and_unit_labels: <boolean> | default = false]
 ```
 
 ### `etcd_config`

docs/configuration/v1-guarantees.md

@@ -118,6 +118,7 @@ Currently experimental features are:
   - `alertmanager-sharding-ring.final-sleep` (duration) CLI flag
 - OTLP Receiver
   - Ingest delta temporality OTLP metrics (`-distributor.otlp.allow-delta-temporality=true`)
+  - Add `__type__` and `__unit__` labels (`-distributor.otlp.enable-type-and-unit-labels`)
 - Persistent tokens in the Ruler Ring:
   - `-ruler.ring.tokens-file-path` (path) CLI flag
 - Native Histograms

pkg/distributor/distributor.go

@@ -193,9 +193,10 @@ type InstanceLimits struct {
 }
 
 type OTLPConfig struct {
-	ConvertAllAttributes  bool `yaml:"convert_all_attributes"`
-	DisableTargetInfo     bool `yaml:"disable_target_info"`
-	AllowDeltaTemporality bool `yaml:"allow_delta_temporality"`
+	ConvertAllAttributes    bool `yaml:"convert_all_attributes"`
+	DisableTargetInfo       bool `yaml:"disable_target_info"`
+	AllowDeltaTemporality   bool `yaml:"allow_delta_temporality"`
+	EnableTypeAndUnitLabels bool `yaml:"enable_type_and_unit_labels"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
@@ -224,6 +225,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&cfg.OTLPConfig.ConvertAllAttributes, "distributor.otlp.convert-all-attributes", false, "If true, all resource attributes are converted to labels.")
 	f.BoolVar(&cfg.OTLPConfig.DisableTargetInfo, "distributor.otlp.disable-target-info", false, "If true, a target_info metric is not ingested. (refer to: https://github.com/prometheus/OpenMetrics/blob/main/specification/OpenMetrics.md#supporting-target-metadata-in-both-push-based-and-pull-based-systems)")
 	f.BoolVar(&cfg.OTLPConfig.AllowDeltaTemporality, "distributor.otlp.allow-delta-temporality", false, "EXPERIMENTAL: If true, delta temporality otlp metrics to be ingested.")
+	f.BoolVar(&cfg.OTLPConfig.EnableTypeAndUnitLabels, "distributor.otlp.enable-type-and-unit-labels", false, "EXPERIMENTAL: If true, the '__type__' and '__unit__' labels are added for the OTLP metrics.")
 }
 
 // Validate config and returns error on failure

pkg/util/push/otlp.go

@@ -178,9 +178,10 @@ func decodeOTLPWriteRequest(ctx context.Context, r *http.Request, maxSize int) (
 func convertToPromTS(ctx context.Context, pmetrics pmetric.Metrics, cfg distributor.OTLPConfig, overrides *validation.Overrides, userID string, logger log.Logger) ([]prompb.TimeSeries, []prompb.MetricMetadata, error) {
 	promConverter := prometheusremotewrite.NewPrometheusConverter()
 	settings := prometheusremotewrite.Settings{
-		AddMetricSuffixes:     true,
-		DisableTargetInfo:     cfg.DisableTargetInfo,
-		AllowDeltaTemporality: cfg.AllowDeltaTemporality,
+		AddMetricSuffixes:       true,
+		DisableTargetInfo:       cfg.DisableTargetInfo,
+		AllowDeltaTemporality:   cfg.AllowDeltaTemporality,
+		EnableTypeAndUnitLabels: cfg.EnableTypeAndUnitLabels,
 	}
 
 	var annots annotations.Annotations

pkg/util/push/otlp_test.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/go-kit/log"
+	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -27,6 +28,80 @@ import (
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
+func TestOTLP_EnableTypeAndUnitLabels(t *testing.T) {
+	logger := log.NewNopLogger()
+	ctx := context.Background()
+	ts := time.Now()
+
+	tests := []struct {
+		description             string
+		enableTypeAndUnitLabels bool
+		allowDeltaTemporality   bool
+		otlpSeries              pmetric.Metric
+		expectedLabels          labels.Labels
+		expectedMetadata        prompb.MetricMetadata
+	}{
+		{
+			description:             "[enableTypeAndUnitLabels: true], the '__type__' label should be attached with the gauge type",
+			enableTypeAndUnitLabels: true,
+			otlpSeries:              createOtelSum("test", "seconds", pmetric.AggregationTemporalityCumulative, ts),
+			expectedLabels: labels.FromMap(map[string]string{
+				"__name__":   "test_seconds",
+				"__type__":   "gauge",
+				"__unit__":   "seconds",
+				"test_label": "test_value",
+			}),
+			expectedMetadata: createPromMetadata("test_seconds", "seconds", prompb.MetricMetadata_GAUGE),
+		},
+		{
+			description:             "[enableTypeAndUnitLabels: true], the '__type__' label should not be attached with the unknown type ",
+			enableTypeAndUnitLabels: true,
+			allowDeltaTemporality:   true,
+			otlpSeries:              createOtelSum("test", "seconds", pmetric.AggregationTemporalityDelta, ts),
+			expectedLabels: labels.FromMap(map[string]string{
+				"__name__":   "test_seconds",
+				"__unit__":   "seconds",
+				"test_label": "test_value",
+			}),
+			expectedMetadata: createPromMetadata("test_seconds", "seconds", prompb.MetricMetadata_UNKNOWN),
+		},
+		{
+			description:             "[enableTypeAndUnitLabels: false]",
+			enableTypeAndUnitLabels: false,
+			otlpSeries:              createOtelSum("test", "seconds", pmetric.AggregationTemporalityCumulative, ts),
+			expectedLabels: labels.FromMap(map[string]string{
+				"__name__":   "test_seconds",
+				"test_label": "test_value",
+			}),
+			expectedMetadata: createPromMetadata("test_seconds", "seconds", prompb.MetricMetadata_GAUGE),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.description, func(t *testing.T) {
+			cfg := distributor.OTLPConfig{
+				EnableTypeAndUnitLabels: test.enableTypeAndUnitLabels,
+				AllowDeltaTemporality:   test.allowDeltaTemporality,
+			}
+			metrics := pmetric.NewMetrics()
+			rm := metrics.ResourceMetrics().AppendEmpty()
+			sm := rm.ScopeMetrics().AppendEmpty()
+
+			test.otlpSeries.CopyTo(sm.Metrics().AppendEmpty())
+
+			limits := validation.Limits{}
+			overrides := validation.NewOverrides(limits, nil)
+			promSeries, metadata, err := convertToPromTS(ctx, metrics, cfg, overrides, "user-1", logger)
+			require.NoError(t, err)
+			require.Equal(t, 1, len(promSeries))
+			require.Equal(t, prompb.FromLabels(test.expectedLabels, nil), promSeries[0].Labels)
+
+			require.Equal(t, 1, len(metadata))
+			require.Equal(t, test.expectedMetadata, metadata[0])
+		})
+	}
+}
+
 func TestOTLP_AllowDeltaTemporality(t *testing.T) {
 	logger := log.NewNopLogger()
 	ctx := context.Background()
@@ -44,24 +119,24 @@ func TestOTLP_AllowDeltaTemporality(t *testing.T) {
 			description:           "[allowDeltaTemporality: false] cumulative type should be converted",
 			allowDeltaTemporality: false,
 			otlpSeries: []pmetric.Metric{
-				createOtelSum("test_1", pmetric.AggregationTemporalityCumulative, ts),
-				createOtelSum("test_2", pmetric.AggregationTemporalityCumulative, ts),
+				createOtelSum("test_1", "", pmetric.AggregationTemporalityCumulative, ts),
+				createOtelSum("test_2", "", pmetric.AggregationTemporalityCumulative, ts),
 			},
 			expectedSeries: []prompb.TimeSeries{
 				createPromFloatSeries("test_1", ts),
 				createPromFloatSeries("test_2", ts),
 			},
 			expectedMetadata: []prompb.MetricMetadata{
-				createPromMetadata("test_1", prompb.MetricMetadata_GAUGE),
-				createPromMetadata("test_2", prompb.MetricMetadata_GAUGE),
+				createPromMetadata("test_1", "", prompb.MetricMetadata_GAUGE),
+				createPromMetadata("test_2", "", prompb.MetricMetadata_GAUGE),
 			},
 		},
 		{
 			description:           "[allowDeltaTemporality: false] delta type should not be converted",
 			allowDeltaTemporality: false,
 			otlpSeries: []pmetric.Metric{
-				createOtelSum("test_1", pmetric.AggregationTemporalityDelta, ts),
-				createOtelSum("test_2", pmetric.AggregationTemporalityDelta, ts),
+				createOtelSum("test_1", "", pmetric.AggregationTemporalityDelta, ts),
+				createOtelSum("test_2", "", pmetric.AggregationTemporalityDelta, ts),
 			},
 			expectedSeries:   []prompb.TimeSeries{},
 			expectedMetadata: []prompb.MetricMetadata{},
@@ -71,30 +146,30 @@ func TestOTLP_AllowDeltaTemporality(t *testing.T) {
 			description:           "[allowDeltaTemporality: true] delta type should be converted",
 			allowDeltaTemporality: true,
 			otlpSeries: []pmetric.Metric{
-				createOtelSum("test_1", pmetric.AggregationTemporalityDelta, ts),
-				createOtelSum("test_2", pmetric.AggregationTemporalityDelta, ts),
+				createOtelSum("test_1", "", pmetric.AggregationTemporalityDelta, ts),
+				createOtelSum("test_2", "", pmetric.AggregationTemporalityDelta, ts),
 			},
 			expectedSeries: []prompb.TimeSeries{
 				createPromFloatSeries("test_1", ts),
 				createPromFloatSeries("test_2", ts),
 			},
 			expectedMetadata: []prompb.MetricMetadata{
-				createPromMetadata("test_1", prompb.MetricMetadata_UNKNOWN),
-				createPromMetadata("test_2", prompb.MetricMetadata_UNKNOWN),
+				createPromMetadata("test_1", "", prompb.MetricMetadata_UNKNOWN),
+				createPromMetadata("test_2", "", prompb.MetricMetadata_UNKNOWN),
 			},
 		},
 		{
 			description:           "[allowDeltaTemporality: false] mixed delta and cumulative, should be converted only for cumulative type",
 			allowDeltaTemporality: false,
 			otlpSeries: []pmetric.Metric{
-				createOtelSum("test_1", pmetric.AggregationTemporalityDelta, ts),
-				createOtelSum("test_2", pmetric.AggregationTemporalityCumulative, ts),
+				createOtelSum("test_1", "", pmetric.AggregationTemporalityDelta, ts),
+				createOtelSum("test_2", "", pmetric.AggregationTemporalityCumulative, ts),
 			},
 			expectedSeries: []prompb.TimeSeries{
 				createPromFloatSeries("test_2", ts),
 			},
 			expectedMetadata: []prompb.MetricMetadata{
-				createPromMetadata("test_2", prompb.MetricMetadata_GAUGE),
+				createPromMetadata("test_2", "", prompb.MetricMetadata_GAUGE),
 			},
 			expectedErr: `invalid temporality and type combination for metric "test_1"`,
 		},
@@ -110,8 +185,8 @@ func TestOTLP_AllowDeltaTemporality(t *testing.T) {
 				createPromNativeHistogramSeries("test_2", prompb.Histogram_UNKNOWN, ts),
 			},
 			expectedMetadata: []prompb.MetricMetadata{
-				createPromMetadata("test_1", prompb.MetricMetadata_HISTOGRAM),
-				createPromMetadata("test_2", prompb.MetricMetadata_HISTOGRAM),
+				createPromMetadata("test_1", "", prompb.MetricMetadata_HISTOGRAM),
+				createPromMetadata("test_2", "", prompb.MetricMetadata_HISTOGRAM),
 			},
 		},
 		{
@@ -137,8 +212,8 @@ func TestOTLP_AllowDeltaTemporality(t *testing.T) {
 				createPromNativeHistogramSeries("test_2", prompb.Histogram_GAUGE, ts),
 			},
 			expectedMetadata: []prompb.MetricMetadata{
-				createPromMetadata("test_1", prompb.MetricMetadata_UNKNOWN),
-				createPromMetadata("test_2", prompb.MetricMetadata_UNKNOWN),
+				createPromMetadata("test_1", "", prompb.MetricMetadata_UNKNOWN),
+				createPromMetadata("test_2", "", prompb.MetricMetadata_UNKNOWN),
 			},
 		},
 		{
@@ -152,7 +227,7 @@ func TestOTLP_AllowDeltaTemporality(t *testing.T) {
 				createPromNativeHistogramSeries("test_2", prompb.Histogram_UNKNOWN, ts),
 			},
 			expectedMetadata: []prompb.MetricMetadata{
-				createPromMetadata("test_2", prompb.MetricMetadata_HISTOGRAM),
+				createPromMetadata("test_2", "", prompb.MetricMetadata_HISTOGRAM),
 			},
 			expectedErr: `invalid temporality and type combination for metric "test_1"`,
 		},
@@ -184,9 +259,10 @@ func TestOTLP_AllowDeltaTemporality(t *testing.T) {
 	}
 }
 
-func createPromMetadata(name string, metadataType prompb.MetricMetadata_MetricType) prompb.MetricMetadata {
+func createPromMetadata(name, unit string, metadataType prompb.MetricMetadata_MetricType) prompb.MetricMetadata {
 	return prompb.MetricMetadata{
 		Type:             metadataType,
+		Unit:             unit,
 		MetricFamilyName: name,
 	}
 }
@@ -221,10 +297,11 @@ func createPromFloatSeries(name string, ts time.Time) prompb.TimeSeries {
 }
 
 // copied from: https://github.com/prometheus/prometheus/blob/v3.5.0/storage/remote/otlptranslator/prometheusremotewrite/metrics_to_prw.go
-func createOtelSum(name string, temporality pmetric.AggregationTemporality, ts time.Time) pmetric.Metric {
+func createOtelSum(name, unit string, temporality pmetric.AggregationTemporality, ts time.Time) pmetric.Metric {
 	metrics := pmetric.NewMetricSlice()
 	m := metrics.AppendEmpty()
 	m.SetName(name)
+	m.SetUnit(unit)
 	sum := m.SetEmptySum()
 	sum.SetAggregationTemporality(temporality)
 	dp := sum.DataPoints().AppendEmpty()