func()

in web/api/v1/api.go [1086:1196]
97 lines of code
18 McCabe index (conditional complexity)

func (api *API) remoteRead(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	if err := api.remoteReadGate.Start(ctx); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	remoteReadQueries.Inc()

	defer api.remoteReadGate.Done()
	defer remoteReadQueries.Dec()

	req, err := remote.DecodeReadRequest(r)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	externalLabels := api.config().GlobalConfig.ExternalLabels.Map()

	sortedExternalLabels := make([]prompb.Label, 0, len(externalLabels))
	for name, value := range externalLabels {
		sortedExternalLabels = append(sortedExternalLabels, prompb.Label{
			Name:  string(name),
			Value: string(value),
		})
	}
	sort.Slice(sortedExternalLabels, func(i, j int) bool {
		return sortedExternalLabels[i].Name < sortedExternalLabels[j].Name
	})

	responseType, err := remote.NegotiateResponseType(req.AcceptedResponseTypes)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	switch responseType {
	case prompb.ReadRequest_STREAMED_XOR_CHUNKS:
		w.Header().Set("Content-Type", "application/x-streamed-protobuf; proto=prometheus.ChunkedReadResponse")

		f, ok := w.(http.Flusher)
		if !ok {
			http.Error(w, "internal http.ResponseWriter does not implement http.Flusher interface", http.StatusInternalServerError)
			return
		}
		for i, query := range req.Queries {
			err := api.remoteReadQuery(ctx, query, externalLabels, func(querier storage.Querier, selectParams *storage.SelectParams, filteredMatchers []*labels.Matcher) error {
				// The streaming API provides sorted series.
				set, _, err := querier.SelectSorted(selectParams, filteredMatchers...)
				if err != nil {
					return err
				}

				return remote.StreamChunkedReadResponses(
					remote.NewChunkedWriter(w, f),
					int64(i),
					set,
					sortedExternalLabels,
					api.remoteReadMaxBytesInFrame,
				)
			})
			if err != nil {
				if httpErr, ok := err.(remote.HTTPError); ok {
					http.Error(w, httpErr.Error(), httpErr.Status())
					return
				}
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
		}
	default:
		w.Header().Set("Content-Type", "application/x-protobuf")
		w.Header().Set("Content-Encoding", "snappy")

		// On empty or unknown types in req.AcceptedResponseTypes we default to non streamed, raw samples response.
		resp := prompb.ReadResponse{
			Results: make([]*prompb.QueryResult, len(req.Queries)),
		}
		for i, query := range req.Queries {
			err := api.remoteReadQuery(ctx, query, externalLabels, func(querier storage.Querier, selectParams *storage.SelectParams, filteredMatchers []*labels.Matcher) error {
				set, _, err := querier.Select(selectParams, filteredMatchers...)
				if err != nil {
					return err
				}

				resp.Results[i], err = remote.ToQueryResult(set, api.remoteReadSampleLimit)
				if err != nil {
					return err
				}

				for _, ts := range resp.Results[i].Timeseries {
					ts.Labels = remote.MergeLabels(ts.Labels, sortedExternalLabels)
				}
				return nil
			})
			if err != nil {
				if httpErr, ok := err.(remote.HTTPError); ok {
					http.Error(w, httpErr.Error(), httpErr.Status())
					return
				}
				http.Error(w, err.Error(), http.StatusInternalServerError)
				return
			}
		}

		if err := remote.EncodeReadResponse(&resp, w); err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
	}
}