Replacing statsd with datadog statsd

Gopkg.toml

@@ -1,7 +1,3 @@
-[[constraint]]
-  name = "github.com/alexcesaro/statsd"
-  version = "2.0.0"
-
 [[constraint]]
   name = "github.com/confluentinc/confluent-kafka-go"
   version = "0.11.0"
@@ -45,3 +41,7 @@
 [[constraint]]
   name = "gopkg.in/pg.v5"
   version = "5.3.3"
+
+[[constraint]]
+  name = "github.com/DataDog/datadog-go"
+  version = "1.1.0"

config/default.yaml

@@ -45,7 +45,7 @@ stats:
   statsd:
     host: "localhost:8125"
     prefix: "push"
-    flushIntervalMs: 5000
+    buflen: 1
 invalidToken:
   handlers:
     - pg

extensions/apns_message_handler_test.go

@@ -384,7 +384,7 @@ var _ = FDescribe("APNS Message Handler", func() {
 				handler.sendMessage(kafkaMessage)
 				handler.sendMessage(kafkaMessage)
 
-				Expect(mockStatsDClient.Count["apns.game.sent"]).To(Equal(2))
+				Expect(mockStatsDClient.Count["sent"]).To(Equal(2))
 			})
 
 			It("should call HandleNotificationSuccess upon message response received", func() {
@@ -398,7 +398,7 @@ var _ = FDescribe("APNS Message Handler", func() {
 
 				handler.handleAPNSResponse(res)
 				handler.handleAPNSResponse(res)
-				Expect(mockStatsDClient.Count["apns.game.ack"]).To(Equal(2))
+				Expect(mockStatsDClient.Count["ack"]).To(Equal(2))
 			})
 
 			It("should call HandleNotificationFailure upon message response received", func() {
@@ -413,8 +413,8 @@ var _ = FDescribe("APNS Message Handler", func() {
 				handler.handleAPNSResponse(res)
 				handler.handleAPNSResponse(res)
 
-				Expect(mockStatsDClient.Count["apns.game.failed"]).To(Equal(2))
-				Expect(mockStatsDClient.Count["apns.game.missing-device-token"]).To(Equal(2))
+				Expect(mockStatsDClient.Count["failed"]).To(Equal(2))
+				Expect(mockStatsDClient.Count["missing-device-token"]).To(Equal(2))
 			})
 		})
 

extensions/statsd.go → extensions/datadog_statsd.go

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016 TFG Co <backend@tfgco.com>
+ * Copyright (c) 2018 TFG Co <backend@tfgco.com>
  * Author: TFG Co <backend@tfgco.com>
  *
  * Permission is hereby granted, free of charge, to any person obtaining a copy of
@@ -23,9 +23,9 @@
 package extensions
 
 import (
-	"time"
+	"fmt"
 
-	"github.com/alexcesaro/statsd"
+	"github.com/DataDog/datadog-go/statsd"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 	"github.com/topfreegames/pusher/errors"
@@ -56,26 +56,26 @@ func NewStatsD(config *viper.Viper, logger *logrus.Logger, clientOrNil ...interf
 func (s *StatsD) loadConfigurationDefaults() {
 	s.Config.SetDefault("stats.statsd.host", "localhost:8125")
 	s.Config.SetDefault("stats.statsd.prefix", "test")
-	s.Config.SetDefault("stats.statsd.flushIntervalMs", 5000)
+	s.Config.SetDefault("stats.statsd.buflen", 1)
 }
 
 func (s *StatsD) configure(client interfaces.StatsDClient) error {
 	s.loadConfigurationDefaults()
 
 	host := s.Config.GetString("stats.statsd.host")
 	prefix := s.Config.GetString("stats.statsd.prefix")
-	flushIntervalMs := s.Config.GetInt("stats.statsd.flushIntervalMs")
-	flushPeriod := time.Duration(flushIntervalMs) * time.Millisecond
+	buflen := s.Config.GetInt("stats.statsd.buflen")
 
 	l := s.Logger.WithFields(logrus.Fields{
-		"host":            host,
-		"prefix":          prefix,
-		"flushIntervalMs": flushIntervalMs,
+		"host":   host,
+		"prefix": prefix,
+		"buflen": buflen,
 	})
 
 	if client == nil {
-		var err error
-		client, err = statsd.New(statsd.Address(host), statsd.FlushPeriod(flushPeriod), statsd.Prefix(prefix))
+		ddClient, err := statsd.NewBuffered(host, buflen)
+		ddClient.Namespace = prefix
+		client = ddClient
 
 		if err != nil {
 			l.WithError(err).Error("Error configuring statsd client.")
@@ -90,30 +90,29 @@ func (s *StatsD) configure(client interfaces.StatsDClient) error {
 
 //HandleNotificationSent stores notification count in StatsD
 func (s *StatsD) HandleNotificationSent(game string, platform string) {
-	s.Client.Increment(platform + "." + game + "." + "sent")
+	s.Client.Incr("sent", []string{fmt.Sprintf("platform:%s", platform), fmt.Sprintf("game:%s", game)}, 1)
 }
 
 //HandleNotificationSuccess stores notifications success in StatsD
 func (s *StatsD) HandleNotificationSuccess(game string, platform string) {
-	s.Client.Increment(platform + "." + game + "." + "ack")
+	s.Client.Incr("ack", []string{fmt.Sprintf("platform:%s", platform), fmt.Sprintf("game:%s", game)}, 1)
 }
 
 //HandleNotificationFailure stores each type of failure
 func (s *StatsD) HandleNotificationFailure(game string, platform string, err *errors.PushError) {
-	s.Client.Increment(platform + "." + game + "." + "failed")
-	s.Client.Increment(platform + "." + game + "." + err.Key)
+	s.Client.Incr("failed", []string{fmt.Sprintf("platform:%s", platform), fmt.Sprintf("game:%s", game)}, 1)
+	s.Client.Incr(err.Key, []string{fmt.Sprintf("platform:%s", platform), fmt.Sprintf("game:%s", game)}, 1)
 }
 
 //ReportGoStats reports go stats in statsd
 func (s *StatsD) ReportGoStats(
 	numGoRoutines int,
 	allocatedAndNotFreed, heapObjects, nextGCBytes, pauseGCNano uint64,
 ) {
-	s.Client.Gauge("num_goroutine", numGoRoutines)
-	s.Client.Gauge("allocated_not_freed", allocatedAndNotFreed)
-	s.Client.Gauge("heap_objects", heapObjects)
-	s.Client.Gauge("next_gc_bytes", nextGCBytes)
-	s.Client.Timing("gc_pause_duration_ms", pauseGCNano/1000000)
+	s.Client.Gauge("num_goroutine", float64(numGoRoutines), nil, 1)
+	s.Client.Gauge("allocated_not_freed", float64(allocatedAndNotFreed), nil, 1)
+	s.Client.Gauge("heap_objects", float64(heapObjects), nil, 1)
+	s.Client.Gauge("next_gc_bytes", float64(nextGCBytes), nil, 1)
 }
 
 //Cleanup closes statsd connection

extensions/statsd_test.go → extensions/datadog_statsd_test.go

@@ -53,7 +53,7 @@ var _ = Describe("StatsD Extension", func() {
 
 				statsd.HandleNotificationSent("game", "apns")
 				statsd.HandleNotificationSent("game", "apns")
-				Expect(mockClient.Count["apns.game.sent"]).To(Equal(2))
+				Expect(mockClient.Count["sent"]).To(Equal(2))
 			})
 		})
 
@@ -65,7 +65,7 @@ var _ = Describe("StatsD Extension", func() {
 
 				statsd.HandleNotificationSuccess("game", "apns")
 				statsd.HandleNotificationSuccess("game", "apns")
-				Expect(mockClient.Count["apns.game.ack"]).To(Equal(2))
+				Expect(mockClient.Count["ack"]).To(Equal(2))
 			})
 		})
 
@@ -82,7 +82,6 @@ var _ = Describe("StatsD Extension", func() {
 				Expect(mockClient.Gauges["allocated_not_freed"]).To(BeEquivalentTo(3))
 				Expect(mockClient.Gauges["heap_objects"]).To(BeEquivalentTo(4))
 				Expect(mockClient.Gauges["next_gc_bytes"]).To(BeEquivalentTo(5))
-				Expect(mockClient.Timings["gc_pause_duration_ms"]).To(BeEquivalentTo(6))
 			})
 		})
 
@@ -97,8 +96,8 @@ var _ = Describe("StatsD Extension", func() {
 				statsd.HandleNotificationFailure("game", "apns", pErr)
 				statsd.HandleNotificationFailure("game", "apns", pErr)
 
-				Expect(mockClient.Count["apns.game.failed"]).To(Equal(2))
-				Expect(mockClient.Count["apns.game.some-key"]).To(Equal(2))
+				Expect(mockClient.Count["failed"]).To(Equal(2))
+				Expect(mockClient.Count["some-key"]).To(Equal(2))
 			})
 		})
 	})