Method queryStream

pkg/ingester/ingester.go:2429–2473

(ctx context.Context, userID string, req *client.QueryRequest, stream client.Ingester_QueryStreamServer, spanlog *spanlogger.SpanLogger)

}

func (i *Ingester) queryStream(ctx context.Context, userID string, req *client.QueryRequest, stream client.Ingester_QueryStreamServer, spanlog *spanlogger.SpanLogger) error {
	from, through, matchers, err := client.FromQueryRequest(i.matchersCache, req)
	if err != nil {
		return err
	}

	matchers, shardMatcher, err := querysharding.ExtractShardingMatchers(matchers)
	if err != nil {
		return err
	}

	defer shardMatcher.Close()

	i.metrics.queries.Inc()

	db, err := i.getTSDB(userID)
	if err != nil || db == nil {
		return nil
	}

	if err := db.acquireReadLock(); err != nil {
		return nil
	}
	defer db.releaseReadLock()

	numSamples := 0
	numSeries := 0
	totalDataBytes := 0
	numChunks := 0
	numSeries, numSamples, totalDataBytes, numChunks, err = i.queryStreamChunks(ctx, userID, db, int64(from), int64(through), matchers, shardMatcher, stream)

	if err != nil {
		return err
	}

	i.metrics.queriedSeries.Observe(float64(numSeries))
	i.metrics.queriedSamples.Observe(float64(numSamples))
	i.metrics.queriedChunks.Observe(float64(numChunks))
	level.Debug(spanlog).Log("series", numSeries, "samples", numSamples, "data_bytes", totalDataBytes, "chunks", numChunks)
	spanlog.SetTag("series", numSeries)
	spanlog.SetTag("samples", numSamples)
	spanlog.SetTag("data_bytes", totalDataBytes)
	spanlog.SetTag("chunks", numChunks)
	return nil
}

func (i *Ingester) trackInflightQueryRequest() (func(), error) {
	gl := i.getInstanceLimits()