GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/pint

Public

mirrored from https://github.com/cloudflare/pintAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
v0.18.0

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/range.go

143lines · modecode

RawDownload
Latest commit unavailable.
1package promapi
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "strings"
8 "time"
9
10 v1 "github.com/prometheus/client_golang/api/prometheus/v1"
11 "github.com/prometheus/common/model"
12 "github.com/rs/zerolog/log"
13
14 "github.com/cloudflare/pint/internal/output"
15)
16
17type RangeQueryResult struct {
18 URI string
19 Samples []*model.SampleStream
20 Start time.Time
21 End time.Time
22 DurationSeconds float64
23}
24
25func (p *Prometheus) RangeQuery(ctx context.Context, expr string, lookback, step time.Duration) (*RangeQueryResult, error) {
26 log.Debug().
27 Str("uri", p.uri).
28 Str("query", expr).
29 Str("delta", output.HumanizeDuration(lookback)).
30 Str("step", output.HumanizeDuration(step)).
31 Msg("Scheduling prometheus range query")
32
33 lockKey := "/api/v1/query/range"
34 p.lock.lock(lockKey)
35 defer p.lock.unlock(lockKey)
36
37 cacheKey := strings.Join([]string{expr, lookback.String(), step.String()}, "\n")
38 return p.realRangeQuery(ctx, expr, lookback, step, cacheKey, false)
39}
40
41func (p *Prometheus) realRangeQuery(
42 ctx context.Context,
43 expr string, lookback, step time.Duration,
44 cacheKey string, isRetry bool,
45) (*RangeQueryResult, error) {
46 if v, ok := p.cache.Get(cacheKey); ok {
47 log.Debug().
48 Str("uri", p.uri).
49 Str("query", expr).
50 Str("delta", output.HumanizeDuration(lookback)).
51 Str("step", output.HumanizeDuration(step)).
52 Msg("Cache hit")
53 prometheusCacheHitsTotal.WithLabelValues(p.name, "/api/v1/query_range").Inc()
54 r := v.(RangeQueryResult)
55 return &r, nil
56 }
57 log.Debug().
58 Str("uri", p.uri).
59 Str("query", expr).
60 Str("delta", output.HumanizeDuration(lookback)).
61 Str("step", output.HumanizeDuration(step)).
62 Msg("Cache miss")
63
64 prometheusQueriesTotal.WithLabelValues(p.name, "/api/v1/query_range").Inc()
65 now := time.Now()
66 r := v1.Range{
67 Start: now.Add(lookback * -1),
68 End: now,
69 Step: step,
70 }
71
72 if !isRetry {
73 p.slowQueryLock.Lock()
74 if v, ok := p.slowQueryCache.Get(expr); ok {
75 log.Debug().
76 Str("query", expr).
77 Str("delta", output.HumanizeDuration(v.(time.Duration))).
78 Str("start", r.Start.String()).
79 Str("cached", r.End.Add(v.(time.Duration)*-1).String()).
80 Msg("Got cached slow query delta")
81 r.Start = r.End.Add(v.(time.Duration) * -1)
82 }
83 p.slowQueryLock.Unlock()
84 }
85
86 rctx, cancel := context.WithTimeout(ctx, p.timeout)
87 defer cancel()
88
89 log.Debug().
90 Str("uri", p.uri).
91 Str("query", expr).
92 Str("delta", output.HumanizeDuration(lookback)).
93 Bool("retry", isRetry).
94 Msg("Executing range query")
95 qstart := time.Now()
96 result, _, err := p.api.QueryRange(rctx, expr, r)
97 duration := time.Since(qstart)
98 log.Debug().
99 Str("uri", p.uri).
100 Str("query", expr).
101 Str("duration", output.HumanizeDuration(duration)).
102 Msg("Range query completed")
103 if err != nil {
104 log.Error().Err(err).Str("uri", p.uri).Str("query", expr).Msg("Range query failed")
105 prometheusQueryErrorsTotal.WithLabelValues(p.name, "/api/v1/query_range", errReason(err)).Inc()
106 if delta, retryOK := CanRetryError(err, lookback); retryOK {
107 if delta < step*2 {
108 log.Error().Str("uri", p.uri).Str("query", expr).Msg("No more retries possible")
109 return nil, errors.New("no more retries possible")
110 }
111 log.Warn().Str("delta", output.HumanizeDuration(delta)).Msg("Retrying request with smaller range")
112 p.slowQueryLock.Lock()
113 p.slowQueryCache.Remove(expr)
114 p.slowQueryCache.Add(expr, delta)
115 p.slowQueryLock.Unlock()
116 return p.realRangeQuery(ctx, expr, delta, step, cacheKey, true)
117 }
118 return nil, err
119 }
120
121 qr := RangeQueryResult{
122 URI: p.uri,
123 DurationSeconds: duration.Seconds(),
124 Start: r.Start,
125 End: r.End,
126 }
127
128 switch result.Type() {
129 case model.ValMatrix:
130 samples := result.(model.Matrix)
131 qr.Samples = samples
132 default:
133 log.Error().Err(err).Str("uri", p.uri).Str("query", expr).Msgf("Range query returned unknown result type: %v", result.Type())
134 prometheusQueryErrorsTotal.WithLabelValues(p.name, "/api/v1/query_range", "unknown result type").Inc()
135 return nil, fmt.Errorf("unknown result type: %v", result.Type())
136 }
137 log.Debug().Str("uri", p.uri).Str("query", expr).Int("samples", len(qr.Samples)).Msg("Parsed range response")
138
139 log.Debug().Str("query", expr).Str("uri", p.uri).Msg("Range query cache miss")
140 p.cache.Add(cacheKey, qr)
141
142 return &qr, nil
143}
144