/
statsd.go
144 lines (119 loc) · 4.37 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
package main
import (
"fmt"
"github.com/bitly/nsq/util"
"log"
"math"
"runtime"
"sort"
"time"
)
type Uint64Slice []uint64
func (s Uint64Slice) Len() int {
return len(s)
}
func (s Uint64Slice) Swap(i, j int) {
s[i], s[j] = s[j], s[i]
}
func (s Uint64Slice) Less(i, j int) bool {
return s[i] < s[j]
}
func (n *NSQd) statsdLoop() {
var lastMemStats runtime.MemStats
lastStats := make([]TopicStats, 0)
ticker := time.NewTicker(n.options.statsdInterval)
for {
select {
case <-n.exitChan:
goto exit
case <-ticker.C:
statsd := util.NewStatsdClient(n.options.statsdAddress, n.options.statsdPrefix)
err := statsd.CreateSocket()
if err != nil {
log.Printf("ERROR: failed to create UDP socket to statsd(%s)", statsd)
continue
}
log.Printf("STATSD: pushing stats to %s", statsd)
stats := n.getStats()
for _, topic := range stats {
// try to find the topic in the last collection
lastTopic := TopicStats{}
for _, checkTopic := range lastStats {
if topic.TopicName == checkTopic.TopicName {
lastTopic = checkTopic
break
}
}
diff := topic.MessageCount - lastTopic.MessageCount
stat := fmt.Sprintf("topic.%s.message_count", topic.TopicName)
statsd.Incr(stat, int64(diff))
stat = fmt.Sprintf("topic.%s.depth", topic.TopicName)
statsd.Gauge(stat, topic.Depth)
stat = fmt.Sprintf("topic.%s.backend_depth", topic.TopicName)
statsd.Gauge(stat, topic.BackendDepth)
for _, channel := range topic.Channels {
// try to find the channel in the last collection
lastChannel := ChannelStats{}
for _, checkChannel := range lastTopic.Channels {
if channel.ChannelName == checkChannel.ChannelName {
lastChannel = checkChannel
break
}
}
diff := channel.MessageCount - lastChannel.MessageCount
stat := fmt.Sprintf("topic.%s.channel.%s.message_count", topic.TopicName, channel.ChannelName)
statsd.Incr(stat, int64(diff))
stat = fmt.Sprintf("topic.%s.channel.%s.depth", topic.TopicName, channel.ChannelName)
statsd.Gauge(stat, channel.Depth)
stat = fmt.Sprintf("topic.%s.channel.%s.backend_depth", topic.TopicName, channel.ChannelName)
statsd.Gauge(stat, channel.BackendDepth)
stat = fmt.Sprintf("topic.%s.channel.%s.in_flight_count", topic.TopicName, channel.ChannelName)
statsd.Gauge(stat, int64(channel.InFlightCount))
stat = fmt.Sprintf("topic.%s.channel.%s.deferred_count", topic.TopicName, channel.ChannelName)
statsd.Gauge(stat, int64(channel.DeferredCount))
diff = channel.RequeueCount - lastChannel.RequeueCount
stat = fmt.Sprintf("topic.%s.channel.%s.requeue_count", topic.TopicName, channel.ChannelName)
statsd.Incr(stat, int64(diff))
diff = channel.TimeoutCount - lastChannel.TimeoutCount
stat = fmt.Sprintf("topic.%s.channel.%s.timeout_count", topic.TopicName, channel.ChannelName)
statsd.Incr(stat, int64(diff))
stat = fmt.Sprintf("topic.%s.channel.%s.clients", topic.TopicName, channel.ChannelName)
statsd.Gauge(stat, int64(len(channel.Clients)))
}
}
lastStats = stats
if *statsdMemStats {
var memStats runtime.MemStats
runtime.ReadMemStats(&memStats)
// sort the GC pause array
length := 256
if len(memStats.PauseNs) <= 256 {
length = int(memStats.NumGC)
}
gcPauses := make(Uint64Slice, length)
copy(gcPauses, memStats.PauseNs[:])
sort.Sort(gcPauses)
statsd.Gauge("mem.heap_objects", int64(memStats.HeapObjects))
statsd.Gauge("mem.heap_idle_bytes", int64(memStats.HeapIdle))
statsd.Gauge("mem.heap_in_use_bytes", int64(memStats.HeapInuse))
statsd.Gauge("mem.heap_released_bytes", int64(memStats.HeapReleased))
statsd.Gauge("mem.gc_pause_usec_100", int64(percentile(100.0, gcPauses, len(gcPauses))/1000))
statsd.Gauge("mem.gc_pause_usec_99", int64(percentile(99.0, gcPauses, len(gcPauses))/1000))
statsd.Gauge("mem.gc_pause_usec_95", int64(percentile(95.0, gcPauses, len(gcPauses))/1000))
statsd.Gauge("mem.next_gc_bytes", int64(memStats.NextGC))
statsd.Incr("mem.gc_runs", int64(memStats.NumGC-lastMemStats.NumGC))
lastMemStats = memStats
}
statsd.Close()
}
}
exit:
ticker.Stop()
}
func percentile(perc float64, arr []uint64, length int) uint64 {
indexOfPerc := int(math.Ceil(((perc / 100.0) * float64(length)) + 0.5))
if indexOfPerc >= length {
indexOfPerc = length - 1
}
return arr[indexOfPerc]
}