OTLP: Use optimized OTel metric translator, converting directly to Mimir format

.github/workflows/test-build-deploy.yml

@@ -30,7 +30,6 @@ jobs:
       # Determine if we will deploy (aka push) the image to the registry.
       is_deploy: ${{ (startsWith(github.ref, 'refs/tags/') || startsWith(github.ref, 'refs/heads/r')) && github.event_name == 'push' && github.repository == 'grafana/mimir' }}
 
-
   goversion:
     runs-on: ubuntu-latest
     needs: prepare
@@ -103,6 +102,8 @@ jobs:
         run: make BUILD_IN_CONTAINER=false check-license
       - name: Check Docker-Compose YAML
         run: make BUILD_IN_CONTAINER=false check-mimir-microservices-mode-docker-compose-yaml check-mimir-read-write-mode-docker-compose-yaml
+      - name: Check Generated OTLP Code
+        run: make BUILD_IN_CONTAINER=false check-generated-otlp-code
 
   doc-validator:
     runs-on: ubuntu-latest

CHANGELOG.md

@@ -41,6 +41,7 @@
 * [ENHANCEMENT] Store-gateway: improve performance when streaming chunks to queriers is enabled (`-querier.prefer-streaming-chunks-from-store-gateways=true`) and the query selects fewer than `-blocks-storage.bucket-store.batch-series-size` series (defaults to 5000 series). #8039
 * [ENHANCEMENT] Ingester: active series are now updated along with owned series. They decrease when series change ownership between ingesters. This helps provide a more accurate total of active series when ingesters are added. This is only enabled when `-ingester.track-ingester-owned-series` or `-ingester.use-ingester-owned-series-for-limits` are enabled. #8084
 * [ENHANCEMENT] Query-frontend: include route name in query stats log lines. #8191
+* [ENHANCEMENT] OTLP: Speed up conversion from OTel to Mimir format by about 8% and reduce memory consumption by about 30%. #7957
 * [BUGFIX] Rules: improve error handling when querier is local to the ruler. #7567
 * [BUGFIX] Querier, store-gateway: Protect against panics raised during snappy encoding. #7520
 * [BUGFIX] Ingester: Prevent timely compaction of empty blocks. #7624

Makefile

@@ -231,6 +231,9 @@ images: ## Print all image names.
 PROTO_DEFS := $(shell find . $(DONT_FIND) -type f -name '*.proto' -print)
 PROTO_GOS := $(patsubst %.proto,%.pb.go,$(PROTO_DEFS))
 
+# Generating OTLP translation code is automated.
+OTLP_GOS := $(shell find ./pkg/distributor/otlp/ -type f -name '*_generated.go' -print)
+
 # Building binaries is now automated. The convention is to build a binary
 # for every directory with main.go in it.
 MAIN_GO := $(shell find . $(DONT_FIND) -type f -name 'main.go' -print)
@@ -260,7 +263,7 @@ mimir-build-image/$(UPTODATE): mimir-build-image/*
 # All the boiler plate for building golang follows:
 SUDO := $(shell docker info >/dev/null 2>&1 || echo "sudo -E")
 BUILD_IN_CONTAINER ?= true
-LATEST_BUILD_IMAGE_TAG ?= pr8107-6dc27b0894
+LATEST_BUILD_IMAGE_TAG ?= pr7957-d7652788a7
 
 # TTY is parameterized to allow Google Cloud Builder to run builds,
 # as it currently disallows TTY devices. This value needs to be overridden
@@ -771,3 +774,11 @@ test-packages: packages packaging/rpm/centos-systemd/$(UPTODATE) packaging/deb/d
 
 docs: doc
 	cd docs && $(MAKE) docs
+
+.PHONY: generate-otlp
+generate-otlp:
+	cd pkg/distributor/otlp && rm -f *_generated.go && go generate
+
+.PHONY: check-generated-otlp-code
+check-generated-otlp-code: generate-otlp
+	@./tools/find-diff-or-untracked.sh $(OTLP_GOS) || (echo "Please rebuild OTLP code by running 'generate-otlp'" && false)

go.mod

@@ -140,7 +140,7 @@ require (
 	github.com/bits-and-blooms/bitset v1.13.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
-	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
@@ -237,7 +237,7 @@ require (
 	go.etcd.io/etcd/client/v3 v3.5.4 // indirect
 	go.mongodb.org/mongo-driver v1.14.0 // indirect
 	go.opencensus.io v0.24.0 // indirect
-	go.opentelemetry.io/collector/semconv v0.98.0 // indirect
+	go.opentelemetry.io/collector/semconv v0.98.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.50.0 // indirect
 	go.opentelemetry.io/otel/metric v1.26.0 // indirect
 	go.uber.org/zap v1.21.0 // indirect

mimir-build-image/Dockerfile

@@ -56,6 +56,7 @@ RUN GO111MODULE=on \
 	go install github.com/google/go-jsonnet/cmd/[email protected] && \
 	go install github.com/norwoodj/helm-docs/cmd/[email protected] && \
 	go install github.com/open-policy-agent/[email protected] && \
+	go install github.com/uber-go/[email protected] && \
 	rm -rf /go/pkg /go/src /root/.cache
 
 COPY --from=helm /usr/bin/helm /usr/bin/helm

pkg/distributor/otel.go

@@ -20,12 +20,12 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/prompb"
 	prometheustranslator "github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheus"
-	"github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheusremotewrite"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
 	"go.uber.org/multierr"
 
+	"github.com/grafana/mimir/pkg/distributor/otlp"
 	"github.com/grafana/mimir/pkg/mimirpb"
 	"github.com/grafana/mimir/pkg/util"
 	"github.com/grafana/mimir/pkg/util/spanlogger"
@@ -259,11 +259,11 @@ func otelMetricsToMetadata(addSuffixes bool, md pmetric.Metrics) []*mimirpb.Metr
 }
 
 func otelMetricsToTimeseries(tenantID string, addSuffixes bool, discardedDueToOtelParseError *prometheus.CounterVec, logger log.Logger, md pmetric.Metrics) ([]mimirpb.PreallocTimeseries, error) {
-	converter := prometheusremotewrite.NewPrometheusConverter()
-	errs := converter.FromMetrics(md, prometheusremotewrite.Settings{
+	converter := otlp.NewMimirConverter()
+	errs := converter.FromMetrics(md, otlp.Settings{
 		AddMetricSuffixes: addSuffixes,
 	})
-	promTS := converter.TimeSeries()
+	mimirTS := converter.TimeSeries()
 	if errs != nil {
 		dropped := len(multierr.Errors(errs))
 		discardedDueToOtelParseError.WithLabelValues(tenantID, "").Add(float64(dropped)) // Group is empty here as metrics couldn't be parsed
@@ -273,106 +273,16 @@ func otelMetricsToTimeseries(tenantID string, addSuffixes bool, discardedDueToOt
 			parseErrs = parseErrs[:maxErrMsgLen]
 		}
 
-		if len(promTS) == 0 {
+		if len(mimirTS) == 0 {
 			return nil, errors.New(parseErrs)
 		}
 
 		level.Warn(logger).Log("msg", "OTLP parse error", "err", parseErrs)
 	}
 
-	mimirTS := mimirpb.PreallocTimeseriesSliceFromPool()
-	for _, ts := range promTS {
-		mimirTS = append(mimirTS, promToMimirTimeseries(&ts))
-	}
-
 	return mimirTS, nil
 }
 
-func promToMimirTimeseries(promTs *prompb.TimeSeries) mimirpb.PreallocTimeseries {
-	labels := make([]mimirpb.LabelAdapter, 0, len(promTs.Labels))
-	for _, label := range promTs.Labels {
-		labels = append(labels, mimirpb.LabelAdapter{
-			Name:  label.Name,
-			Value: label.Value,
-		})
-	}
-
-	samples := make([]mimirpb.Sample, 0, len(promTs.Samples))
-	for _, sample := range promTs.Samples {
-		samples = append(samples, mimirpb.Sample{
-			TimestampMs: sample.Timestamp,
-			Value:       sample.Value,
-		})
-	}
-
-	histograms := make([]mimirpb.Histogram, 0, len(promTs.Histograms))
-	for idx := range promTs.Histograms {
-		histograms = append(histograms, promToMimirHistogram(&promTs.Histograms[idx]))
-	}
-
-	exemplars := make([]mimirpb.Exemplar, 0, len(promTs.Exemplars))
-	for _, exemplar := range promTs.Exemplars {
-		labels := make([]mimirpb.LabelAdapter, 0, len(exemplar.Labels))
-		for _, label := range exemplar.Labels {
-			labels = append(labels, mimirpb.LabelAdapter{
-				Name:  label.Name,
-				Value: label.Value,
-			})
-		}
-
-		exemplars = append(exemplars, mimirpb.Exemplar{
-			Labels:      labels,
-			Value:       exemplar.Value,
-			TimestampMs: exemplar.Timestamp,
-		})
-	}
-
-	ts := mimirpb.TimeseriesFromPool()
-	ts.Labels = labels
-	ts.Samples = samples
-	ts.Histograms = histograms
-	ts.Exemplars = exemplars
-
-	return mimirpb.PreallocTimeseries{TimeSeries: ts}
-}
-
-func promToMimirHistogram(h *prompb.Histogram) mimirpb.Histogram {
-	pSpans := make([]mimirpb.BucketSpan, 0, len(h.PositiveSpans))
-	for _, span := range h.PositiveSpans {
-		pSpans = append(
-			pSpans, mimirpb.BucketSpan{
-				Offset: span.Offset,
-				Length: span.Length,
-			},
-		)
-	}
-	nSpans := make([]mimirpb.BucketSpan, 0, len(h.NegativeSpans))
-	for _, span := range h.NegativeSpans {
-		nSpans = append(
-			nSpans, mimirpb.BucketSpan{
-				Offset: span.Offset,
-				Length: span.Length,
-			},
-		)
-	}
-
-	return mimirpb.Histogram{
-		Count:          &mimirpb.Histogram_CountInt{CountInt: h.GetCountInt()},
-		Sum:            h.Sum,
-		Schema:         h.Schema,
-		ZeroThreshold:  h.ZeroThreshold,
-		ZeroCount:      &mimirpb.Histogram_ZeroCountInt{ZeroCountInt: h.GetZeroCountInt()},
-		NegativeSpans:  nSpans,
-		NegativeDeltas: h.NegativeDeltas,
-		NegativeCounts: h.NegativeCounts,
-		PositiveSpans:  pSpans,
-		PositiveDeltas: h.PositiveDeltas,
-		PositiveCounts: h.PositiveCounts,
-		Timestamp:      h.Timestamp,
-		ResetHint:      mimirpb.Histogram_ResetHint(h.ResetHint),
-	}
-}
-
 // TimeseriesToOTLPRequest is used in tests.
 func TimeseriesToOTLPRequest(timeseries []prompb.TimeSeries, metadata []mimirpb.MetricMetadata) pmetricotlp.ExportRequest {
 	d := pmetric.NewMetrics()

pkg/distributor/otlp/generate.sh

@@ -0,0 +1,26 @@
+#!/bin/bash
+# SPDX-License-Identifier: AGPL-3.0-only
+
+set -euo pipefail
+
+# Use GNU sed on MacOS falling back to `sed` everywhere else
+SED=$(which gsed || which sed)
+
+FILES=$(find ../../../vendor/github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheusremotewrite -name '*.go' ! -name timeseries.go ! -name "*_test.go")
+
+for SRC in $FILES
+do
+  BASENAME=$(basename "$SRC")
+  DST="${BASENAME%%.go}_generated.go"
+
+  rm -f "$DST"
+  echo "Processing $SRC to $DST"
+  printf "// Code generated from Prometheus sources - DO NOT EDIT.\n\n" >"$DST"
+  cat "$SRC" >> "$DST"
+
+  gopatch -p mimirpb.patch "$DST"
+
+  $SED -i "s/PrometheusConverter/MimirConverter/g" "$DST"
+  $SED -i "s/Prometheus remote write format/Mimir remote write format/g" "$DST"
+  goimports -w -local github.com/grafana/mimir "$DST"
+done