QueryStream implements service.IngesterServer Streams metrics from a TSDB. This implements the client.IngesterServer interface
(req *client.QueryRequest, stream client.Ingester_QueryStreamServer)
| 2403 | // QueryStream implements service.IngesterServer |
| 2404 | // Streams metrics from a TSDB. This implements the client.IngesterServer interface |
| 2405 | func (i *Ingester) QueryStream(req *client.QueryRequest, stream client.Ingester_QueryStreamServer) (err error) { |
| 2406 | defer recoverIngester(i.logger, &err) |
| 2407 | defer req.Free() |
| 2408 | |
| 2409 | if err = i.checkRunning(); err != nil { |
| 2410 | return err |
| 2411 | } |
| 2412 | |
| 2413 | spanlog, ctx := spanlogger.New(stream.Context(), "QueryStream") |
| 2414 | defer spanlog.Finish() |
| 2415 | |
| 2416 | userID, err := users.TenantID(ctx) |
| 2417 | if err != nil { |
| 2418 | return err |
| 2419 | } |
| 2420 | |
| 2421 | // Set pprof labels for profiling |
| 2422 | pprof.Do(ctx, pprof.Labels("user", userID), func(ctx context.Context) { |
| 2423 | err = i.queryStream(ctx, userID, req, stream, spanlog) |
| 2424 | }) |
| 2425 | |
| 2426 | return err |
| 2427 | } |
| 2428 | |
| 2429 | func (i *Ingester) queryStream(ctx context.Context, userID string, req *client.QueryRequest, stream client.Ingester_QueryStreamServer, spanlog *spanlogger.SpanLogger) error { |
| 2430 | from, through, matchers, err := client.FromQueryRequest(i.matchersCache, req) |
nothing calls this directly
no test coverage detected