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

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/metadata.go

135lines · 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)
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, int) {
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, 1
46 }
47 defer resp.Body.Close()
48
49 if resp.StatusCode/100 != 2 {
50 qr.err = tryDecodingAPIError(resp)
51 return qr, 1
52 }
53
54 meta, err := streamMetadata(resp.Body)
55 qr.value, qr.err = meta, err
56 var cost int
57 for _, vals := range meta {
58 cost += len(vals)
59 }
60 if cost > 0 {
61 return qr, cost
62 }
63 return qr, 1
64}
65
66func (q metadataQuery) Endpoint() string {
67 return "/api/v1/metadata"
68}
69
70func (q metadataQuery) String() string {
71 return q.metric
72}
73
74func (q metadataQuery) CacheKey() uint64 {
75 return hash(q.prom.unsafeURI, q.Endpoint(), q.metric)
76}
77
78func (q metadataQuery) CacheTTL() time.Duration {
79 return time.Minute * 10
80}
81
82func (p *Prometheus) Metadata(ctx context.Context, metric string) (*MetadataResult, error) {
83 log.Debug().Str("uri", p.safeURI).Str("metric", metric).Msg("Scheduling Prometheus metrics metadata query")
84
85 key := fmt.Sprintf("/api/v1/metadata/%s", metric)
86 p.locker.lock(key)
87 defer p.locker.unlock(key)
88
89 resultChan := make(chan queryResult)
90 p.queries <- queryRequest{
91 query: metadataQuery{prom: p, ctx: ctx, metric: metric, timestamp: time.Now()},
92 result: resultChan,
93 }
94
95 result := <-resultChan
96 if result.err != nil {
97 return nil, QueryError{err: result.err, msg: decodeError(result.err)}
98 }
99
100 metadata := MetadataResult{URI: p.safeURI, Metadata: result.value.(map[string][]v1.Metadata)[metric]}
101
102 return &metadata, nil
103}
104
105func streamMetadata(r io.Reader) (meta map[string][]v1.Metadata, err error) {
106 defer dummyReadAll(r)
107
108 var status, errType, errText string
109 meta = map[string][]v1.Metadata{}
110 decoder := current.Object(
111 current.Key("status", current.Value(func(s string, isNil bool) {
112 status = s
113 })),
114 current.Key("error", current.Value(func(s string, isNil bool) {
115 errText = s
116 })),
117 current.Key("errorType", current.Value(func(s string, isNil bool) {
118 errType = s
119 })),
120 current.Key("data", current.Map(func(k string, v []v1.Metadata) {
121 meta[k] = v
122 })),
123 )
124
125 dec := json.NewDecoder(r)
126 if err = decoder.Stream(dec); err != nil {
127 return nil, APIError{Status: status, ErrorType: v1.ErrBadResponse, Err: fmt.Sprintf("JSON parse error: %s", err)}
128 }
129
130 if status != "success" {
131 return nil, APIError{Status: status, ErrorType: decodeErrorType(errType), Err: errText}
132 }
133
134 return meta, nil
135}