Function Handler

pkg/util/push/push.go:49–200

Handler is a http.Handler which accepts WriteRequests.

(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool, maxRecvMsgSize int, overrides *validation.Overrides, sourceIPs *middleware.SourceIPExtractor, push Func, requestTotal *prometheus.CounterVec)

// Handler is a http.Handler which accepts WriteRequests.
func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool, maxRecvMsgSize int, overrides *validation.Overrides, sourceIPs *middleware.SourceIPExtractor, push Func, requestTotal *prometheus.CounterVec) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()
		logger := log.WithContext(ctx, log.Logger)
		if sourceIPs != nil {
			source := sourceIPs.Get(r)
			if source != "" {
				ctx = util.AddSourceIPsToOutgoingContext(ctx, source)
				logger = log.WithSourceIPs(source, logger)
			}
		}

		handlePRW1 := func() {
			var req cortexpb.PreallocWriteRequest
			err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy)
			if err != nil {
				level.Error(logger).Log("err", err.Error())
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}

			req.SkipLabelNameValidation = false
			if req.Source == 0 {
				req.Source = cortexpb.API
			}

			if _, err := push(ctx, &req.WriteRequest); err != nil {
				resp, ok := httpgrpc.HTTPResponseFromError(err)
				if !ok {
					http.Error(w, err.Error(), http.StatusInternalServerError)
					return
				}
				if resp.GetCode()/100 == 5 {
					level.Error(logger).Log("msg", "push error", "err", err)
				} else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests {
					level.Warn(logger).Log("msg", "push refused", "err", err)
				}
				http.Error(w, string(resp.Body), int(resp.Code))
			}
		}

		handlePRW2 := func() {
			userID, err := users.TenantID(ctx)
			if err != nil {
				return
			}

			req := cortexpb.PreallocWriteRequestV2FromPool()
			// v1 request is put back into the pool by the Distributor.
			defer cortexpb.ReuseWriteRequestV2(req)

			err = util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, req, util.RawSnappy)
			if err != nil {
				level.Error(logger).Log("err", err.Error())
				http.Error(w, err.Error(), http.StatusBadRequest)
				return
			}