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

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/prometheus.go

148lines · modecode

RawDownload
Latest commit unavailable.
1package promapi
2
3import (
4 "sync"
5 "time"
6
7 lru "github.com/hashicorp/golang-lru"
8 "github.com/prometheus/client_golang/api"
9 v1 "github.com/prometheus/client_golang/api/prometheus/v1"
10 "github.com/rs/zerolog/log"
11
12 "github.com/cloudflare/pint/internal/output"
13)
14
15type QueryError struct {
16 err error
17 msg string
18}
19
20func (qe QueryError) Error() string {
21 return qe.msg
22}
23
24func (qe QueryError) Unwrap() error {
25 return qe.err
26}
27
28type querier interface {
29 Endpoint() string
30 String() string
31 CacheKey() string
32 Run() (any, error)
33}
34
35type queryRequest struct {
36 query querier
37 result chan queryResult
38}
39
40type queryResult struct {
41 value any
42 err error
43}
44
45type Prometheus struct {
46 name string
47 uri string
48 api v1.API
49 timeout time.Duration
50 concurrency int
51 cache *lru.ARCCache
52
53 wg sync.WaitGroup
54 queries chan queryRequest
55}
56
57func NewPrometheus(name, uri string, timeout time.Duration, concurrency int) *Prometheus {
58 client, err := api.NewClient(api.Config{Address: uri})
59 if err != nil {
60 // config validation should prevent this from ever happening
61 // panic so we don't need to return an error and it's easier to
62 // use this code in tests
63 panic(err)
64 }
65 cache, _ := lru.NewARC(1000)
66
67 prom := Prometheus{
68 name: name,
69 uri: uri,
70 api: v1.NewAPI(client),
71 timeout: timeout,
72 cache: cache,
73 concurrency: concurrency,
74 }
75 return &prom
76}
77
78func (prom *Prometheus) Close() {
79 log.Debug().Str("name", prom.name).Str("uri", prom.uri).Msg("Stopping query workers")
80 close(prom.queries)
81 prom.wg.Wait()
82}
83
84func (prom *Prometheus) StartWorkers() {
85 log.Debug().
86 Str("name", prom.name).
87 Str("uri", prom.uri).
88 Int("workers", prom.concurrency).
89 Msg("Starting query workers")
90
91 prom.queries = make(chan queryRequest, prom.concurrency*10)
92
93 for w := 1; w <= prom.concurrency; w++ {
94 prom.wg.Add(1)
95 go func() {
96 defer prom.wg.Done()
97 queryWorker(prom, prom.queries)
98 }()
99 }
100}
101
102func queryWorker(prom *Prometheus, queries chan queryRequest) {
103 for job := range queries {
104 job := job
105
106 cacheKey := job.query.CacheKey()
107 if cacheKey != "" {
108 if cached, ok := prom.cache.Get(cacheKey); ok {
109 job.result <- queryResult{value: cached}
110 prometheusCacheHitsTotal.WithLabelValues(prom.name, job.query.Endpoint()).Inc()
111 log.Debug().
112 Str("uri", prom.uri).
113 Str("query", job.query.String()).
114 Msg("Cache hit")
115 continue
116 }
117 }
118
119 prometheusQueriesTotal.WithLabelValues(prom.name, job.query.Endpoint()).Inc()
120 prometheusQueriesRunning.WithLabelValues(prom.name, job.query.Endpoint()).Inc()
121 start := time.Now()
122 result, err := job.query.Run()
123 dur := time.Since(start)
124 log.Debug().
125 Str("uri", prom.uri).
126 Str("query", job.query.String()).
127 Str("endpoint", job.query.Endpoint()).
128 Str("duration", output.HumanizeDuration(dur)).
129 Msg("Query completed")
130 prometheusQueriesRunning.WithLabelValues(prom.name, job.query.Endpoint()).Dec()
131 if err != nil {
132 prometheusQueryErrorsTotal.WithLabelValues(prom.name, job.query.Endpoint(), errReason(err)).Inc()
133 log.Error().
134 Err(err).
135 Str("uri", prom.uri).
136 Str("query", job.query.String()).
137 Msg("Query returned an error")
138 job.result <- queryResult{err: err}
139 continue
140 }
141
142 if cacheKey != "" {
143 prom.cache.Add(cacheKey, result)
144 }
145
146 job.result <- queryResult{value: result}
147 }
148}
149