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

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/config.go

153lines · modecode

RawDownload
Latest commit unavailable.
1package promapi
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "io"
8 "net/http"
9 "net/url"
10 "time"
11
12 v1 "github.com/prometheus/client_golang/api/prometheus/v1"
13 "github.com/prymitive/current"
14 "github.com/rs/zerolog/log"
15 "gopkg.in/yaml.v3"
16)
17
18type ConfigSectionGlobal struct {
19 ScrapeInterval time.Duration `yaml:"scrape_interval"`
20 ScrapeTimeout time.Duration `yaml:"scrape_timeout"`
21 EvaluationInterval time.Duration `yaml:"evaluation_interval"`
22 ExternalLabels map[string]string `yaml:"external_labels"`
23}
24
25type PrometheusConfig struct {
26 Global ConfigSectionGlobal `yaml:"global"`
27}
28
29type ConfigResult struct {
30 URI string
31 Config PrometheusConfig
32}
33
34type configQuery struct {
35 prom *Prometheus
36 ctx context.Context
37 timestamp time.Time
38}
39
40func (q configQuery) Run() queryResult {
41 log.Debug().
42 Str("uri", q.prom.safeURI).
43 Msg("Getting prometheus configuration")
44
45 ctx, cancel := context.WithTimeout(q.ctx, q.prom.timeout)
46 defer cancel()
47
48 qr := queryResult{expires: q.timestamp.Add(cacheExpiry * 2)}
49
50 args := url.Values{}
51 resp, err := q.prom.doRequest(ctx, http.MethodGet, q.Endpoint(), args)
52 if err != nil {
53 qr.err = fmt.Errorf("failed to query Prometheus config: %w", err)
54 return qr
55 }
56 defer resp.Body.Close()
57
58 if resp.StatusCode/100 != 2 {
59 qr.err = tryDecodingAPIError(resp)
60 return qr
61 }
62
63 qr.value, qr.err = streamConfig(resp.Body)
64 return qr
65}
66
67func (q configQuery) Endpoint() string {
68 return "/api/v1/status/config"
69}
70
71func (q configQuery) String() string {
72 return "/api/v1/status/config"
73}
74
75func (q configQuery) CacheAfter() int {
76 return 0
77}
78
79func (q configQuery) CacheKey() uint64 {
80 return hash(q.prom.unsafeURI, q.Endpoint(), q.timestamp.Round(cacheExpiry).Format(time.RFC3339))
81}
82
83func (p *Prometheus) Config(ctx context.Context) (*ConfigResult, error) {
84 log.Debug().Str("uri", p.safeURI).Msg("Scheduling Prometheus configuration query")
85
86 key := "/api/v1/status/config"
87 p.locker.lock(key)
88 defer p.locker.unlock(key)
89
90 resultChan := make(chan queryResult)
91 p.queries <- queryRequest{
92 query: configQuery{prom: p, ctx: ctx, timestamp: time.Now()},
93 result: resultChan,
94 }
95
96 result := <-resultChan
97 if result.err != nil {
98 return nil, QueryError{err: result.err, msg: decodeError(result.err)}
99 }
100
101 var cfg PrometheusConfig
102 if err := yaml.Unmarshal([]byte(result.value.(string)), &cfg); err != nil {
103 prometheusQueryErrorsTotal.WithLabelValues(p.name, "/api/v1/status/config", errReason(err)).Inc()
104 return nil, fmt.Errorf("failed to decode config data in %s response: %w", p.safeURI, err)
105 }
106
107 if cfg.Global.ScrapeInterval == 0 {
108 cfg.Global.ScrapeInterval = time.Minute
109 }
110 if cfg.Global.ScrapeTimeout == 0 {
111 cfg.Global.ScrapeTimeout = time.Second * 10
112 }
113 if cfg.Global.EvaluationInterval == 0 {
114 cfg.Global.EvaluationInterval = time.Minute
115 }
116
117 r := ConfigResult{URI: p.safeURI, Config: cfg}
118
119 return &r, nil
120}
121
122func streamConfig(r io.Reader) (cfg string, err error) {
123 defer dummyReadAll(r)
124
125 var status, errType, errText string
126 decoder := current.Object(
127 current.Key("status", current.Value(func(s string, isNil bool) {
128 status = s
129 })),
130 current.Key("error", current.Value(func(s string, isNil bool) {
131 errText = s
132 })),
133 current.Key("errorType", current.Value(func(s string, isNil bool) {
134 errType = s
135 })),
136 current.Key("data", current.Object(
137 current.Key("yaml", current.Value(func(s string, isNil bool) {
138 cfg = s
139 })),
140 )),
141 )
142
143 dec := json.NewDecoder(r)
144 if err = decoder.Stream(dec); err != nil {
145 return cfg, APIError{Status: status, ErrorType: v1.ErrBadResponse, Err: fmt.Sprintf("JSON parse error: %s", err)}
146 }
147
148 if status != "success" {
149 return cfg, APIError{Status: status, ErrorType: decodeErrorType(errType), Err: errText}
150 }
151
152 return cfg, nil
153}
154