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.83.0

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/failover.go

320lines · modecode

RawDownload
Latest commit unavailable.
1package promapi
2
3import (
4 "context"
5 "errors"
6 "log/slog"
7 "regexp"
8 "slices"
9 "sync"
10 "time"
11
12 "github.com/prometheus/client_golang/prometheus"
13)
14
15type FailoverGroupError struct {
16 err error
17 uri string
18 isStrict bool
19}
20
21func (e *FailoverGroupError) Unwrap() error {
22 return e.err
23}
24
25func (e *FailoverGroupError) Error() string {
26 return e.err.Error()
27}
28
29func (e *FailoverGroupError) URI() string {
30 return e.uri
31}
32
33func (e *FailoverGroupError) IsStrict() bool {
34 return e.isStrict
35}
36
37func cacheCleaner(cache *queryCache, interval time.Duration, quit chan bool) {
38 ticker := time.NewTicker(interval)
39 for {
40 select {
41 case <-quit:
42 return
43 case <-ticker.C:
44 cache.gc()
45 }
46 }
47}
48
49type disabledChecks struct {
50 // Key is the name of the unsupported API, value is the list of checks disabled because of it.
51 apis map[string][]string
52 mtx sync.Mutex
53}
54
55func (dc *disabledChecks) disable(api, check string) {
56 dc.mtx.Lock()
57 if _, ok := dc.apis[api]; !ok {
58 dc.apis[api] = []string{}
59 }
60 if !slices.Contains(dc.apis[api], check) {
61 dc.apis[api] = append(dc.apis[api], check)
62 }
63 dc.mtx.Unlock()
64}
65
66func (dc *disabledChecks) read() map[string][]string {
67 dc.mtx.Lock()
68 defer dc.mtx.Unlock()
69 return dc.apis
70}
71
72type FailoverGroup struct {
73 disabledChecks disabledChecks
74
75 name string
76 uri string
77 servers []*Prometheus
78 uptimeMetric string
79 cacheCollector *cacheCollector
80 quitChan chan bool
81
82 pathsInclude []*regexp.Regexp
83 pathsExclude []*regexp.Regexp
84 tags []string
85 started bool
86 strictErrors bool
87}
88
89func NewFailoverGroup(name, uri string, servers []*Prometheus, strictErrors bool, uptimeMetric string, include, exclude []*regexp.Regexp, tags []string) *FailoverGroup {
90 return &FailoverGroup{ // nolint: exhaustruct
91 name: name,
92 uri: uri,
93 servers: servers,
94 strictErrors: strictErrors,
95 uptimeMetric: uptimeMetric,
96 pathsInclude: include,
97 pathsExclude: exclude,
98 tags: tags,
99 disabledChecks: disabledChecks{apis: map[string][]string{}}, // nolint: exhaustruct
100 }
101}
102
103func (fg *FailoverGroup) Name() string {
104 return fg.name
105}
106
107func (fg *FailoverGroup) URI() string {
108 return fg.uri
109}
110
111func (fg *FailoverGroup) DisableCheck(api, s string) {
112 fg.disabledChecks.disable(api, s)
113}
114
115func (fg *FailoverGroup) GetDisabledChecks() map[string][]string {
116 return fg.disabledChecks.read()
117}
118
119func (fg *FailoverGroup) Include() []string {
120 sl := make([]string, 0, len(fg.pathsInclude))
121 for _, re := range fg.pathsInclude {
122 sl = append(sl, re.String())
123 }
124 return sl
125}
126
127func (fg *FailoverGroup) Exclude() []string {
128 sl := make([]string, 0, len(fg.pathsExclude))
129 for _, re := range fg.pathsExclude {
130 sl = append(sl, re.String())
131 }
132 return sl
133}
134
135func (fg *FailoverGroup) Tags() []string {
136 return fg.tags
137}
138
139func (fg *FailoverGroup) UptimeMetric() string {
140 return fg.uptimeMetric
141}
142
143func (fg *FailoverGroup) ServerCount() int {
144 return len(fg.servers)
145}
146
147func (fg *FailoverGroup) MergeUpstreams(src *FailoverGroup) {
148 for _, ns := range src.servers {
149 var present bool
150 for _, ol := range fg.servers {
151 if ol.unsafeURI == ns.unsafeURI {
152 present = true
153 break
154 }
155 }
156 if !present {
157 fg.servers = append(fg.servers, ns)
158 slog.LogAttrs(
159 context.Background(), slog.LevelDebug,
160 "Added new failover URI",
161 slog.String("name", ns.name),
162 slog.String("uri", ns.safeURI),
163 )
164 }
165 }
166}
167
168func (fg *FailoverGroup) IsEnabledForPath(path string) bool {
169 if len(fg.pathsInclude) == 0 && len(fg.pathsExclude) == 0 {
170 return true
171 }
172 for _, re := range fg.pathsExclude {
173 if re.MatchString(path) {
174 return false
175 }
176 }
177 for _, re := range fg.pathsInclude {
178 if re.MatchString(path) {
179 return true
180 }
181 }
182 return false
183}
184
185func (fg *FailoverGroup) StartWorkers(reg *prometheus.Registry) {
186 if fg.started {
187 return
188 }
189
190 queryCache := newQueryCache(time.Hour, time.Now)
191 fg.quitChan = make(chan bool)
192 go cacheCleaner(queryCache, time.Minute*2, fg.quitChan)
193
194 fg.cacheCollector = newCacheCollector(queryCache, fg.name)
195 reg.MustRegister(fg.cacheCollector)
196 for _, prom := range fg.servers {
197 prom.cache = queryCache
198 prom.StartWorkers()
199 }
200 fg.started = true
201}
202
203func (fg *FailoverGroup) Close(reg *prometheus.Registry) {
204 if !fg.started {
205 return
206 }
207 for _, prom := range fg.servers {
208 prom.Close()
209 }
210 reg.Unregister(fg.cacheCollector)
211 fg.quitChan <- true
212}
213
214func (fg *FailoverGroup) CleanCache() {
215 for _, prom := range fg.servers {
216 if prom.cache != nil {
217 prom.cache.gc()
218 return
219 }
220 }
221}
222
223func (fg *FailoverGroup) Config(ctx context.Context, cacheTTL time.Duration) (cfg *ConfigResult, err error) {
224 var uri string
225 for _, prom := range fg.servers {
226 uri = prom.safeURI
227 cfg, err = prom.Config(ctx, cacheTTL)
228 if err == nil {
229 return cfg, nil
230 }
231 if !IsUnavailableError(err) && !errors.Is(err, ErrUnsupported) {
232 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
233 }
234 }
235 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
236}
237
238func (fg *FailoverGroup) Query(ctx context.Context, expr string) (qr *QueryResult, err error) {
239 var uri string
240 for try, prom := range fg.servers {
241 if try > 0 {
242 slog.LogAttrs(
243 ctx, slog.LevelDebug,
244 "Using failover URI",
245 slog.String("name", fg.name),
246 slog.Int("retry", try),
247 slog.String("uri", prom.safeURI),
248 )
249 }
250 uri = prom.safeURI
251 qr, err = prom.Query(ctx, expr)
252 if err == nil {
253 return qr, nil
254 }
255 if !IsUnavailableError(err) {
256 return qr, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
257 }
258 }
259 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
260}
261
262func (fg *FailoverGroup) RangeQuery(ctx context.Context, expr string, params RangeQueryTimes) (rqr *RangeQueryResult, err error) {
263 var uri string
264 for _, prom := range fg.servers {
265 uri = prom.safeURI
266 rqr, err = prom.RangeQuery(ctx, expr, params)
267 if err == nil {
268 return rqr, nil
269 }
270 if !IsUnavailableError(err) {
271 return rqr, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
272 }
273 }
274 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
275}
276
277func (fg *FailoverGroup) Metadata(ctx context.Context, metric string) (metadata *MetadataResult, err error) {
278 var uri string
279 for _, prom := range fg.servers {
280 uri = prom.safeURI
281 metadata, err = prom.Metadata(ctx, metric)
282 if err == nil {
283 return metadata, nil
284 }
285 if !IsUnavailableError(err) && !errors.Is(err, ErrUnsupported) {
286 return metadata, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
287 }
288 }
289 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
290}
291
292func (fg *FailoverGroup) Flags(ctx context.Context) (flags *FlagsResult, err error) {
293 var uri string
294 for _, prom := range fg.servers {
295 uri = prom.safeURI
296 flags, err = prom.Flags(ctx)
297 if err == nil {
298 return flags, nil
299 }
300 if !IsUnavailableError(err) && !errors.Is(err, ErrUnsupported) {
301 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
302 }
303 }
304 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
305}
306
307func (fg *FailoverGroup) BuildInfo(ctx context.Context) (bi *BuildInfoResult, err error) {
308 var uri string
309 for _, prom := range fg.servers {
310 uri = prom.safeURI
311 bi, err = prom.BuildInfo(ctx)
312 if err == nil {
313 return bi, nil
314 }
315 if !IsUnavailableError(err) && !errors.Is(err, ErrUnsupported) {
316 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
317 }
318 }
319 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
320}
321