Codebase list golang-github-go-kit-kit / dda4d22
adding tests for statsd.Emitter JP Robinson 7 years ago
2 changed file(s) with 127 addition(s) and 6 deletion(s). Raw diff Collapse all Expand all
+11
-6
metrics/statsd/emitter.go less more
5353 keyVals: make(chan keyVal),
5454 quitc: make(chan chan struct{}),
5555 }
56 var b bytes.Buffer
57 go e.loop(flushInterval, &b)
56 go e.loop(flushInterval)
5857 return e
5958 }
6059
6362 // report interval or until the buffer exceeds a max packet size, whichever
6463 // comes first. Fields are ignored.
6564 func (e *Emitter) NewCounter(key string) metrics.Counter {
65 key = e.prefix + key
6666 return &statsdCounter{
67 key: e.prefix + key,
67 key: key,
6868 c: stringToKeyVal(key, e.keyVals),
6969 }
7070 }
8787 //
8888 // TODO: support for sampling.
8989 func (e *Emitter) NewHistogram(key string) metrics.Histogram {
90 key = e.prefix + key
9091 return &statsdHistogram{
91 key: e.prefix + key,
92 key: key,
9293 h: stringToKeyVal(key, e.keyVals),
9394 }
9495 }
100101 //
101102 // TODO: support for sampling
102103 func (e *Emitter) NewGauge(key string) metrics.Gauge {
104 key = e.prefix + key
103105 return &statsdGauge{
104 key: e.prefix + key,
106 key: key,
105107 g: stringToKeyVal(key, e.keyVals),
106108 }
107109 }
108110
109 func (e *Emitter) loop(d time.Duration, buf *bytes.Buffer) {
111 func (e *Emitter) loop(d time.Duration) {
110112 ticker := time.NewTicker(d)
111113 defer ticker.Stop()
114 buf := &bytes.Buffer{}
112115 for {
113116 select {
114117 case kv := <-e.keyVals:
149152 if err != nil {
150153 e.logger.Log("during", "flush", "err", err)
151154 }
155 buf.Reset()
156
152157 e.mgr.Put(err)
153158 }
+116
-0
metrics/statsd/statsd_test.go less more
22 import (
33 "bytes"
44 "fmt"
5 "net"
56 "strings"
67 "sync"
78 "testing"
89 "time"
10
11 "github.com/go-kit/kit/log"
12 "github.com/go-kit/kit/util/conn"
913 )
14
15 func TestEmitterCounter(t *testing.T) {
16 e, buf := testEmitter()
17
18 c := e.NewCounter("test_statsd_counter")
19 c.Add(1)
20 c.Add(2)
21
22 // give time for things to emit
23 time.Sleep(time.Millisecond * 250)
24 // force a flush and stop
25 e.Stop()
26
27 want := "prefix.test_statsd_counter:1|c\nprefix.test_statsd_counter:2|c\n"
28 have := buf.String()
29 if want != have {
30 t.Errorf("want %q, have %q", want, have)
31 }
32 }
33
34 func TestEmitterGauge(t *testing.T) {
35 e, buf := testEmitter()
36
37 g := e.NewGauge("test_statsd_gauge")
38
39 delta := 1.0
40 g.Add(delta)
41
42 // give time for things to emit
43 time.Sleep(time.Millisecond * 250)
44 // force a flush and stop
45 e.Stop()
46
47 want := fmt.Sprintf("prefix.test_statsd_gauge:+%f|g\n", delta)
48 have := buf.String()
49 if want != have {
50 t.Errorf("want %q, have %q", want, have)
51 }
52 }
53
54 func TestEmitterHistogram(t *testing.T) {
55 e, buf := testEmitter()
56 h := e.NewHistogram("test_statsd_histogram")
57
58 h.Observe(123)
59
60 // give time for things to emit
61 time.Sleep(time.Millisecond * 250)
62 // force a flush and stop
63 e.Stop()
64
65 want := "prefix.test_statsd_histogram:123|ms\n"
66 have := buf.String()
67 if want != have {
68 t.Errorf("want %q, have %q", want, have)
69 }
70 }
1071
1172 func TestCounter(t *testing.T) {
1273 buf := &syncbuf{buf: &bytes.Buffer{}}
140201 defer s.mtx.Unlock()
141202 s.buf.Reset()
142203 }
204
205 func testEmitter() (*Emitter, *syncbuf) {
206 buf := &syncbuf{buf: &bytes.Buffer{}}
207 e := &Emitter{
208 prefix: "prefix.",
209 mgr: conn.NewManager(mockDialer(buf), "", "", time.After, log.NewNopLogger()),
210 logger: log.NewNopLogger(),
211 keyVals: make(chan keyVal),
212 quitc: make(chan chan struct{}),
213 }
214 go e.loop(time.Millisecond * 20)
215 return e, buf
216 }
217
218 func mockDialer(buf *syncbuf) conn.Dialer {
219 return func(net, addr string) (net.Conn, error) {
220 return &mockConn{buf}, nil
221 }
222 }
223
224 type mockConn struct {
225 buf *syncbuf
226 }
227
228 func (c *mockConn) Read(b []byte) (n int, err error) {
229 panic("not implemented")
230 }
231
232 func (c *mockConn) Write(b []byte) (n int, err error) {
233 return c.buf.Write(b)
234 }
235
236 func (c *mockConn) Close() error {
237 panic("not implemented")
238 }
239
240 func (c *mockConn) LocalAddr() net.Addr {
241 panic("not implemented")
242 }
243
244 func (c *mockConn) RemoteAddr() net.Addr {
245 panic("not implemented")
246 }
247
248 func (c *mockConn) SetDeadline(t time.Time) error {
249 panic("not implemented")
250 }
251
252 func (c *mockConn) SetReadDeadline(t time.Time) error {
253 panic("not implemented")
254 }
255
256 func (c *mockConn) SetWriteDeadline(t time.Time) error {
257 panic("not implemented")
258 }