Add influx push endpoint to mimir

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/grafana/dskit v0.0.0-20241125123840-77bb9ddddb0c
 	github.com/grafana/e2e v0.1.2-0.20240118170847-db90b84177fc
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
+	github.com/influxdata/influxdb/v2 v2.7.11
 	github.com/json-iterator/go v1.1.12
 	github.com/minio/minio-go/v7 v7.0.81
 	github.com/mitchellh/go-wordwrap v1.0.1

go.sum

@@ -1390,6 +1390,8 @@ github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4=
 github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/influxdata/influxdb/v2 v2.7.11 h1:qs9qr5hsuFrlTiBtr5lBrALbQ2dHAanf21fBLlLpKww=
+github.com/influxdata/influxdb/v2 v2.7.11/go.mod h1:zNOyzQy6WbfvGi1CK1aJ2W8khOq9+Gdsj8yLj8bHHqg=
 github.com/ionos-cloud/sdk-go/v6 v6.2.1 h1:mxxN+frNVmbFrmmFfXnBC3g2USYJrl6mc1LW2iNYbFY=
 github.com/ionos-cloud/sdk-go/v6 v6.2.1/go.mod h1:SXrO9OGyWjd2rZhAhEpdYN6VUAODzzqRdqA9BCviQtI=
 github.com/jessevdk/go-flags v1.5.0 h1:1jKYvbxEjfUl0fmqTCOfonvskHHXMjBySTLW4y9LFvc=

pkg/api/api.go

@@ -259,6 +259,7 @@ func (a *API) RegisterRuntimeConfig(runtimeConfigHandler http.HandlerFunc, userL
 
 const PrometheusPushEndpoint = "/api/v1/push"
 const OTLPPushEndpoint = "/otlp/v1/metrics"
+const InfluxPushEndpoint = "/api/v1/influx/push"
 
 // RegisterDistributor registers the endpoints associated with the distributor.
 func (a *API) RegisterDistributor(d *distributor.Distributor, pushConfig distributor.Config, reg prometheus.Registerer, limits *validation.Overrides) {
@@ -268,6 +269,10 @@ func (a *API) RegisterDistributor(d *distributor.Distributor, pushConfig distrib
 		pushConfig.MaxRecvMsgSize, d.RequestBufferPool, a.sourceIPs, a.cfg.SkipLabelNameValidationHeader,
 		a.cfg.SkipLabelCountValidationHeader, limits, pushConfig.RetryConfig, d.PushWithMiddlewares, d.PushMetrics, a.logger,
 	), true, false, "POST")
+	// TODO(alexg): hidden behind a featureflag or experimental config option?
+	a.RegisterRoute(InfluxPushEndpoint, distributor.InfluxHandler(
+		pushConfig.MaxInfluxRequestSize, d.RequestBufferPool, a.sourceIPs, pushConfig.RetryConfig, d.PushWithMiddlewares, d.PushMetrics, reg, a.logger,
+	), true, false, "POST")
 	a.RegisterRoute(OTLPPushEndpoint, distributor.OTLPHandler(
 		pushConfig.MaxOTLPRequestSize, d.RequestBufferPool, a.sourceIPs, limits, pushConfig.OTelResourceAttributePromotionConfig,
 		pushConfig.RetryConfig, d.PushWithMiddlewares, d.PushMetrics, reg, a.logger,

pkg/distributor/distributor.go

@@ -86,7 +86,8 @@ const (
 	// metaLabelTenantID is the name of the metric_relabel_configs label with tenant ID.
 	metaLabelTenantID = model.MetaLabelPrefix + "tenant_id"
 
-	maxOTLPRequestSizeFlag = "distributor.max-otlp-request-size"
+	maxOTLPRequestSizeFlag   = "distributor.max-otlp-request-size"
+	maxInfluxRequestSizeFlag = "distributor.max-influx-request-size"
 
 	instanceIngestionRateTickInterval = time.Second
 
@@ -200,6 +201,7 @@ type Config struct {
 
 	MaxRecvMsgSize           int           `yaml:"max_recv_msg_size" category:"advanced"`
 	MaxOTLPRequestSize       int           `yaml:"max_otlp_request_size" category:"experimental"`
+	MaxInfluxRequestSize     int           `yaml:"max_influx_request_size" category:"experimental"`
 	MaxRequestPoolBufferSize int           `yaml:"max_request_pool_buffer_size" category:"experimental"`
 	RemoteTimeout            time.Duration `yaml:"remote_timeout" category:"advanced"`
 
@@ -255,6 +257,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet, logger log.Logger) {
 
 	f.IntVar(&cfg.MaxRecvMsgSize, "distributor.max-recv-msg-size", 100<<20, "Max message size in bytes that the distributors will accept for incoming push requests to the remote write API. If exceeded, the request will be rejected.")
 	f.IntVar(&cfg.MaxOTLPRequestSize, maxOTLPRequestSizeFlag, 100<<20, "Maximum OTLP request size in bytes that the distributors accept. Requests exceeding this limit are rejected.")
+	f.IntVar(&cfg.MaxInfluxRequestSize, maxInfluxRequestSizeFlag, 100<<20, "Maximum Influx request size in bytes that the distributors accept. Requests exceeding this limit are rejected.")
 	f.IntVar(&cfg.MaxRequestPoolBufferSize, "distributor.max-request-pool-buffer-size", 0, "Max size of the pooled buffers used for marshaling write requests. If 0, no max size is enforced.")
 	f.DurationVar(&cfg.RemoteTimeout, "distributor.remote-timeout", 2*time.Second, "Timeout for downstream ingesters.")
 	f.BoolVar(&cfg.WriteRequestsBufferPoolingEnabled, "distributor.write-requests-buffer-pooling-enabled", true, "Enable pooling of buffers used for marshaling write requests.")
@@ -282,6 +285,7 @@ const (
 )
 
 type PushMetrics struct {
+	// TODO(alexg): influx metrics here?
 	otlpRequestCounter   *prometheus.CounterVec
 	uncompressedBodySize *prometheus.HistogramVec
 }

pkg/distributor/influx.go

@@ -0,0 +1,177 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package distributor
+
+import (
+	"context"
+	"errors"
+	"net/http"
+
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/grafana/dskit/grpcutil"
+	"github.com/grafana/dskit/httpgrpc"
+	"github.com/grafana/dskit/middleware"
+	io2 "github.com/influxdata/influxdb/v2/kit/io"
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/grafana/mimir/pkg/distributor/influxpush"
+	"github.com/grafana/mimir/pkg/mimirpb"
+	"github.com/grafana/mimir/pkg/util"
+	utillog "github.com/grafana/mimir/pkg/util/log"
+	"github.com/grafana/mimir/pkg/util/spanlogger"
+)
+
+func parser(ctx context.Context, r *http.Request, maxSize int, _ *util.RequestBuffers, req *mimirpb.PreallocWriteRequest, logger log.Logger) error {
+	spanLogger, ctx := spanlogger.NewWithLogger(ctx, logger, "Distributor.InfluxHandler.decodeAndConvert")
+	defer spanLogger.Span.Finish()
+
+	spanLogger.SetTag("content_type", r.Header.Get("Content-Type"))
+	spanLogger.SetTag("content_encoding", r.Header.Get("Content-Encoding"))
+	spanLogger.SetTag("content_length", r.ContentLength)
+
+	ts, bytesRead, err := influxpush.ParseInfluxLineReader(ctx, r, maxSize)
+	// TODO(alexg): one argument for splitting up the decoding and conversion is to facilitate granular timings
+	// right now since ParseInfluxLineReader() does both
+	// The otel version decodes the whole input and then processes it, the existing Influx code parses each line as it
+	// decodes it.
+	level.Debug(spanLogger).Log(
+		"msg", "decodeAndConvert complete",
+		"bytesRead", bytesRead,
+	)
+	if err != nil {
+		level.Error(logger).Log("err", err.Error())
+		// TODO(alexg): need to pass on the http.StatusBadRequest
+		// http.Error(w, err.Error(), http.StatusBadRequest)
+		return err
+	}
+
+	// Sigh, a write API optimisation needs me to jump through hoops.
+	pts := make([]mimirpb.PreallocTimeseries, 0, len(ts))
+	for i := range ts {
+		pts = append(pts, mimirpb.PreallocTimeseries{
+			TimeSeries: &ts[i],
+		})
+	}
+
+	level.Debug(spanLogger).Log(
+		"msg", "Influx to Prometheus conversion complete",
+		"metric_count", len(ts),
+	)
+
+	req.Timeseries = pts
+	return nil
+}
+
+// InfluxHandler is a http.Handler which accepts Influx Line protocol and converts it to WriteRequests.
+func InfluxHandler(
+	maxRecvMsgSize int,
+	requestBufferPool util.Pool,
+	sourceIPs *middleware.SourceIPExtractor,
+	retryCfg RetryConfig,
+	push PushFunc,
+	_ *PushMetrics, // TODO(alexg) add pushMetrics()
+	_ prometheus.Registerer, // TODO(alexg): add reg
+	logger log.Logger,
+) http.Handler {
+	//TODO(alexg): mirror otel.go implementation where we do decoding here rather than in parser() func?
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx := r.Context()
+		logger := utillog.WithContext(ctx, logger)
+		if sourceIPs != nil {
+			source := sourceIPs.Get(r)
+			if source != "" {
+				logger = utillog.WithSourceIPs(source, logger)
+			}
+		}
+
+		supplier := func() (*mimirpb.WriteRequest, func(), error) {
+			rb := util.NewRequestBuffers(requestBufferPool)
+			var req mimirpb.PreallocWriteRequest
+
+			if err := parser(ctx, r, maxRecvMsgSize, rb, &req, logger); err != nil {
+				// TODO(alexg): Do we even need httpgrpc here?
+				// Check for httpgrpc error, default to client error if parsing failed
+				if _, ok := httpgrpc.HTTPResponseFromError(err); !ok {
+					err = httpgrpc.Error(http.StatusBadRequest, err.Error())
+				}
+
+				rb.CleanUp()
+				return nil, nil, err
+			}
+
+			cleanup := func() {
+				mimirpb.ReuseSlice(req.Timeseries)
+				rb.CleanUp()
+			}
+			return &req.WriteRequest, cleanup, nil
+		}
+		req := newRequest(supplier)
+		// https://docs.influxdata.com/influxdb/cloud/api/v2/#tag/Response-codes
+		if err := push(ctx, req); err != nil {
+			if errors.Is(err, context.Canceled) {
+				level.Warn(logger).Log("msg", "push request canceled", "err", err)
+				w.WriteHeader(statusClientClosedRequest)
+				return
+			}
+			if errors.Is(err, io2.ErrReadLimitExceeded) {
+				// TODO(alexg): One thing we have seen in the past is that telegraf clients send a batch of data
+				// if it is too big they should respond to the 413 below, but if a client doesn't understand this
+				// it just sends the next batch that is even bigger. In the past this has had to be dealt with by
+				// adding rate limits to drop the payloads.
+				level.Warn(logger).Log("msg", "request too large", "err", err)
+				// TODO(alexg): max size and bytes received in error?
+				w.WriteHeader(http.StatusRequestEntityTooLarge)
+				return
+			}
+			// From: https://github.com/grafana/influx2cortex/blob/main/pkg/influx/errors.go
+
+			var (
+				httpCode int
+				errorMsg string
+			)
+			if st, ok := grpcutil.ErrorToStatus(err); ok {
+				// TODO(alexg): Hmm, still needed?
+				// This code is needed for a correct handling of errors returned by the supplier function.
+				// These errors are created by using the httpgrpc package.
+				httpCode = int(st.Code())
+				errorMsg = st.Message()
+			} else {
+				var distributorErr Error
+				errorMsg = err.Error()
+				if errors.Is(err, context.DeadlineExceeded) || !errors.As(err, &distributorErr) {
+					httpCode = http.StatusServiceUnavailable
+				} else {
+					httpCode = errorCauseToHTTPStatusCode(distributorErr.Cause(), false)
+				}
+			}
+			if httpCode != 202 {
+				// This error message is consistent with error message in Prometheus remote-write handler, and ingester's ingest-storage pushToStorage method.
+				msgs := []interface{}{"msg", "detected an error while ingesting Influx metrics request (the request may have been partially ingested)", "httpCode", httpCode, "err", err}
+				if httpCode/100 == 4 {
+					// TODO(alexg): what is this?
+					msgs = append(msgs, "insight", true)
+				}
+				level.Error(logger).Log(msgs...)
+			}
+			if httpCode < 500 {
+				level.Info(logger).Log("msg", errorMsg, "response_code", httpCode, "err", tryUnwrap(err))
+			} else if httpCode >= 500 {
+				level.Warn(logger).Log("msg", errorMsg, "response_code", httpCode, "err", tryUnwrap(err))
+			}
+			addHeaders(w, err, r, httpCode, retryCfg)
+			w.WriteHeader(httpCode)
+		} else {
+			w.WriteHeader(http.StatusNoContent) // Needed for Telegraf, otherwise it tries to marshal JSON and considers the write a failure.
+		}
+	})
+}
+
+// Imported from: https://github.com/grafana/influx2cortex/blob/main/pkg/influx/errors.go
+
+func tryUnwrap(err error) error {
+	if wrapped, ok := err.(interface{ Unwrap() error }); ok {
+		return wrapped.Unwrap()
+	}
+	return err
+}

pkg/distributor/influx_test.go

@@ -0,0 +1,147 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package distributor
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/go-kit/log"
+	io2 "github.com/influxdata/influxdb/v2/kit/io"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/grafana/mimir/pkg/mimirpb"
+)
+
+func TestInfluxHandleSeriesPush(t *testing.T) {
+	defaultExpectedWriteRequest := &mimirpb.WriteRequest{
+		Timeseries: []mimirpb.PreallocTimeseries{
+			{
+				TimeSeries: &mimirpb.TimeSeries{
+					Labels: []mimirpb.LabelAdapter{
+						{Name: "__mimir_source__", Value: "influx"},
+						{Name: "__name__", Value: "measurement_f1"},
+						{Name: "t1", Value: "v1"},
+					},
+					Samples: []mimirpb.Sample{
+						{Value: 2, TimestampMs: 1465839830100},
+					},
+				},
+			},
+		},
+	}
+
+	tests := []struct {
+		name                string
+		url                 string
+		data                string
+		expectedCode        int
+		push                func(t *testing.T) PushFunc
+		maxRequestSizeBytes int
+	}{
+		{
+			name:         "POST",
+			url:          "/write",
+			data:         "measurement,t1=v1 f1=2 1465839830100400200",
+			expectedCode: http.StatusNoContent,
+			push: func(t *testing.T) PushFunc {
+				return func(_ context.Context, pushReq *Request) error {
+					req, err := pushReq.WriteRequest()
+					assert.Equal(t, defaultExpectedWriteRequest, req)
+					assert.Nil(t, err)
+					return err
+				}
+			},
+			maxRequestSizeBytes: 1 << 20,
+		},
+		{
+			name:         "POST with precision",
+			url:          "/write?precision=ns",
+			data:         "measurement,t1=v1 f1=2 1465839830100400200",
+			expectedCode: http.StatusNoContent,
+			push: func(t *testing.T) PushFunc {
+				return func(_ context.Context, pushReq *Request) error {
+					req, err := pushReq.WriteRequest()
+					assert.Equal(t, defaultExpectedWriteRequest, req)
+					assert.Nil(t, err)
+					return err
+				}
+			},
+			maxRequestSizeBytes: 1 << 20,
+		},
+		{
+			name:         "invalid parsing error handling",
+			url:          "/write",
+			data:         "measurement,t1=v1 f1= 1465839830100400200",
+			expectedCode: http.StatusBadRequest,
+			push: func(t *testing.T) PushFunc {
+				return func(_ context.Context, pushReq *Request) error {
+					req, err := pushReq.WriteRequest()
+					assert.Nil(t, req)
+					// TODO(alexg): assert on specific err
+					// assert.NoError(t) // reminder to fix
+					return err
+				}
+			},
+			maxRequestSizeBytes: 1 << 20,
+		},
+		{
+			name:         "invalid query params",
+			url:          "/write?precision=?",
+			data:         "measurement,t1=v1 f1=2 1465839830100400200",
+			expectedCode: http.StatusBadRequest,
+			push: func(t *testing.T) PushFunc {
+				// return func(ctx context.Context, req *mimirpb.WriteRequest) error {
+				return func(_ context.Context, pushReq *Request) error {
+					req, err := pushReq.WriteRequest()
+					assert.Nil(t, req)
+					// TODO(alexg): assert on specific err
+					// assert.NoError(t, err) // reminder to fix
+					return err
+				}
+			},
+			maxRequestSizeBytes: 1 << 20,
+		},
+		{
+			name:         "internal server error",
+			url:          "/write",
+			data:         "measurement,t1=v1 f1=2 1465839830100400200",
+			expectedCode: http.StatusServiceUnavailable,
+			push: func(t *testing.T) PushFunc {
+				return func(_ context.Context, _ *Request) error {
+					assert.Error(t, context.DeadlineExceeded)
+					return context.DeadlineExceeded
+				}
+			},
+			maxRequestSizeBytes: 1 << 20,
+		},
+		{
+			name:         "max batch size violated",
+			url:          "/write",
+			data:         "measurement,t1=v1 f1=2 0123456789",
+			expectedCode: http.StatusBadRequest,
+			push: func(t *testing.T) PushFunc {
+				return func(_ context.Context, pushReq *Request) error {
+					req, err := pushReq.WriteRequest()
+					assert.Nil(t, req)
+					assert.Error(t, io2.ErrReadLimitExceeded)
+					return err
+				}
+			},
+			maxRequestSizeBytes: 10,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handler := InfluxHandler(tt.maxRequestSizeBytes, nil, nil, RetryConfig{}, tt.push(t), nil, nil, log.NewNopLogger())
+			req := httptest.NewRequest("POST", tt.url, bytes.NewReader([]byte(tt.data)))
+			rec := httptest.NewRecorder()
+			handler.ServeHTTP(rec, req)
+			assert.Equal(t, tt.expectedCode, rec.Code)
+		})
+	}
+}