13 | 13 |
"fmt"
|
14 | 14 |
"io"
|
15 | 15 |
"strings"
|
|
16 |
"sync"
|
|
17 |
"sync/atomic"
|
16 | 18 |
"time"
|
17 | 19 |
|
18 | 20 |
"github.com/go-kit/kit/log"
|
19 | 21 |
"github.com/go-kit/kit/metrics"
|
|
22 |
"github.com/go-kit/kit/metrics/generic"
|
20 | 23 |
"github.com/go-kit/kit/metrics/internal/lv"
|
21 | 24 |
"github.com/go-kit/kit/metrics/internal/ratemap"
|
22 | 25 |
"github.com/go-kit/kit/util/conn"
|
|
33 | 36 |
// To regularly report metrics to an io.Writer, use the WriteLoop helper method.
|
34 | 37 |
// To send to a DogStatsD server, use the SendLoop helper method.
|
35 | 38 |
type Dogstatsd struct {
|
|
39 |
mtx sync.RWMutex
|
36 | 40 |
prefix string
|
37 | 41 |
rates *ratemap.RateMap
|
38 | 42 |
counters *lv.Space
|
39 | |
gauges *lv.Space
|
|
43 |
gauges map[string]*gaugeNode
|
40 | 44 |
timings *lv.Space
|
41 | 45 |
histograms *lv.Space
|
42 | 46 |
logger log.Logger
|
|
47 |
lvs lv.LabelValues
|
43 | 48 |
}
|
44 | 49 |
|
45 | 50 |
// New returns a Dogstatsd object that may be used to create metrics. Prefix is
|
46 | 51 |
// applied to all created metrics. Callers must ensure that regular calls to
|
47 | 52 |
// WriteTo are performed, either manually or with one of the helper methods.
|
48 | |
func New(prefix string, logger log.Logger) *Dogstatsd {
|
|
53 |
func New(prefix string, logger log.Logger, lvs ...string) *Dogstatsd {
|
|
54 |
if len(lvs)%2 != 0 {
|
|
55 |
panic("odd number of LabelValues; programmer error!")
|
|
56 |
}
|
49 | 57 |
return &Dogstatsd{
|
50 | 58 |
prefix: prefix,
|
51 | 59 |
rates: ratemap.New(),
|
52 | 60 |
counters: lv.NewSpace(),
|
53 | |
gauges: lv.NewSpace(),
|
|
61 |
gauges: map[string]*gaugeNode{},
|
54 | 62 |
timings: lv.NewSpace(),
|
55 | 63 |
histograms: lv.NewSpace(),
|
56 | 64 |
logger: logger,
|
|
65 |
lvs: lvs,
|
57 | 66 |
}
|
58 | 67 |
}
|
59 | 68 |
|
60 | 69 |
// NewCounter returns a counter, sending observations to this Dogstatsd object.
|
61 | 70 |
func (d *Dogstatsd) NewCounter(name string, sampleRate float64) *Counter {
|
62 | |
d.rates.Set(d.prefix+name, sampleRate)
|
|
71 |
d.rates.Set(name, sampleRate)
|
63 | 72 |
return &Counter{
|
64 | |
name: d.prefix + name,
|
|
73 |
name: name,
|
65 | 74 |
obs: d.counters.Observe,
|
66 | 75 |
}
|
67 | 76 |
}
|
68 | 77 |
|
69 | 78 |
// NewGauge returns a gauge, sending observations to this Dogstatsd object.
|
70 | 79 |
func (d *Dogstatsd) NewGauge(name string) *Gauge {
|
71 | |
return &Gauge{
|
72 | |
name: d.prefix + name,
|
73 | |
obs: d.gauges.Observe,
|
74 | |
add: d.gauges.Add,
|
75 | |
}
|
|
80 |
d.mtx.Lock()
|
|
81 |
n, ok := d.gauges[name]
|
|
82 |
if !ok {
|
|
83 |
n = &gaugeNode{gauge: &Gauge{g: generic.NewGauge(name), ddog: d}}
|
|
84 |
d.gauges[name] = n
|
|
85 |
}
|
|
86 |
d.mtx.Unlock()
|
|
87 |
return n.gauge
|
76 | 88 |
}
|
77 | 89 |
|
78 | 90 |
// NewTiming returns a histogram whose observations are interpreted as
|
79 | 91 |
// millisecond durations, and are forwarded to this Dogstatsd object.
|
80 | 92 |
func (d *Dogstatsd) NewTiming(name string, sampleRate float64) *Timing {
|
81 | |
d.rates.Set(d.prefix+name, sampleRate)
|
|
93 |
d.rates.Set(name, sampleRate)
|
82 | 94 |
return &Timing{
|
83 | |
name: d.prefix + name,
|
|
95 |
name: name,
|
84 | 96 |
obs: d.timings.Observe,
|
85 | 97 |
}
|
86 | 98 |
}
|
|
88 | 100 |
// NewHistogram returns a histogram whose observations are of an unspecified
|
89 | 101 |
// unit, and are forwarded to this Dogstatsd object.
|
90 | 102 |
func (d *Dogstatsd) NewHistogram(name string, sampleRate float64) *Histogram {
|
91 | |
d.rates.Set(d.prefix+name, sampleRate)
|
|
103 |
d.rates.Set(name, sampleRate)
|
92 | 104 |
return &Histogram{
|
93 | |
name: d.prefix + name,
|
|
105 |
name: name,
|
94 | 106 |
obs: d.histograms.Observe,
|
95 | 107 |
}
|
96 | 108 |
}
|
|
124 | 136 |
var n int
|
125 | 137 |
|
126 | 138 |
d.counters.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
|
127 | |
n, err = fmt.Fprintf(w, "%s:%f|c%s%s\n", name, sum(values), sampling(d.rates.Get(name)), tagValues(lvs))
|
|
139 |
n, err = fmt.Fprintf(w, "%s%s:%f|c%s%s\n", d.prefix, name, sum(values), sampling(d.rates.Get(name)), d.tagValues(lvs))
|
128 | 140 |
if err != nil {
|
129 | 141 |
return false
|
130 | 142 |
}
|
|
135 | 147 |
return count, err
|
136 | 148 |
}
|
137 | 149 |
|
138 | |
d.gauges.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
|
139 | |
n, err = fmt.Fprintf(w, "%s:%f|g%s\n", name, last(values), tagValues(lvs))
|
140 | |
if err != nil {
|
141 | |
return false
|
142 | |
}
|
143 | |
count += int64(n)
|
|
150 |
d.mtx.RLock()
|
|
151 |
for _, root := range d.gauges {
|
|
152 |
root.walk(func(name string, lvs lv.LabelValues, value float64) bool {
|
|
153 |
n, err = fmt.Fprintf(w, "%s%s:%f|g%s\n", d.prefix, name, value, d.tagValues(lvs))
|
|
154 |
if err != nil {
|
|
155 |
return false
|
|
156 |
}
|
|
157 |
count += int64(n)
|
|
158 |
return true
|
|
159 |
})
|
|
160 |
}
|
|
161 |
d.mtx.RUnlock()
|
|
162 |
|
|
163 |
d.timings.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
|
|
164 |
sampleRate := d.rates.Get(name)
|
|
165 |
for _, value := range values {
|
|
166 |
n, err = fmt.Fprintf(w, "%s%s:%f|ms%s%s\n", d.prefix, name, value, sampling(sampleRate), d.tagValues(lvs))
|
|
167 |
if err != nil {
|
|
168 |
return false
|
|
169 |
}
|
|
170 |
count += int64(n)
|
|
171 |
}
|
144 | 172 |
return true
|
145 | 173 |
})
|
146 | 174 |
if err != nil {
|
147 | 175 |
return count, err
|
148 | 176 |
}
|
149 | 177 |
|
150 | |
d.timings.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
|
|
178 |
d.histograms.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
|
151 | 179 |
sampleRate := d.rates.Get(name)
|
152 | 180 |
for _, value := range values {
|
153 | |
n, err = fmt.Fprintf(w, "%s:%f|ms%s%s\n", name, value, sampling(sampleRate), tagValues(lvs))
|
|
181 |
n, err = fmt.Fprintf(w, "%s%s:%f|h%s%s\n", d.prefix, name, value, sampling(sampleRate), d.tagValues(lvs))
|
154 | 182 |
if err != nil {
|
155 | 183 |
return false
|
156 | 184 |
}
|
|
162 | 190 |
return count, err
|
163 | 191 |
}
|
164 | 192 |
|
165 | |
d.histograms.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
|
166 | |
sampleRate := d.rates.Get(name)
|
167 | |
for _, value := range values {
|
168 | |
n, err = fmt.Fprintf(w, "%s:%f|h%s%s\n", name, value, sampling(sampleRate), tagValues(lvs))
|
169 | |
if err != nil {
|
170 | |
return false
|
171 | |
}
|
172 | |
count += int64(n)
|
173 | |
}
|
174 | |
return true
|
175 | |
})
|
176 | |
if err != nil {
|
177 | |
return count, err
|
178 | |
}
|
179 | |
|
180 | 193 |
return count, err
|
181 | 194 |
}
|
182 | 195 |
|
|
200 | 213 |
return sv
|
201 | 214 |
}
|
202 | 215 |
|
203 | |
func tagValues(labelValues []string) string {
|
|
216 |
func (d *Dogstatsd) tagValues(labelValues []string) string {
|
204 | 217 |
if len(labelValues) == 0 {
|
205 | 218 |
return ""
|
206 | 219 |
}
|
207 | 220 |
if len(labelValues)%2 != 0 {
|
208 | 221 |
panic("tagValues received a labelValues with an odd number of strings")
|
209 | 222 |
}
|
210 | |
pairs := make([]string, 0, len(labelValues)/2)
|
|
223 |
pairs := make([]string, 0, (len(d.lvs)+len(labelValues))/2)
|
|
224 |
for i := 0; i < len(d.lvs); i += 2 {
|
|
225 |
pairs = append(pairs, d.lvs[i]+":"+d.lvs[i+1])
|
|
226 |
}
|
211 | 227 |
for i := 0; i < len(labelValues); i += 2 {
|
212 | 228 |
pairs = append(pairs, labelValues[i]+":"+labelValues[i+1])
|
213 | 229 |
}
|
|
241 | 257 |
// Gauge is a DogStatsD gauge. Observations are forwarded to a Dogstatsd
|
242 | 258 |
// object, and aggregated (the last observation selected) per timeseries.
|
243 | 259 |
type Gauge struct {
|
244 | |
name string
|
245 | |
lvs lv.LabelValues
|
246 | |
obs observeFunc
|
247 | |
add observeFunc
|
|
260 |
g *generic.Gauge
|
|
261 |
ddog *Dogstatsd
|
|
262 |
set int32
|
248 | 263 |
}
|
249 | 264 |
|
250 | 265 |
// With implements metrics.Gauge.
|
251 | 266 |
func (g *Gauge) With(labelValues ...string) metrics.Gauge {
|
252 | |
return &Gauge{
|
253 | |
name: g.name,
|
254 | |
lvs: g.lvs.With(labelValues...),
|
255 | |
obs: g.obs,
|
256 | |
add: g.add,
|
257 | |
}
|
|
267 |
g.ddog.mtx.RLock()
|
|
268 |
node := g.ddog.gauges[g.g.Name]
|
|
269 |
g.ddog.mtx.RUnlock()
|
|
270 |
|
|
271 |
ga := &Gauge{g: g.g.With(labelValues...).(*generic.Gauge), ddog: g.ddog}
|
|
272 |
return node.addGauge(ga, ga.g.LabelValues())
|
258 | 273 |
}
|
259 | 274 |
|
260 | 275 |
// Set implements metrics.Gauge.
|
261 | 276 |
func (g *Gauge) Set(value float64) {
|
262 | |
g.obs(g.name, g.lvs, value)
|
|
277 |
g.g.Set(value)
|
|
278 |
g.touch()
|
263 | 279 |
}
|
264 | 280 |
|
265 | 281 |
// Add implements metrics.Gauge.
|
266 | 282 |
func (g *Gauge) Add(delta float64) {
|
267 | |
g.add(g.name, g.lvs, delta)
|
|
283 |
g.g.Add(delta)
|
|
284 |
g.touch()
|
268 | 285 |
}
|
269 | 286 |
|
270 | 287 |
// Timing is a DogStatsD timing, or metrics.Histogram. Observations are
|
|
311 | 328 |
func (h *Histogram) Observe(value float64) {
|
312 | 329 |
h.obs(h.name, h.lvs, value)
|
313 | 330 |
}
|
|
331 |
|
|
332 |
type pair struct{ label, value string }
|
|
333 |
|
|
334 |
type gaugeNode struct {
|
|
335 |
mtx sync.RWMutex
|
|
336 |
gauge *Gauge
|
|
337 |
children map[pair]*gaugeNode
|
|
338 |
}
|
|
339 |
|
|
340 |
func (n *gaugeNode) addGauge(g *Gauge, lvs lv.LabelValues) *Gauge {
|
|
341 |
n.mtx.Lock()
|
|
342 |
defer n.mtx.Unlock()
|
|
343 |
if len(lvs) == 0 {
|
|
344 |
if n.gauge == nil {
|
|
345 |
n.gauge = g
|
|
346 |
}
|
|
347 |
return n.gauge
|
|
348 |
}
|
|
349 |
if len(lvs) < 2 {
|
|
350 |
panic("too few LabelValues; programmer error!")
|
|
351 |
}
|
|
352 |
head, tail := pair{lvs[0], lvs[1]}, lvs[2:]
|
|
353 |
if n.children == nil {
|
|
354 |
n.children = map[pair]*gaugeNode{}
|
|
355 |
}
|
|
356 |
child, ok := n.children[head]
|
|
357 |
if !ok {
|
|
358 |
child = &gaugeNode{}
|
|
359 |
n.children[head] = child
|
|
360 |
}
|
|
361 |
return child.addGauge(g, tail)
|
|
362 |
}
|
|
363 |
|
|
364 |
func (n *gaugeNode) walk(fn func(string, lv.LabelValues, float64) bool) bool {
|
|
365 |
n.mtx.RLock()
|
|
366 |
defer n.mtx.RUnlock()
|
|
367 |
if n.gauge != nil {
|
|
368 |
value, ok := n.gauge.read()
|
|
369 |
if ok && !fn(n.gauge.g.Name, n.gauge.g.LabelValues(), value) {
|
|
370 |
return false
|
|
371 |
}
|
|
372 |
}
|
|
373 |
for _, child := range n.children {
|
|
374 |
if !child.walk(fn) {
|
|
375 |
return false
|
|
376 |
}
|
|
377 |
}
|
|
378 |
return true
|
|
379 |
}
|
|
380 |
|
|
381 |
func (g *Gauge) touch() {
|
|
382 |
atomic.StoreInt32(&(g.set), 1)
|
|
383 |
}
|
|
384 |
|
|
385 |
func (g *Gauge) read() (float64, bool) {
|
|
386 |
set := atomic.SwapInt32(&(g.set), 0)
|
|
387 |
return g.g.Value(), set != 0
|
|
388 |
}
|