Skip to content

Commit c8b83d2

Browse files
christopherzlichristopherzli
committed
query_bytes_fetched
add thanos stats queried
1 parent da7ccc1 commit c8b83d2

File tree

5 files changed

+63
-49
lines changed

5 files changed

+63
-49
lines changed

‎pkg/api/query/v1.go‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -295,8 +295,9 @@ type queryData struct {
295295
Result parser.Value `json:"result"`
296296
Stats stats.QueryStats `json:"stats,omitempty"`
297297
// Additional Thanos Response field.
298-
QueryAnalysis queryTelemetry `json:"analysis,omitempty"`
299-
Warnings []error `json:"warnings,omitempty"`
298+
QueryAnalysis queryTelemetry `json:"analysis,omitempty"`
299+
Warnings []error `json:"warnings,omitempty"`
300+
SeriesStatsCounter storepb.SeriesStatsCounter `json:"seriesStatsCounter,omitempty"`
300301
}
301302

302303
type queryTelemetry struct {
@@ -705,6 +706,7 @@ func (qapi *QueryAPI) query(r *http.Request) (interface{}, []error, *api.ApiErro
705706
for i := range seriesStats {
706707
aggregator.Aggregate(seriesStats[i])
707708
}
709+
seriesStatsCounter := aggregator.GetSeriesStatsCounter()
708710
aggregator.Observe(time.Since(beforeRange).Seconds())
709711

710712
// Optional stats field in response if parameter "stats" is not empty.
@@ -713,10 +715,11 @@ func (qapi *QueryAPI) query(r *http.Request) (interface{}, []error, *api.ApiErro
713715
qs = stats.NewQueryStats(qry.Stats())
714716
}
715717
return &queryData{
716-
ResultType: res.Value.Type(),
717-
Result: res.Value,
718-
Stats: qs,
719-
QueryAnalysis: analysis,
718+
ResultType: res.Value.Type(),
719+
Result: res.Value,
720+
Stats: qs,
721+
QueryAnalysis: analysis,
722+
SeriesStatsCounter: seriesStatsCounter,
720723
}, res.Warnings.AsErrors(), nil, qry.Close
721724
}
722725

@@ -1004,6 +1007,7 @@ func (qapi *QueryAPI) queryRange(r *http.Request) (interface{}, []error, *api.Ap
10041007
for i := range seriesStats {
10051008
aggregator.Aggregate(seriesStats[i])
10061009
}
1010+
seriesStatsCounter := aggregator.GetSeriesStatsCounter()
10071011
aggregator.Observe(time.Since(beforeRange).Seconds())
10081012

10091013
// Optional stats field in response if parameter "stats" is not empty.
@@ -1012,10 +1016,11 @@ func (qapi *QueryAPI) queryRange(r *http.Request) (interface{}, []error, *api.Ap
10121016
qs = stats.NewQueryStats(qry.Stats())
10131017
}
10141018
return &queryData{
1015-
ResultType: res.Value.Type(),
1016-
Result: res.Value,
1017-
Stats: qs,
1018-
QueryAnalysis: analysis,
1019+
ResultType: res.Value.Type(),
1020+
Result: res.Value,
1021+
Stats: qs,
1022+
QueryAnalysis: analysis,
1023+
SeriesStatsCounter: seriesStatsCounter,
10191024
}, res.Warnings.AsErrors(), nil, qry.Close
10201025
}
10211026

‎pkg/query/querier.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -206,7 +206,7 @@ func (s *seriesServer) Send(r *storepb.SeriesResponse) error {
206206

207207
if r.GetSeries() != nil {
208208
s.seriesSet = append(s.seriesSet, *r.GetSeries())
209-
s.seriesSetStats.Count(r.GetSeries())
209+
s.seriesSetStats.Count(r)
210210
return nil
211211
}
212212

‎pkg/store/proxy_heap.go‎

Lines changed: 4 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -392,14 +392,13 @@ func newLazyRespSet(
392392
}
393393

394394
go func(st string, l *lazyRespSet) {
395-
bytesProcessed := 0
396395
seriesStats := &storepb.SeriesStatsCounter{}
397396

398397
defer func() {
399398
l.span.SetTag("processed.series", seriesStats.Series)
400399
l.span.SetTag("processed.chunks", seriesStats.Chunks)
401400
l.span.SetTag("processed.samples", seriesStats.Samples)
402-
l.span.SetTag("processed.bytes", bytesProcessed)
401+
l.span.SetTag("processed.bytes", seriesStats.Bytes)
403402
l.span.Finish()
404403
}()
405404

@@ -459,15 +458,12 @@ func newLazyRespSet(
459458
}
460459

461460
numResponses++
462-
bytesProcessed += resp.Size()
463461

464462
if resp.GetSeries() != nil && applySharding && !shardMatcher.MatchesZLabels(resp.GetSeries().Labels) {
465463
return true
466464
}
467465

468-
if resp.GetSeries() != nil {
469-
seriesStats.Count(resp.GetSeries())
470-
}
466+
seriesStats.Count(resp)
471467

472468
l.bufferedResponsesMtx.Lock()
473469
l.bufferedResponses = append(l.bufferedResponses, resp)
@@ -675,13 +671,12 @@ func newEagerRespSet(
675671
// Start a goroutine and immediately buffer everything.
676672
go func(l *eagerRespSet) {
677673
seriesStats := &storepb.SeriesStatsCounter{}
678-
bytesProcessed := 0
679674

680675
defer func() {
681676
l.span.SetTag("processed.series", seriesStats.Series)
682677
l.span.SetTag("processed.chunks", seriesStats.Chunks)
683678
l.span.SetTag("processed.samples", seriesStats.Samples)
684-
l.span.SetTag("processed.bytes", bytesProcessed)
679+
l.span.SetTag("processed.bytes", seriesStats.Bytes)
685680
l.span.Finish()
686681
ret.wg.Done()
687682
}()
@@ -728,15 +723,12 @@ func newEagerRespSet(
728723
}
729724

730725
numResponses++
731-
bytesProcessed += resp.Size()
732726

733727
if resp.GetSeries() != nil && applySharding && !shardMatcher.MatchesZLabels(resp.GetSeries().Labels) {
734728
return true
735729
}
736730

737-
if resp.GetSeries() != nil {
738-
seriesStats.Count(resp.GetSeries())
739-
}
731+
seriesStats.Count(resp)
740732

741733
l.bufferedResponses = append(l.bufferedResponses, resp)
742734
return true

‎pkg/store/storepb/custom.go‎

Lines changed: 33 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -484,6 +484,7 @@ type SeriesStatsCounter struct {
484484
Series int
485485
Chunks int
486486
Samples int
487+
Bytes uint64
487488
}
488489

489490
func (c *SeriesStatsCounter) CountSeries(seriesLabels []labelpb.ZLabel) {
@@ -494,39 +495,45 @@ func (c *SeriesStatsCounter) CountSeries(seriesLabels []labelpb.ZLabel) {
494495
}
495496
}
496497

497-
func (c *SeriesStatsCounter) Count(series *Series) {
498-
c.CountSeries(series.Labels)
499-
for _, chk := range series.Chunks {
500-
if chk.Raw != nil {
501-
c.Chunks++
502-
c.Samples += chk.Raw.XORNumSamples()
503-
}
498+
func (c *SeriesStatsCounter) Count(r *SeriesResponse) {
499+
if r.GetSeries() != nil {
500+
series := r.GetSeries()
501+
c.CountSeries(series.Labels)
502+
for _, chk := range series.Chunks {
503+
if chk.Raw != nil {
504+
c.Chunks++
505+
c.Samples += chk.Raw.XORNumSamples()
506+
}
504507

505-
if chk.Count != nil {
506-
c.Chunks++
507-
c.Samples += chk.Count.XORNumSamples()
508-
}
508+
if chk.Count != nil {
509+
c.Chunks++
510+
c.Samples += chk.Count.XORNumSamples()
511+
}
509512

510-
if chk.Counter != nil {
511-
c.Chunks++
512-
c.Samples += chk.Counter.XORNumSamples()
513-
}
513+
if chk.Counter != nil {
514+
c.Chunks++
515+
c.Samples += chk.Counter.XORNumSamples()
516+
}
514517

515-
if chk.Max != nil {
516-
c.Chunks++
517-
c.Samples += chk.Max.XORNumSamples()
518-
}
518+
if chk.Max != nil {
519+
c.Chunks++
520+
c.Samples += chk.Max.XORNumSamples()
521+
}
519522

520-
if chk.Min != nil {
521-
c.Chunks++
522-
c.Samples += chk.Min.XORNumSamples()
523-
}
523+
if chk.Min != nil {
524+
c.Chunks++
525+
c.Samples += chk.Min.XORNumSamples()
526+
}
524527

525-
if chk.Sum != nil {
526-
c.Chunks++
527-
c.Samples += chk.Sum.XORNumSamples()
528+
if chk.Sum != nil {
529+
c.Chunks++
530+
c.Samples += chk.Sum.XORNumSamples()
531+
}
528532
}
529533
}
534+
535+
//aggregate # of bytes fetched
536+
c.Bytes += uint64(r.Size())
530537
}
531538

532539
func (m *SeriesRequest) ToPromQL() string {

‎pkg/store/telemetry.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,10 @@ type seriesStatsAggregator struct {
2424
seriesStats storepb.SeriesStatsCounter
2525
}
2626

27+
func (s *seriesStatsAggregator) GetSeriesStatsCounter() storepb.SeriesStatsCounter {
28+
return s.seriesStats
29+
}
30+
2731
type seriesStatsAggregatorFactory struct {
2832
queryDuration *prometheus.HistogramVec
2933
seriesLeBuckets []float64
@@ -81,6 +85,7 @@ func (s *seriesStatsAggregator) Aggregate(stats storepb.SeriesStatsCounter) {
8185
s.seriesStats.Series += stats.Series
8286
s.seriesStats.Samples += stats.Samples
8387
s.seriesStats.Chunks += stats.Chunks
88+
s.seriesStats.Bytes += stats.Bytes
8489
}
8590

8691
// Observe commits the aggregated SeriesStatsCounter as an observation.
@@ -124,11 +129,16 @@ type SeriesQueryPerformanceMetricsAggregatorFactory interface {
124129
type SeriesQueryPerformanceMetricsAggregator interface {
125130
Aggregate(seriesStats storepb.SeriesStatsCounter)
126131
Observe(duration float64)
132+
GetSeriesStatsCounter() storepb.SeriesStatsCounter
127133
}
128134

129135
// NoopSeriesStatsAggregator is a query performance series aggregator that does nothing.
130136
type NoopSeriesStatsAggregator struct{}
131137

138+
func (s *NoopSeriesStatsAggregator) GetSeriesStatsCounter() storepb.SeriesStatsCounter {
139+
return storepb.SeriesStatsCounter{}
140+
}
141+
132142
func (s *NoopSeriesStatsAggregator) Aggregate(_ storepb.SeriesStatsCounter) {}
133143

134144
func (s *NoopSeriesStatsAggregator) Observe(_ float64) {}

0 commit comments

Comments
 (0)