cloudflare/pint

Public

mirrored from https://github.com/cloudflare/pintAvailable

CodeCommitsIssuesPull requestsActionsInsightsSecurity
v0.42.1

Branches

Tags

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

Clone

HTTPS

Download ZIP

internal/promapi/metadata.go

128lines · modecode

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)
16
17type MetadataResult struct {
18 URI string
19 Metadata []v1.Metadata
20}
21
22type metadataQuery struct {
23 prom *Prometheus
24 ctx context.Context
25 metric string
26 timestamp time.Time
27}
28
29func (q metadataQuery) Run() queryResult {
30 log.Debug().
31 Str("uri", q.prom.safeURI).
32 Str("metric", q.metric).
33 Msg("Getting prometheus metrics metadata")
34
35 ctx, cancel := context.WithTimeout(q.ctx, q.prom.timeout)
36 defer cancel()
37
38 var qr queryResult
39
40 args := url.Values{}
41 args.Set("metric", q.metric)
42 resp, err := q.prom.doRequest(ctx, http.MethodGet, q.Endpoint(), args)
43 if err != nil {
44 qr.err = fmt.Errorf("failed to query Prometheus metrics metadata: %w", err)
45 return qr
46 }
47 defer resp.Body.Close()
48
49 if resp.StatusCode/100 != 2 {
50 qr.err = tryDecodingAPIError(resp)
51 return qr
52 }
53
54 meta, err := streamMetadata(resp.Body)
55 qr.value, qr.err = meta, err
56 return qr
57}
58
59func (q metadataQuery) Endpoint() string {
60 return "/api/v1/metadata"
61}
62
63func (q metadataQuery) String() string {
64 return q.metric
65}
66
67func (q metadataQuery) CacheKey() uint64 {
68 return hash(q.prom.unsafeURI, q.Endpoint(), q.metric)
69}
70
71func (q metadataQuery) CacheTTL() time.Duration {
72 return time.Minute * 10
73}
74
75func (p *Prometheus) Metadata(ctx context.Context, metric string) (*MetadataResult, error) {
76 log.Debug().Str("uri", p.safeURI).Str("metric", metric).Msg("Scheduling Prometheus metrics metadata query")
77
78 key := fmt.Sprintf("/api/v1/metadata/%s", metric)
79 p.locker.lock(key)
80 defer p.locker.unlock(key)
81
82 resultChan := make(chan queryResult)
83 p.queries <- queryRequest{
84 query: metadataQuery{prom: p, ctx: ctx, metric: metric, timestamp: time.Now()},
85 result: resultChan,
86 }
87
88 result := <-resultChan
89 if result.err != nil {
90 return nil, QueryError{err: result.err, msg: decodeError(result.err)}
91 }
92
93 metadata := MetadataResult{URI: p.safeURI, Metadata: result.value.(map[string][]v1.Metadata)[metric]}
94
95 return &metadata, nil
96}
97
98func streamMetadata(r io.Reader) (meta map[string][]v1.Metadata, err error) {
99 defer dummyReadAll(r)
100
101 var status, errType, errText string
102 meta = map[string][]v1.Metadata{}
103 decoder := current.Object(
104 current.Key("status", current.Value(func(s string, isNil bool) {
105 status = s
106 })),
107 current.Key("error", current.Value(func(s string, isNil bool) {
108 errText = s
109 })),
110 current.Key("errorType", current.Value(func(s string, isNil bool) {
111 errType = s
112 })),
113 current.Key("data", current.Map(func(k string, v []v1.Metadata) {
114 meta[k] = v
115 })),
116 )
117
118 dec := json.NewDecoder(r)
119 if err = decoder.Stream(dec); err != nil {
120 return nil, APIError{Status: status, ErrorType: v1.ErrBadResponse, Err: fmt.Sprintf("JSON parse error: %s", err)}
121 }
122
123 if status != "success" {
124 return nil, APIError{Status: status, ErrorType: decodeErrorType(errType), Err: errText}
125 }
126
127 return meta, nil
128}
129