feat: support metric otlp http ingestion

api/metrics/v1/http.go

@@ -34,6 +34,8 @@ const (
 	RulesRoute       = "/api/v1/rules"
 	RulesRawRoute    = "/api/v1/rules/raw"
 
+	OTLPRoute = "/otlp/v1/metrics"
+
 	AlertmanagerAlertsRoute   = "/am/api/v2/alerts"
 	AlertmanagerSilencesRoute = "/am/api/v2/silences"
 )
@@ -167,6 +169,7 @@ func (n nopInstrumentHandler) NewHandler(_ prometheus.Labels, handler http.Handl
 type Endpoints struct {
 	ReadEndpoint         *url.URL
 	WriteEndpoint        *url.URL
+	OTLPWriteEndpoint    *url.URL
 	RulesEndpoint        *url.URL
 	AlertmanagerEndpoint *url.URL
 }
@@ -410,6 +413,47 @@ func NewHandler(endpoints Endpoints, tlsOptions *tls.UpstreamOptions, opts ...Ha
 		})
 	}
 
+	if endpoints.OTLPWriteEndpoint != nil {
+		var proxyOTLPWrite http.Handler
+		{
+			middlewares := proxy.Middlewares(
+				proxy.MiddlewareSetUpstream(endpoints.OTLPWriteEndpoint),
+				proxy.MiddlewareSetPrefixHeader(),
+				proxy.MiddlewareLogger(c.logger),
+				proxy.MiddlewareMetrics(c.registry, prometheus.Labels{"proxy": "metricsv1-otlp-write"}),
+			)
+
+			t := &http.Transport{
+				DialContext: (&net.Dialer{
+					Timeout: dialTimeout,
+				}).DialContext,
+				TLSClientConfig: tlsOptions.NewClientConfig(),
+			}
+
+			proxyOTLPWrite = &httputil.ReverseProxy{
+				Director:  middlewares,
+				ErrorLog:  proxy.Logger(c.logger),
+				Transport: otelhttp.NewTransport(t),
+			}
+		}
+		r.Group(func(r chi.Router) {
+			r.Use(func(handler http.Handler) http.Handler {
+				return server.InjectLabelsCtx(
+					prometheus.Labels{"group": "metricsv1", "handler": "otlp"},
+					handler,
+				)
+			})
+			r.Use(c.writeMiddlewares...)
+			r.Use(server.StripTenantPrefix("/api/metrics/v1"))
+			r.Handle(OTLPRoute,
+				otelhttp.WithRouteTag(
+					c.spanRoutePrefix+OTLPRoute,
+					proxyOTLPWrite,
+				),
+			)
+		})
+	}
+
 	if endpoints.RulesEndpoint != nil {
 		client, err := rules.NewClient(endpoints.RulesEndpoint.String())
 		if err != nil {

client/spec.yaml

@@ -390,6 +390,45 @@ paths:
           description: Delete silence response
         5XX:
           description: Server side error
+  /api/metrics/v1/{tenant}/otlp/v1/metrics:
+    parameters:
+      - $ref: '#/components/parameters/tenant'
+    post:
+      tags:
+        - metrics/otlpv1
+      summary: Send metrics in OTLP format
+      operationId: postOTLPMetrics
+      description: |
+        Send metrics in OpenTelemetry Protocol (OTLP) format.
+        The request is proxied to the configured OTLP HTTP backend.
+      requestBody:
+        description: OTLP ExportMetricsServiceRequest
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/OTLPExportMetricsServiceRequest'
+          application/x-protobuf:
+            schema:
+              type: string
+              format: binary
+      responses:
+        '200':
+          description: Successfully exported metrics
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/OTLPExportMetricsServiceResponse'
+        '400':
+          description: Bad request
+        '401':
+          description: Unauthorized - missing or invalid tenant ID
+        '403':
+          description: Forbidden - insufficient permissions
+        '429':
+          description: Too many requests - rate limit exceeded
+        5XX:
+          description: Server side error
   /api/logs/v1/{tenant}/loki/api/v1/push:
     parameters:
       - $ref: '#/components/parameters/tenant'
@@ -1710,6 +1749,73 @@ components:
         data:
           type: object
           $ref: '#/components/schemas/Rules'
+    OTLPExportMetricsServiceRequest:
+      type: object
+      description: OTLP ExportMetricsServiceRequest containing resource metrics
+      properties:
+        resourceMetrics:
+          type: array
+          items:
+            type: object
+            properties:
+              resource:
+                type: object
+                properties:
+                  attributes:
+                    type: array
+                    items:
+                      $ref: '#/components/schemas/OTLPKeyValue'
+              scopeMetrics:
+                type: array
+                items:
+                  type: object
+                  properties:
+                    scope:
+                      type: object
+                      properties:
+                        name:
+                          type: string
+                        version:
+                          type: string
+                    metrics:
+                      type: array
+                      items:
+                        type: object
+                        properties:
+                          name:
+                            type: string
+                          unit:
+                            type: string
+                          description:
+                            type: string
+    OTLPExportMetricsServiceResponse:
+      type: object
+      description: OTLP ExportMetricsServiceResponse
+      properties:
+        partialSuccess:
+          type: object
+          properties:
+            rejectedDataPoints:
+              type: integer
+              format: int64
+            errorMessage:
+              type: string
+    OTLPKeyValue:
+      type: object
+      properties:
+        key:
+          type: string
+        value:
+          type: object
+          properties:
+            stringValue:
+              type: string
+            intValue:
+              type: string
+            doubleValue:
+              type: number
+            boolValue:
+              type: boolean
 x-tagGroups:
   - name: metrics
     tags:
@@ -1719,6 +1825,7 @@ x-tagGroups:
       - metrics/labelvaluesv1
       - metrics/queryv1
       - metrics/seriesv1
+      - metrics/otlpv1
   - name: logs
     tags:
       - logs/pushv1

main.go

@@ -145,6 +145,7 @@ type tlsConfig struct {
 type metricsConfig struct {
 	readEndpoint         *url.URL
 	writeEndpoint        *url.URL
+	otlpWriteEndpoint    *url.URL
 	rulesEndpoint        *url.URL
 	alertmanagerEndpoint *url.URL
 	upstreamWriteTimeout time.Duration
@@ -644,6 +645,7 @@ func main() {
 				eps := metricsv1.Endpoints{
 					ReadEndpoint:         cfg.metrics.readEndpoint,
 					WriteEndpoint:        cfg.metrics.writeEndpoint,
+					OTLPWriteEndpoint:    cfg.metrics.otlpWriteEndpoint,
 					RulesEndpoint:        cfg.metrics.rulesEndpoint,
 					AlertmanagerEndpoint: cfg.metrics.alertmanagerEndpoint,
 				}
@@ -1108,10 +1110,11 @@ func (m *multiStringFlag) String() string {
 func parseFlags() (config, error) {
 	var (
 		rawTLSCipherSuites             string
-		rawMetricsReadEndpoint         string
-		rawMetricsWriteEndpoint        string
-		rawMetricsRulesEndpoint        string
-		rawMetricsAlertmanagerEndpoint string
+		rawMetricsReadEndpoint          string
+		rawMetricsWriteEndpoint         string
+		rawMetricsWriteOTLPHTTPEndpoint string
+		rawMetricsRulesEndpoint         string
+		rawMetricsAlertmanagerEndpoint  string
 		rawLogsReadEndpoint            string
 		rawLogsRulesEndpoint           string
 		rawLogsTailEndpoint            string
@@ -1192,6 +1195,8 @@ func parseFlags() (config, error) {
 		"The endpoint against which to send read requests for metrics.")
 	flag.StringVar(&rawMetricsWriteEndpoint, "metrics.write.endpoint", "",
 		"The endpoint against which to make write requests for metrics.")
+	flag.StringVar(&rawMetricsWriteOTLPHTTPEndpoint, "metrics.write.otlphttp.endpoint", "",
+		"The endpoint against which to make OTLP HTTP write requests for metrics.")
 	flag.StringVar(&rawMetricsRulesEndpoint, "metrics.rules.endpoint", "",
 		"The endpoint against which to make get requests for listing recording/alerting rules and put requests for creating/updating recording/alerting rules.")
 	flag.StringVar(&rawMetricsAlertmanagerEndpoint, "metrics.alertmanager.endpoint", "",
@@ -1320,6 +1325,17 @@ func parseFlags() (config, error) {
 		cfg.metrics.writeEndpoint = metricsWriteEndpoint
 	}
 
+	if rawMetricsWriteOTLPHTTPEndpoint != "" {
+		cfg.metrics.enabled = true
+
+		metricsOTLPWriteEndpoint, err := url.ParseRequestURI(rawMetricsWriteOTLPHTTPEndpoint)
+		if err != nil {
+			return cfg, fmt.Errorf("--metrics.write.otlphttp.endpoint %q is invalid: %w", rawMetricsWriteOTLPHTTPEndpoint, err)
+		}
+
+		cfg.metrics.otlpWriteEndpoint = metricsOTLPWriteEndpoint
+	}
+
 	if rawMetricsRulesEndpoint != "" {
 		cfg.metrics.enabled = true
 
@@ -1642,6 +1658,7 @@ var metricsV1Group = []groupHandler{
 	{"metricsv1", "labels"},
 	{"metricsv1", "labelvalues"},
 	{"metricsv1", "receive"},
+	{"metricsv1", "otlp"},
 	{"metricsv1", "rules"},
 	{"metricsv1", "rules-raw"},
 	{"metricsv1", "alerts"},

test/e2e/configs.go

@@ -18,6 +18,7 @@ type testType string
 
 const (
 	metrics        testType = "metrics"
+	metricsOTLP    testType = "metricsOTLP"
 	rules          testType = "rules"
 	alerts         testType = "alerts"
 	logs           testType = "logs"
@@ -32,6 +33,7 @@ const (
 	configSharedDir = "config"
 
 	envMetricsName        = "metrics"
+	envMetricsOTLPName    = "metrics-otlp"
 	envRulesAPIName       = "rules-api"
 	envAlertmanagerName   = "alertmanager-api"
 	envLogsName           = "logs-tail"
@@ -379,6 +381,65 @@ func createOtelForwardingCollectorConfigYAML(
 	testutil.Ok(t, err)
 }
 
+// OTel collector config for metrics: receives OTLP metrics and exports via Prometheus remote write.
+const otelMetricsConfigTpl = `
+receivers:
+    otlp:
+      protocols:
+        http:
+            endpoint: "0.0.0.0:4318"
+
+exporters:
+    debug:
+        verbosity: detailed
+    prometheusremotewrite:
+        endpoint: "http://%[1]s/api/v1/receive"
+        tls:
+          insecure: true
+        headers:
+          THANOS-TENANT: "%[2]s"
+
+service:
+    telemetry:
+      metrics:
+        readers:
+          - pull:
+              exporter:
+                prometheus:
+                  host: 0.0.0.0
+                  port: 8888
+        level: detailed
+      logs:
+        level: DEBUG
+    pipelines:
+        metrics:
+            receivers: [otlp]
+            exporters: [debug,prometheusremotewrite]
+`
+
+func createOtelMetricsCollectorConfigYAML(
+	t *testing.T,
+	e e2e.Environment,
+	thanosReceiveEndpoint string,
+	tenantID string,
+) {
+	if strings.ContainsRune(otelMetricsConfigTpl, '\t') {
+		t.Errorf("Tab in the YAML")
+	}
+
+	yamlContent := []byte(fmt.Sprintf(
+		otelMetricsConfigTpl,
+		thanosReceiveEndpoint,
+		tenantID))
+
+	err := os.WriteFile(
+		filepath.Join(e.SharedDir(), configSharedDir, "metrics-collector.yaml"),
+		yamlContent,
+		os.FileMode(0644),
+	)
+	testutil.Ok(t, err)
+}
+
 const lokiYAMLTpl = `auth_enabled: true
 
 server:

test/e2e/helpers.go

@@ -88,6 +88,8 @@ func getContainerName(t *testing.T, tt testType, serviceName string) string {
 		return envLogsName + "-" + serviceName
 	case metrics:
 		return envMetricsName + "-" + serviceName
+	case metricsOTLP:
+		return envMetricsOTLPName + "-" + serviceName
 	case rules:
 		return envRulesAPIName + "-" + serviceName
 	case alerts: