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

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/config.go

156lines · modecode

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