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

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/failover.go

198lines · modecode

RawDownload
Latest commit unavailable.
1package promapi
2
3import (
4 "context"
5 "regexp"
6 "time"
7
8 "github.com/prometheus/client_golang/prometheus"
9)
10
11type FailoverGroupError struct {
12 err error
13 uri string
14 isStrict bool
15}
16
17func (e *FailoverGroupError) Unwrap() error {
18 return e.err
19}
20
21func (e *FailoverGroupError) Error() string {
22 return e.err.Error()
23}
24
25func (e *FailoverGroupError) URI() string {
26 return e.uri
27}
28
29func (e *FailoverGroupError) IsStrict() bool {
30 return e.isStrict
31}
32
33func cacheCleaner(cache *queryCache, interval time.Duration, quit chan bool) {
34 ticker := time.NewTicker(interval)
35 for {
36 select {
37 case <-quit:
38 return
39 case <-ticker.C:
40 cache.gc()
41 }
42 }
43}
44
45type FailoverGroup struct {
46 name string
47 servers []*Prometheus
48 cacheSize int
49 strictErrors bool
50 uptimeMetric string
51 cacheCollector *cacheCollector
52 quitChan chan bool
53
54 pathsInclude []*regexp.Regexp
55 pathsExclude []*regexp.Regexp
56}
57
58func NewFailoverGroup(name string, servers []*Prometheus, cacheSize int, strictErrors bool, uptimeMetric string, include, exclude []*regexp.Regexp) *FailoverGroup {
59 return &FailoverGroup{
60 name: name,
61 servers: servers,
62 cacheSize: cacheSize,
63 strictErrors: strictErrors,
64 uptimeMetric: uptimeMetric,
65 pathsInclude: include,
66 pathsExclude: exclude,
67 }
68}
69
70func (fg *FailoverGroup) Name() string {
71 return fg.name
72}
73
74func (fg *FailoverGroup) UptimeMetric() string {
75 return fg.uptimeMetric
76}
77
78func (fg *FailoverGroup) IsEnabledForPath(path string) bool {
79 if len(fg.pathsInclude) == 0 && len(fg.pathsExclude) == 0 {
80 return true
81 }
82 for _, re := range fg.pathsExclude {
83 if re.MatchString(path) {
84 return false
85 }
86 }
87 for _, re := range fg.pathsInclude {
88 if re.MatchString(path) {
89 return true
90 }
91 }
92 return false
93}
94
95func (fg *FailoverGroup) StartWorkers() {
96 queryCache := newQueryCache(fg.cacheSize)
97 fg.quitChan = make(chan bool)
98 go cacheCleaner(queryCache, time.Minute*2, fg.quitChan)
99
100 fg.cacheCollector = newCacheCollector(queryCache, fg.name)
101 prometheus.MustRegister(fg.cacheCollector)
102 for _, prom := range fg.servers {
103 prom.cache = queryCache
104 prom.StartWorkers()
105 }
106}
107
108func (fg *FailoverGroup) Close() {
109 for _, prom := range fg.servers {
110 prom.Close()
111 }
112 prometheus.Unregister(fg.cacheCollector)
113 fg.quitChan <- true
114}
115
116func (fg *FailoverGroup) CleanCache() {
117 for _, prom := range fg.servers {
118 if prom.cache != nil {
119 prom.cache.gc()
120 return
121 }
122 }
123}
124
125func (fg *FailoverGroup) Config(ctx context.Context) (cfg *ConfigResult, err error) {
126 var uri string
127 for _, prom := range fg.servers {
128 uri = prom.safeURI
129 cfg, err = prom.Config(ctx)
130 if err == nil {
131 return
132 }
133 if !IsUnavailableError(err) {
134 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
135 }
136 }
137 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
138}
139
140func (fg *FailoverGroup) Query(ctx context.Context, expr string) (qr *QueryResult, err error) {
141 var uri string
142 for _, prom := range fg.servers {
143 uri = prom.safeURI
144 qr, err = prom.Query(ctx, expr)
145 if err == nil {
146 return
147 }
148 if !IsUnavailableError(err) {
149 return qr, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
150 }
151 }
152 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
153}
154
155func (fg *FailoverGroup) RangeQuery(ctx context.Context, expr string, params RangeQueryTimes) (rqr *RangeQueryResult, err error) {
156 var uri string
157 for _, prom := range fg.servers {
158 uri = prom.safeURI
159 rqr, err = prom.RangeQuery(ctx, expr, params)
160 if err == nil {
161 return
162 }
163 if !IsUnavailableError(err) {
164 return rqr, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
165 }
166 }
167 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
168}
169
170func (fg *FailoverGroup) Metadata(ctx context.Context, metric string) (metadata *MetadataResult, err error) {
171 var uri string
172 for _, prom := range fg.servers {
173 uri = prom.safeURI
174 metadata, err = prom.Metadata(ctx, metric)
175 if err == nil {
176 return
177 }
178 if !IsUnavailableError(err) {
179 return metadata, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
180 }
181 }
182 return nil, &FailoverGroupError{err: err, uri: uri, isStrict: fg.strictErrors}
183}
184
185func (fg *FailoverGroup) Flags(ctx context.Context) (flags *FlagsResult, err error) {
186 var uri string
187 for _, prom := range fg.servers {
188 uri = prom.safeURI
189 flags, err = prom.Flags(ctx)
190 if err == nil {
191 return
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