Support otlp metadata ingestion (#6617)

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

Author: SungJin1212

CHANGELOG.md

@@ -12,6 +12,7 @@
 * [ENHANCEMENT] Querier: Apply bytes limiter to LabelNames and LabelValuesForLabelNames. #6568
 * [ENHANCEMENT] Query Frontend: Add a `too_many_tenants` reason label value to `cortex_rejected_queries_total` metric to track the rejected query count due to the # of tenant limits. #6569
 * [ENHANCEMENT] Alertmanager: Add receiver validations for msteamsv2 and rocketchat. #6606
+* [ENHANCEMENT] OTLP: Support otlp metadata ingestion. #6617
 * [BUGFIX] Ingester: Avoid error or early throttling when READONLY ingesters are present in the ring #6517
 * [BUGFIX] Ingester: Fix labelset data race condition. #6573
 * [BUGFIX] Compactor: Cleaner should not put deletion marker for blocks with no-compact marker. #6576

integration/e2ecortex/client.go

@@ -215,14 +215,18 @@ func convertBucketLayout(bucket pmetric.ExponentialHistogramDataPointBuckets, sp
 }
 
 // Convert Timeseries to Metrics
-func convertTimeseriesToMetrics(timeseries []prompb.TimeSeries) pmetric.Metrics {
+func convertTimeseriesToMetrics(timeseries []prompb.TimeSeries, metadata []prompb.MetricMetadata) pmetric.Metrics {
 	metrics := pmetric.NewMetrics()
-	for _, ts := range timeseries {
+	for i, ts := range timeseries {
 		metricName, attributes := getNameAndAttributes(ts)
 		newMetric := metrics.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty().Metrics().AppendEmpty()
 		newMetric.SetName(metricName)
-		//TODO Set description for new metric
-		//TODO Set unit for new metric
+
+		if metadata != nil {
+			newMetric.SetDescription(metadata[i].GetHelp())
+			newMetric.SetUnit(metadata[i].GetUnit())
+		}
+
 		if len(ts.Samples) > 0 {
 			createDataPointsGauge(newMetric, attributes, ts.Samples)
 		} else if len(ts.Histograms) > 0 {
@@ -302,9 +306,9 @@ func (c *Client) OTLPPushExemplar(name string, labels ...prompb.Label) (*http.Re
 }
 
 // Push series to OTLP endpoint
-func (c *Client) OTLP(timeseries []prompb.TimeSeries) (*http.Response, error) {
+func (c *Client) OTLP(timeseries []prompb.TimeSeries, metadata []prompb.MetricMetadata) (*http.Response, error) {
 
-	data, err := pmetricotlp.NewExportRequestFromMetrics(convertTimeseriesToMetrics(timeseries)).MarshalProto()
+	data, err := pmetricotlp.NewExportRequestFromMetrics(convertTimeseriesToMetrics(timeseries, metadata)).MarshalProto()
 	if err != nil {
 		return nil, err
 	}
@@ -356,6 +360,12 @@ func (c *Client) Query(query string, ts time.Time) (model.Value, error) {
 	return value, err
 }
 
+// Metadata runs a metadata query
+func (c *Client) Metadata(name, limit string) (map[string][]promv1.Metadata, error) {
+	metadata, err := c.querierClient.Metadata(context.Background(), name, limit)
+	return metadata, err
+}
+
 // QueryExemplars runs an exemplars query
 func (c *Client) QueryExemplars(query string, start, end time.Time) ([]promv1.ExemplarQueryResult, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), c.timeout)

integration/otlp_test.go

@@ -61,14 +61,21 @@ func TestOTLP(t *testing.T) {
 
 	// Push some series to Cortex.
 	now := time.Now()
-	series, expectedVector := generateSeries("series_1", now, prompb.Label{Name: "foo", Value: "bar"})
+	series, expectedVector := generateSeries("series_1_total", now, prompb.Label{Name: "foo", Value: "bar"})
 
-	res, err := c.OTLP(series)
+	metadata := []prompb.MetricMetadata{
+		{
+			Help: "help",
+			Unit: "total",
+		},
+	}
+
+	res, err := c.OTLP(series, metadata)
 	require.NoError(t, err)
 	require.Equal(t, 200, res.StatusCode)
 
 	// Query the series.
-	result, err := c.Query("series_1", now)
+	result, err := c.Query("series_1_total", now)
 	require.NoError(t, err)
 	require.Equal(t, model.ValVector, result.Type())
 	assert.Equal(t, expectedVector, result.(model.Vector))
@@ -81,13 +88,17 @@ func TestOTLP(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []string{"__name__", "foo"}, labelNames)
 
+	metadataResult, err := c.Metadata("series_1", "")
+	require.NoError(t, err)
+	require.Equal(t, 1, len(metadataResult))
+
 	// Check that a range query does not return an error to sanity check the queryrange tripperware.
 	_, err = c.QueryRange("series_1", now.Add(-15*time.Minute), now, 15*time.Second)
 	require.NoError(t, err)
 
 	i := rand.Uint32()
 	histogramSeries := e2e.GenerateHistogramSeries("histogram_series", now, i, false, prompb.Label{Name: "job", Value: "test"})
-	res, err = c.OTLP(histogramSeries)
+	res, err = c.OTLP(histogramSeries, nil)
 	require.NoError(t, err)
 	require.Equal(t, 200, res.StatusCode)
 

pkg/util/push/otlp.go

@@ -65,7 +65,7 @@ func OTLPHandler(maxRecvMsgSize int, overrides *validation.Overrides, cfg distri
 		}
 
 		// otlp to prompb TimeSeries
-		promTsList, err := convertToPromTS(r.Context(), req.Metrics(), cfg, overrides, userID, logger)
+		promTsList, promMetadata, err := convertToPromTS(r.Context(), req.Metrics(), cfg, overrides, userID, logger)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
@@ -81,7 +81,11 @@ func OTLPHandler(maxRecvMsgSize int, overrides *validation.Overrides, cfg distri
 				Histograms: makeHistograms(v.Histograms),
 			}})
 		}
+
+		metadata := makeMetadata(promMetadata)
+
 		prwReq.Timeseries = tsList
+		prwReq.Metadata = metadata
 
 		if _, err := push(ctx, &prwReq); err != nil {
 			resp, ok := httpgrpc.HTTPResponseFromError(err)
@@ -99,6 +103,19 @@ func OTLPHandler(maxRecvMsgSize int, overrides *validation.Overrides, cfg distri
 	})
 }
 
+func makeMetadata(promMetadata []prompb.MetricMetadata) []*cortexpb.MetricMetadata {
+	metadata := make([]*cortexpb.MetricMetadata, 0, len(promMetadata))
+	for _, m := range promMetadata {
+		metadata = append(metadata, &cortexpb.MetricMetadata{
+			Type:             cortexpb.MetricMetadata_MetricType(m.Type),
+			MetricFamilyName: m.MetricFamilyName,
+			Help:             m.Help,
+			Unit:             m.Unit,
+		})
+	}
+	return metadata
+}
+
 func decodeOTLPWriteRequest(ctx context.Context, r *http.Request, maxSize int) (pmetricotlp.ExportRequest, error) {
 	expectedSize := int(r.ContentLength)
 	if expectedSize > maxSize {
@@ -157,7 +174,7 @@ func decodeOTLPWriteRequest(ctx context.Context, r *http.Request, maxSize int) (
 	return decoderFunc(r.Body)
 }
 
-func convertToPromTS(ctx context.Context, pmetrics pmetric.Metrics, cfg distributor.OTLPConfig, overrides *validation.Overrides, userID string, logger log.Logger) ([]prompb.TimeSeries, error) {
+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,
@@ -181,9 +198,10 @@ func convertToPromTS(ctx context.Context, pmetrics pmetric.Metrics, cfg distribu
 
 	if err != nil {
 		level.Error(logger).Log("msg", "Error translating OTLP metrics to Prometheus write request", "err", err)
-		return nil, err
+		return nil, nil, err
 	}
-	return promConverter.TimeSeries(), nil
+
+	return promConverter.TimeSeries(), promConverter.Metadata(), nil
 }
 
 func makeLabels(in []prompb.Label) []cortexpb.LabelAdapter {

pkg/util/push/otlp_test.go

@@ -195,9 +195,15 @@ func TestOTLPConvertToPromTS(t *testing.T) {
 			}
 			overrides, err := validation.NewOverrides(limits, nil)
 			require.NoError(t, err)
-			tsList, err := convertToPromTS(ctx, d, test.cfg, overrides, "user-1", logger)
+			tsList, metadata, err := convertToPromTS(ctx, d, test.cfg, overrides, "user-1", logger)
 			require.NoError(t, err)
 
+			// test metadata conversion
+			require.Equal(t, 1, len(metadata))
+			require.Equal(t, prompb.MetricMetadata_MetricType(1), metadata[0].Type)
+			require.Equal(t, "test_counter_total", metadata[0].MetricFamilyName)
+			require.Equal(t, "test-counter-description", metadata[0].Help)
+
 			if test.cfg.DisableTargetInfo {
 				require.Equal(t, 1, len(tsList)) // test_counter_total
 			} else {