Repository: nsqio/nsq
File: nsqd/statsd.go (Go)

Code before:
}

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 memStats
	var lastStats []TopicStats
	interval := n.getOpts().StatsdInterval
	ticker := time.NewTicker(interval)
	for {
		select {
		case <-n.exitChan:
			goto exit
		case <-ticker.C:
			addr := n.getOpts().StatsdAddress
			prefix := n.getOpts().StatsdPrefix
			conn, err := net.DialTimeout("udp", addr, time.Second)
			if err != nil {
				n.logf(LOG_ERROR, "failed to create UDP socket to statsd(%s)", addr)
				continue
			}
			sw := writers.NewSpreadWriter(conn, interval-time.Second)
			bw := writers.NewBoundaryBufferedWriter(sw, MaxUDPPacketSize)
			client := statsd.NewClient(bw, prefix)

			n.logf(LOG_INFO, "STATSD: pushing stats to %s", addr)

			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)
				client.Incr(stat, int64(diff))

				stat = fmt.Sprintf("topic.%s.depth", topic.TopicName)
				client.Gauge(stat, topic.Depth)

				stat = fmt.Sprintf("topic.%s.backend_depth", topic.TopicName)

Review comment (suggestion):
This `-time.Second` looks like a tricky bit. It's an assumption that everything besides the deliberate sleeps in `SpreadWriter.flush()` takes less than a second on average. That's probably a generally a reasonable assumption. I wonder if it should be made more obvious though, maybe set as a "const" at the top.

Code after:
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 memStats
	var lastStats []TopicStats
	interval := n.getOpts().StatsdInterval
	ticker := time.NewTicker(interval)
	for {
		select {
		case <-n.exitChan:
			goto exit
		case <-ticker.C:
			addr := n.getOpts().StatsdAddress
			prefix := n.getOpts().StatsdPrefix
			conn, err := net.DialTimeout("udp", addr, time.Second)
			if err != nil {
				n.logf(LOG_ERROR, "failed to create UDP socket to statsd(%s)", addr)
				continue
			}
			sw := writers.NewSpreadWriter(conn, interval-time.Second)
			bw := writers.NewBoundaryBufferedWriter(sw, n.getOpts().StatsdUDPPacketSize)
			client := statsd.NewClient(bw, prefix)

			n.logf(LOG_INFO, "STATSD: pushing stats to %s", addr)

			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)
				client.Incr(stat, int64(diff))

				stat = fmt.Sprintf("topic.%s.depth", topic.TopicName)
				client.Gauge(stat, topic.Depth)

				stat = fmt.Sprintf("topic.%s.backend_depth", topic.TopicName)
				client.Gauge(stat, topic.BackendDepth)