queryStreamChunks streams metrics from a TSDB. This implements the client.IngesterServer interface
(ctx context.Context, userID string, db *userTSDB, from, through int64, matchers []*labels.Matcher, sm *storepb.ShardMatcher, stream client.Ingester_QueryStreamServer)
| 2600 | |
| 2601 | // queryStreamChunks streams metrics from a TSDB. This implements the client.IngesterServer interface |
| 2602 | func (i *Ingester) queryStreamChunks(ctx context.Context, userID string, db *userTSDB, from, through int64, matchers []*labels.Matcher, sm *storepb.ShardMatcher, stream client.Ingester_QueryStreamServer) (numSeries, numSamples, totalBatchSizeBytes, numChunks int, _ error) { |
| 2603 | q, err := db.ChunkQuerier(from, through) |
| 2604 | if err != nil { |
| 2605 | return 0, 0, 0, 0, err |
| 2606 | } |
| 2607 | defer q.Close() |
| 2608 | |
| 2609 | // Check regex matcher limits before executing query if enabled |
| 2610 | if i.cfg.EnableRegexMatcherLimits { |
| 2611 | if err := i.checkRegexMatcherLimits(ctx, userID, db, matchers, from, through); err != nil { |
| 2612 | return 0, 0, 0, 0, err |
| 2613 | } |
| 2614 | } |
| 2615 | |
| 2616 | c, err := i.trackInflightQueryRequest() |
| 2617 | if err != nil { |
| 2618 | return 0, 0, 0, 0, err |
| 2619 | } |
| 2620 | hints := &storage.SelectHints{ |
| 2621 | Start: from, |
| 2622 | End: through, |
| 2623 | DisableTrimming: i.cfg.DisableChunkTrimming, |
| 2624 | } |
| 2625 | var lazyMatchers []*labels.Matcher |
| 2626 | if i.cfg.EnableMatcherOptimization { |
| 2627 | matchers, lazyMatchers = optimizeMatchers(matchers) |
| 2628 | } |
| 2629 | // It's not required to return sorted series because series are sorted by the Cortex querier. |
| 2630 | ss := q.Select(ctx, false, hints, matchers...) |
| 2631 | c() |
| 2632 | if ss.Err() != nil { |
| 2633 | return 0, 0, 0, 0, ss.Err() |
| 2634 | } |
| 2635 | |
| 2636 | chunkSeries := getTimeSeriesChunksSlice() |
| 2637 | defer putTimeSeriesChunksSlice(chunkSeries) |
| 2638 | batchSizeBytes := 0 |
| 2639 | var it chunks.Iterator |
| 2640 | |
| 2641 | now := time.Now() |
| 2642 | // Check sampling decision early to avoid calculating hashes if batch will be skipped |
| 2643 | var queriedSeriesHashes []uint64 |
| 2644 | sampled := false |
| 2645 | if db.activeQueriedSeries != nil { |
| 2646 | sampled = db.activeQueriedSeries.SampleRequest() |
| 2647 | if sampled { |
| 2648 | queriedSeriesHashes = getQueriedSeriesHashesSlice() |
| 2649 | } |
| 2650 | } |
| 2651 | |
| 2652 | for ss.Next() { |
| 2653 | series := ss.At() |
| 2654 | lbls := series.Labels() |
| 2655 | |
| 2656 | if !labelsMatches(lbls, lazyMatchers) { |
| 2657 | continue |
| 2658 | } |
| 2659 |