forked from go-kit/kit
-
Notifications
You must be signed in to change notification settings - Fork 0
/
statsd.go
239 lines (209 loc) · 6.8 KB
/
statsd.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
// Package statsd provides a StatsD backend for package metrics. StatsD has no
// concept of arbitrary key-value tagging, so label values are not supported,
// and With is a no-op on all metrics.
//
// This package batches observations and emits them on some schedule to the
// remote server. This is useful even if you connect to your StatsD server over
// UDP. Emitting one network packet per observation can quickly overwhelm even
// the fastest internal network.
package statsd
import (
"fmt"
"io"
"time"
"github.com/go-kit/kit/log"
"github.com/go-kit/kit/metrics"
"github.com/go-kit/kit/metrics/internal/lv"
"github.com/go-kit/kit/metrics/internal/ratemap"
"github.com/go-kit/kit/util/conn"
)
// Statsd receives metrics observations and forwards them to a StatsD server.
// Create a Statsd object, use it to create metrics, and pass those metrics as
// dependencies to the components that will use them.
//
// All metrics are buffered until WriteTo is called. Counters and gauges are
// aggregated into a single observation per timeseries per write. Timings are
// buffered but not aggregated.
//
// To regularly report metrics to an io.Writer, use the WriteLoop helper method.
// To send to a StatsD server, use the SendLoop helper method.
type Statsd struct {
prefix string
rates *ratemap.RateMap
// The observations are collected in an N-dimensional vector space, even
// though they only take advantage of a single dimension (name). This is an
// implementation detail born purely from convenience. It would be more
// accurate to collect them in a map[string][]float64, but we already have
// this nice data structure and helper methods.
counters *lv.Space
gauges *lv.Space
timings *lv.Space
logger log.Logger
}
// New returns a Statsd object that may be used to create metrics. Prefix is
// applied to all created metrics. Callers must ensure that regular calls to
// WriteTo are performed, either manually or with one of the helper methods.
func New(prefix string, logger log.Logger) *Statsd {
return &Statsd{
prefix: prefix,
rates: ratemap.New(),
counters: lv.NewSpace(),
gauges: lv.NewSpace(),
timings: lv.NewSpace(),
logger: logger,
}
}
// NewCounter returns a counter, sending observations to this Statsd object.
func (s *Statsd) NewCounter(name string, sampleRate float64) *Counter {
s.rates.Set(s.prefix+name, sampleRate)
return &Counter{
name: s.prefix + name,
obs: s.counters.Observe,
}
}
// NewGauge returns a gauge, sending observations to this Statsd object.
func (s *Statsd) NewGauge(name string) *Gauge {
return &Gauge{
name: s.prefix + name,
obs: s.gauges.Observe,
add: s.gauges.Add,
}
}
// NewTiming returns a histogram whose observations are interpreted as
// millisecond durations, and are forwarded to this Statsd object.
func (s *Statsd) NewTiming(name string, sampleRate float64) *Timing {
s.rates.Set(s.prefix+name, sampleRate)
return &Timing{
name: s.prefix + name,
obs: s.timings.Observe,
}
}
// WriteLoop is a helper method that invokes WriteTo to the passed writer every
// time the passed channel fires. This method blocks until the channel is
// closed, so clients probably want to run it in its own goroutine. For typical
// usage, create a time.Ticker and pass its C channel to this method.
func (s *Statsd) WriteLoop(c <-chan time.Time, w io.Writer) {
for range c {
if _, err := s.WriteTo(w); err != nil {
s.logger.Log("during", "WriteTo", "err", err)
}
}
}
// SendLoop is a helper method that wraps WriteLoop, passing a managed
// connection to the network and address. Like WriteLoop, this method blocks
// until the channel is closed, so clients probably want to start it in its own
// goroutine. For typical usage, create a time.Ticker and pass its C channel to
// this method.
func (s *Statsd) SendLoop(c <-chan time.Time, network, address string) {
s.WriteLoop(c, conn.NewDefaultManager(network, address, s.logger))
}
// WriteTo flushes the buffered content of the metrics to the writer, in
// StatsD format. WriteTo abides best-effort semantics, so observations are
// lost if there is a problem with the write. Clients should be sure to call
// WriteTo regularly, ideally through the WriteLoop or SendLoop helper methods.
func (s *Statsd) WriteTo(w io.Writer) (count int64, err error) {
var n int
s.counters.Reset().Walk(func(name string, _ lv.LabelValues, values []float64) bool {
n, err = fmt.Fprintf(w, "%s:%f|c%s\n", name, sum(values), sampling(s.rates.Get(name)))
if err != nil {
return false
}
count += int64(n)
return true
})
if err != nil {
return count, err
}
s.gauges.Reset().Walk(func(name string, _ lv.LabelValues, values []float64) bool {
n, err = fmt.Fprintf(w, "%s:%f|g\n", name, last(values))
if err != nil {
return false
}
count += int64(n)
return true
})
if err != nil {
return count, err
}
s.timings.Reset().Walk(func(name string, _ lv.LabelValues, values []float64) bool {
sampleRate := s.rates.Get(name)
for _, value := range values {
n, err = fmt.Fprintf(w, "%s:%f|ms%s\n", name, value, sampling(sampleRate))
if err != nil {
return false
}
count += int64(n)
}
return true
})
if err != nil {
return count, err
}
return count, err
}
func sum(a []float64) float64 {
var v float64
for _, f := range a {
v += f
}
return v
}
func last(a []float64) float64 {
return a[len(a)-1]
}
func sampling(r float64) string {
var sv string
if r < 1.0 {
sv = fmt.Sprintf("|@%f", r)
}
return sv
}
type observeFunc func(name string, lvs lv.LabelValues, value float64)
// Counter is a StatsD counter. Observations are forwarded to a Statsd object,
// and aggregated (summed) per timeseries.
type Counter struct {
name string
obs observeFunc
}
// With is a no-op.
func (c *Counter) With(...string) metrics.Counter {
return c
}
// Add implements metrics.Counter.
func (c *Counter) Add(delta float64) {
c.obs(c.name, lv.LabelValues{}, delta)
}
// Gauge is a StatsD gauge. Observations are forwarded to a Statsd object, and
// aggregated (the last observation selected) per timeseries.
type Gauge struct {
name string
obs observeFunc
add observeFunc
}
// With is a no-op.
func (g *Gauge) With(...string) metrics.Gauge {
return g
}
// Set implements metrics.Gauge.
func (g *Gauge) Set(value float64) {
g.obs(g.name, lv.LabelValues{}, value)
}
// Add implements metrics.Gauge.
func (g *Gauge) Add(delta float64) {
g.add(g.name, lv.LabelValues{}, delta)
}
// Timing is a StatsD timing, or metrics.Histogram. Observations are
// forwarded to a Statsd object, and collected (but not aggregated) per
// timeseries.
type Timing struct {
name string
obs observeFunc
}
// With is a no-op.
func (t *Timing) With(...string) metrics.Histogram {
return t
}
// Observe implements metrics.Histogram. Value is interpreted as milliseconds.
func (t *Timing) Observe(value float64) {
t.obs(t.name, lv.LabelValues{}, value)
}