RemoteReadHandler handles Prometheus remote read requests.
(q storage.Queryable, logger log.Logger)
| 18 | |
| 19 | // RemoteReadHandler handles Prometheus remote read requests. |
| 20 | func RemoteReadHandler(q storage.Queryable, logger log.Logger) http.Handler { |
| 21 | return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 22 | ctx := r.Context() |
| 23 | var req client.ReadRequest |
| 24 | logger := util_log.WithContext(r.Context(), logger) |
| 25 | if err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRemoteReadQuerySize, &req, util.RawSnappy); err != nil { |
| 26 | level.Error(logger).Log("msg", "failed to parse proto", "err", err.Error()) |
| 27 | http.Error(w, err.Error(), http.StatusBadRequest) |
| 28 | return |
| 29 | } |
| 30 | |
| 31 | // Fetch samples for all queries in parallel. |
| 32 | resp := client.ReadResponse{ |
| 33 | Results: make([]*client.QueryResponse, len(req.Queries)), |
| 34 | } |
| 35 | errors := make(chan error) |
| 36 | for i, qr := range req.Queries { |
| 37 | go func(i int, qr *client.QueryRequest) { |
| 38 | from, to, matchers, err := client.FromQueryRequest(storecache.NoopMatchersCache, qr) |
| 39 | if err != nil { |
| 40 | errors <- err |
| 41 | return |
| 42 | } |
| 43 | |
| 44 | querier, err := q.Querier(int64(from), int64(to)) |
| 45 | if err != nil { |
| 46 | errors <- err |
| 47 | return |
| 48 | } |
| 49 | |
| 50 | params := &storage.SelectHints{ |
| 51 | Start: int64(from), |
| 52 | End: int64(to), |
| 53 | } |
| 54 | seriesSet := querier.Select(ctx, false, params, matchers...) |
| 55 | resp.Results[i], err = client.SeriesSetToQueryResponse(seriesSet) |
| 56 | errors <- err |
| 57 | }(i, qr) |
| 58 | } |
| 59 | |
| 60 | var lastErr error |
| 61 | for range req.Queries { |
| 62 | err := <-errors |
| 63 | if err != nil { |
| 64 | lastErr = err |
| 65 | } |
| 66 | } |
| 67 | if lastErr != nil { |
| 68 | http.Error(w, lastErr.Error(), http.StatusBadRequest) |
| 69 | return |
| 70 | } |
| 71 | w.Header().Add("Content-Type", "application/x-protobuf") |
| 72 | if err := util.SerializeProtoResponse(w, &resp, util.RawSnappy); err != nil { |
| 73 | level.Error(logger).Log("msg", "error sending remote read response", "err", err) |
| 74 | } |
| 75 | }) |
| 76 | } |