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

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/failover.go

266lines · modecode

RawDownload
Latest commit unavailable.
1package promapi
2
3import (
4 "context"
5 "log/slog"
6 "regexp"
7 "time"
8
9 "github.com/prometheus/client_golang/prometheus"
10)
11
12type FailoverGroupError struct {
13 err error
14 uri string
15 isStrict bool
16}
17
18func (e *FailoverGroupError) Unwrap() error {
19 return e.err
20}
21
22func (e *FailoverGroupError) Error() string {
23 return e.err.Error()
24}
25
26func (e *FailoverGroupError) URI() string {
27 return e.uri
28}
29
30func (e *FailoverGroupError) IsStrict() bool {
31 return e.isStrict
32}
33
34func cacheCleaner(cache *queryCache, interval time.Duration, quit chan bool) {
35 ticker := time.NewTicker(interval)
36 for {
37 select {
38 case <-quit:
39 return
40 case <-ticker.C:
41 cache.gc()
42 }
43 }
44}
45
46type FailoverGroup struct {
47 name string
48 uri string
49 servers []*Prometheus
50 uptimeMetric string
51 cacheCollector *cacheCollector
52 quitChan chan bool
53
54 pathsInclude []*regexp.Regexp
55 pathsExclude []*regexp.Regexp
56 tags []string
57 started bool
58 strictErrors bool
59}
60
61func NewFailoverGroup(name, uri string, servers []*Prometheus, strictErrors bool, uptimeMetric string, include, exclude []*regexp.Regexp, tags []string) *FailoverGroup {
62 return &FailoverGroup{
63 name: name,
64 uri: uri,
65 servers: servers,
66 strictErrors: strictErrors,
67 uptimeMetric: uptimeMetric,
68 pathsInclude: include,
69 pathsExclude: exclude,
70 tags: tags,
71 }
72}
73
74func (fg *FailoverGroup) Name() string {
75 return fg.name
76}
77
78func (fg *FailoverGroup) URI() string {
79 return fg.uri
80}
81
82func (fg *FailoverGroup) Include() []string {
83 sl := []string{}
84 for _, re := range fg.pathsInclude {
85 sl = append(sl, re.String())
86 }
87 return sl
88}
89
90func (fg *FailoverGroup) Exclude() []string {
91 sl := []string{}
92 for _, re := range fg.pathsExclude {
93 sl = append(sl, re.String())
94 }
95 return sl
96}
97
98func (fg *FailoverGroup) Tags() []string {
99 return fg.tags
100}
101
102func (fg *FailoverGroup) UptimeMetric() string {
103 return fg.uptimeMetric
104}
105
106func (fg *FailoverGroup) ServerCount() int {
107 return len(fg.servers)
108}
109
110func (fg *FailoverGroup) MergeUpstreams(src *FailoverGroup) {
111 for _, ns := range src.servers {
112 var present bool
113 for _, ol := range fg.servers {
114 if ol.unsafeURI == ns.unsafeURI {
115 present = true
116 break
117 }
118 }
119 if !present {
120 fg.servers = append(fg.servers, ns)
121 slog.Debug(
122 "Added new failover URI",
123 slog.String("name", ns.name),
124 slog.String("uri", ns.safeURI),
125 )
126 }
127 }
128}
129
130func (fg *FailoverGroup) IsEnabledForPath(path string) bool {
131 if len(fg.pathsInclude) == 0 && len(fg.pathsExclude) == 0 {
132 return true
133 }
134 for _, re := range fg.pathsExclude {
135 if re.MatchString(path) {
136 return false
137 }
138 }
139 for _, re := range fg.pathsInclude {
140 if re.MatchString(path) {
141 return true
142 }
143 }
144 return false
145}
146
147func (fg *FailoverGroup) StartWorkers(reg *prometheus.Registry) {
148 if fg.started {
149 return
150 }
151
152 queryCache := newQueryCache(time.Hour)
153 fg.quitChan = make(chan bool)
154 go cacheCleaner(queryCache, time.Minute*2, fg.quitChan)
155
156 fg.cacheCollector = newCacheCollector(queryCache, fg.name)
157 reg.MustRegister(fg.cacheCollector)
158 for _, prom := range fg.servers {
159 prom.cache = queryCache
160 prom.StartWorkers()
161 }
162 fg.started = true
163}
164
165func (fg *FailoverGroup) Close(reg *prometheus.Registry) {
166 if !fg.started {
167 return
168 }
169 for _, prom := range fg.servers {
170 prom.Close()
171 }
172 reg.Unregister(fg.cacheCollector)
173 fg.quitChan <- true
174}
175
176func (fg *FailoverGroup) CleanCache() {
177 for _, prom := range fg.servers {
178 if prom.cache != nil {
179 prom.cache.gc()
180 return
181 }
182 }
183}
184
185func (fg *FailoverGroup) Config(ctx context.Context, cacheTTL time.Duration) (cfg *ConfigResult, err error) {
186 var uri string
187 for _, prom := range fg.servers {
188 uri = prom.safeURI
189 cfg, err = prom.Config(ctx, cacheTTL)
190 if err == nil {
191 return cfg, nil
192 }
193 if !IsUnavailableError(err) {
194 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
195 }
196 }
197 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
198}
199
200func (fg *FailoverGroup) Query(ctx context.Context, expr string) (qr *QueryResult, err error) {
201 var uri string
202 for try, prom := range fg.servers {
203 if try > 0 {
204 slog.Debug(
205 "Using failover URI",
206 slog.String("name", fg.name),
207 slog.Int("retry", try),
208 slog.String("uri", prom.safeURI),
209 )
210 }
211 uri = prom.safeURI
212 qr, err = prom.Query(ctx, expr)
213 if err == nil {
214 return qr, nil
215 }
216 if !IsUnavailableError(err) {
217 return qr, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
218 }
219 }
220 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
221}
222
223func (fg *FailoverGroup) RangeQuery(ctx context.Context, expr string, params RangeQueryTimes) (rqr *RangeQueryResult, err error) {
224 var uri string
225 for _, prom := range fg.servers {
226 uri = prom.safeURI
227 rqr, err = prom.RangeQuery(ctx, expr, params)
228 if err == nil {
229 return rqr, nil
230 }
231 if !IsUnavailableError(err) {
232 return rqr, &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) Metadata(ctx context.Context, metric string) (metadata *MetadataResult, err error) {
239 var uri string
240 for _, prom := range fg.servers {
241 uri = prom.safeURI
242 metadata, err = prom.Metadata(ctx, metric)
243 if err == nil {
244 return metadata, nil
245 }
246 if !IsUnavailableError(err) {
247 return metadata, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
248 }
249 }
250 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
251}
252
253func (fg *FailoverGroup) Flags(ctx context.Context) (flags *FlagsResult, err error) {
254 var uri string
255 for _, prom := range fg.servers {
256 uri = prom.safeURI
257 flags, err = prom.Flags(ctx)
258 if err == nil {
259 return flags, nil
260 }
261 if !IsUnavailableError(err) {
262 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
263 }
264 }
265 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
266}
267