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

Branches

Tags

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

Clone

HTTPS

Download ZIP

pint/internal/promapi

internal/promapi/metadata.go

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