add statsReporter to tokenPG

extensions/apns_message_handler_test.go

@@ -77,7 +77,7 @@ var _ = FDescribe("APNS Message Handler", func() {
 			feedbackClients = []interfaces.FeedbackReporter{kc}
 
 			db = mocks.NewPGMock(0, 1)
-			it, err := NewTokenPG(config, logger, db)
+			it, err := NewTokenPG(config, logger, statsClients, db)
 			Expect(err).NotTo(HaveOccurred())
 			invalidTokenHandlers = []interfaces.InvalidTokenHandler{it}
 
@@ -135,6 +135,10 @@ var _ = FDescribe("APNS Message Handler", func() {
 				Expect(handler.failuresReceived).To(Equal(int64(1)))
 				Eventually(func() []*logrus.Entry { return hook.Entries }).
 					Should(ContainLogMessage("deleting token"))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeleted] }).
+					Should(Equal(1))
 				//Expect(hook.Entries[len(hook.Entries)-2].Data["category"]).To(Equal("TokenError"))
 			})
 
@@ -149,6 +153,10 @@ var _ = FDescribe("APNS Message Handler", func() {
 				Expect(handler.failuresReceived).To(Equal(int64(1)))
 				Eventually(func() []*logrus.Entry { return hook.Entries }).
 					Should(ContainLogMessage("deleting token"))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeleted] }).
+					Should(Equal(1))
 				//Expect(hook.Entries[len(hook.Entries)-2].Data["category"]).To(Equal("TokenError"))
 			})
 
@@ -427,7 +435,7 @@ var _ = FDescribe("APNS Message Handler", func() {
 				feedbackClients = []interfaces.FeedbackReporter{kc}
 
 				db = mocks.NewPGMock(0, 1)
-				it, err := NewTokenPG(config, logger, db)
+				it, err := NewTokenPG(config, logger, statsClients, db)
 				Expect(err).NotTo(HaveOccurred())
 				invalidTokenHandlers = []interfaces.InvalidTokenHandler{it}
 				handler, err = NewAPNSMessageHandler(

extensions/common.go

@@ -120,3 +120,9 @@ func statsReporterHandleNotificationFailure(statsReporters []interfaces.StatsRep
 		statsReporter.HandleNotificationFailure(game, platform, err)
 	}
 }
+
+func statsReporterReportMetricIncrement(statsReporters []interfaces.StatsReporter, metric string, game string, platform string) {
+	for _, statsReporter := range statsReporters {
+		statsReporter.ReportMetricIncrement(metric, game, platform)
+	}
+}

extensions/common_test.go

@@ -45,6 +45,8 @@ var _ = Describe("Common", func() {
 	var feedbackClients []interfaces.FeedbackReporter
 	var invalidTokenHandlers []interfaces.InvalidTokenHandler
 	var db *mocks.PGMock
+	var mockStatsDClient *mocks.StatsDClientMock
+	var statsClients []interfaces.StatsReporter
 
 	configFile := "../config/test.yaml"
 	logger, hook := test.NewNullLogger()
@@ -58,10 +60,16 @@ var _ = Describe("Common", func() {
 			mockKafkaProducerClient = mocks.NewKafkaProducerClientMock()
 			kc, err := NewKafkaProducer(config, logger, mockKafkaProducerClient)
 			Expect(err).NotTo(HaveOccurred())
+
+			mockStatsDClient = mocks.NewStatsDClientMock()
+			c, err := NewStatsD(config, logger, mockStatsDClient)
+			Expect(err).NotTo(HaveOccurred())
+
+			statsClients = []interfaces.StatsReporter{c}
 			feedbackClients = []interfaces.FeedbackReporter{kc}
 
 			db = mocks.NewPGMock(0, 1)
-			it, err := NewTokenPG(config, logger, db)
+			it, err := NewTokenPG(config, logger, statsClients, db)
 			Expect(err).NotTo(HaveOccurred())
 			invalidTokenHandlers = []interfaces.InvalidTokenHandler{it}
 
@@ -80,6 +88,10 @@ var _ = Describe("Common", func() {
 					Should(BeEquivalentTo(query))
 				Eventually(func() interface{} { return db.Execs[1][1] }).
 					Should(BeEquivalentTo([]interface{}{token}))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeleted] }).
+					Should(Equal(1))
 			})
 
 			It("should fail silently", func() {
@@ -94,6 +106,10 @@ var _ = Describe("Common", func() {
 					Should(BeEquivalentTo(query))
 				Eventually(func() interface{} { return db.Execs[1][1] }).
 					Should(BeEquivalentTo([]interface{}{token}))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeletionError] }).
+					Should(Equal(1))
 			})
 
 			Describe("should stop handler gracefully", func() {

extensions/gcm_message_handler_test.go

@@ -73,7 +73,7 @@ var _ = Describe("GCM Message Handler", func() {
 			feedbackClients = []interfaces.FeedbackReporter{kc}
 
 			mockDb = mocks.NewPGMock(0, 1)
-			it, err := NewTokenPG(config, logger, mockDb)
+			it, err := NewTokenPG(config, logger, statsClients, mockDb)
 			Expect(err).NotTo(HaveOccurred())
 			invalidTokenHandlers = []interfaces.InvalidTokenHandler{it}
 
@@ -135,6 +135,10 @@ var _ = Describe("GCM Message Handler", func() {
 				Expect(handler.failuresReceived).To(Equal(int64(1)))
 				Eventually(func() []*logrus.Entry { return hook.Entries }).
 					Should(ContainLogMessage("deleting token"))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeleted] }).
+					Should(Equal(1))
 			})
 
 			It("if response has error BAD_REGISTRATION", func() {
@@ -146,6 +150,10 @@ var _ = Describe("GCM Message Handler", func() {
 				Expect(handler.failuresReceived).To(Equal(int64(1)))
 				Eventually(func() []*logrus.Entry { return hook.Entries }).
 					Should(ContainLogMessage("deleting token"))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeleted] }).
+					Should(Equal(1))
 			})
 
 			It("if response has error INVALID_JSON", func() {
@@ -509,7 +517,7 @@ var _ = Describe("GCM Message Handler", func() {
 				feedbackClients = []interfaces.FeedbackReporter{kc}
 
 				mockClient = mocks.NewGCMClientMock()
-				it, err := NewTokenPG(config, logger, mockDb)
+				it, err := NewTokenPG(config, logger, statsClients, mockDb)
 				Expect(err).NotTo(HaveOccurred())
 				invalidTokenHandlers = []interfaces.InvalidTokenHandler{it}
 

extensions/token_pg.go

@@ -34,6 +34,14 @@ import (
 	"github.com/topfreegames/pusher/util"
 )
 
+// Metrics name sent by TokenPG
+const (
+	MetricsTokensToDelete           = "tokens_to_delete"
+	MetricsTokensDeleted            = "tokens_deleted"
+	MetricsTokensDeletionError      = "tokens_deletion_error"
+	MetricsTokensHandleAfterClosing = "tokens_handle_after_closing"
+)
+
 // TokenPG for deleting invalid tokens from the database
 type TokenPG struct {
 	Client *PGClient
@@ -44,6 +52,8 @@ type TokenPG struct {
 	done           chan struct{}
 	closed         bool
 	wg             sync.WaitGroup
+
+	StatsReporters []interfaces.StatsReporter
 }
 
 // TokenMsg represents a token to be deleted in the db
@@ -54,10 +64,15 @@ type TokenMsg struct {
 }
 
 // NewTokenPG for creating a new TokenPG instance
-func NewTokenPG(config *viper.Viper, logger *logrus.Logger, dbOrNil ...interfaces.DB) (*TokenPG, error) {
+func NewTokenPG(config *viper.Viper,
+	logger *logrus.Logger,
+	statsReporters []interfaces.StatsReporter,
+	dbOrNil ...interfaces.DB,
+) (*TokenPG, error) {
 	q := &TokenPG{
-		Config: config,
-		Logger: logger,
+		Config:         config,
+		Logger:         logger,
+		StatsReporters: statsReporters,
 	}
 	var db interfaces.DB
 	if len(dbOrNil) == 1 {
@@ -104,6 +119,13 @@ func (t *TokenPG) tokenPGWorker() {
 			err := t.deleteToken(tkMsg)
 			if err != nil {
 				l.WithError(err).Error("error deleting token")
+				statsReporterReportMetricIncrement(t.StatsReporters,
+					MetricsTokensDeletionError, tkMsg.Game, tkMsg.Platform,
+				)
+			} else {
+				statsReporterReportMetricIncrement(t.StatsReporters,
+					MetricsTokensDeleted, tkMsg.Game, tkMsg.Platform,
+				)
 			}
 
 		case <-t.done:
@@ -113,7 +135,7 @@ func (t *TokenPG) tokenPGWorker() {
 	}
 }
 
-// HandleToken handles an invalid token. It sends it to a channel to be process by
+// HandleToken handles an invalid token. It sends it to a channel to be processed by
 // the PGToken worker. If the TokenPG was stopped, an error is returned
 func (t *TokenPG) HandleToken(token string, game string, platform string) error {
 	l := t.Logger.WithFields(logrus.Fields{
@@ -130,10 +152,16 @@ func (t *TokenPG) HandleToken(token string, game string, platform string) error
 	if t.closed {
 		e := errors.New("can't handle more tokens. The TokenPG has been stopped")
 		l.Error(e)
+		statsReporterReportMetricIncrement(t.StatsReporters,
+			MetricsTokensHandleAfterClosing, tkMsg.Game, tkMsg.Platform,
+		)
 		return e
 	}
 
 	t.tokensToDelete <- tkMsg
+	statsReporterReportMetricIncrement(t.StatsReporters,
+		MetricsTokensToDelete, tkMsg.Game, tkMsg.Platform,
+	)
 	return nil
 }
 
@@ -174,6 +202,10 @@ func (t *TokenPG) Stop() error {
 	l.Info("waiting the worker to finish")
 	t.wg.Wait()
 	l.Info("tokenPG closed")
+	if ut := len(t.tokensToDelete); ut > 0 {
+		l.Warnf("%d tokens haven't been processed", ut)
+	}
+
 	t.closed = true
 
 	return nil

extensions/token_pg_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/spf13/viper"
+	"github.com/topfreegames/pusher/interfaces"
 	"github.com/topfreegames/pusher/mocks"
 	testing "github.com/topfreegames/pusher/testing"
 	"github.com/topfreegames/pusher/util"
@@ -41,23 +42,32 @@ var _ = Describe("TokenPG Extension", func() {
 	var config *viper.Viper
 	var mockClient *mocks.PGMock
 	var tokenPG *TokenPG
+	var mockStatsDClient *mocks.StatsDClientMock
+	var statsClients []interfaces.StatsReporter
 	logger, hook := test.NewNullLogger()
 	token := uuid.NewV4().String()
 
 	BeforeEach(func() {
 		var err error
 		config, err = util.NewViperWithConfigFile("../config/test.yaml")
 		Expect(err).NotTo(HaveOccurred())
+
+		mockStatsDClient = mocks.NewStatsDClientMock()
+		c, err := NewStatsD(config, logger, mockStatsDClient)
+		Expect(err).NotTo(HaveOccurred())
+
+		statsClients = []interfaces.StatsReporter{c}
+
 		mockClient = mocks.NewPGMock(0, 1)
-		tokenPG, err = NewTokenPG(config, logger, mockClient)
+		tokenPG, err = NewTokenPG(config, logger, statsClients, mockClient)
 		Expect(err).NotTo(HaveOccurred())
 		hook.Reset()
 	})
 
 	Describe("[Unit]", func() {
 		Describe("Creating new client", func() {
 			It("should return connected client", func() {
-				t, err := NewTokenPG(config, logger, mockClient)
+				t, err := NewTokenPG(config, logger, statsClients, mockClient)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(t).NotTo(BeNil())
 				Expect(t.Client).NotTo(BeNil())
@@ -69,7 +79,7 @@ var _ = Describe("TokenPG Extension", func() {
 
 			It("should return an error if failed to connect to postgres", func() {
 				mockClient.RowsReturned = 0
-				t, err := NewTokenPG(config, logger, mockClient)
+				t, err := NewTokenPG(config, logger, statsClients, mockClient)
 				Expect(t).NotTo(BeNil())
 				Expect(err).To(HaveOccurred())
 				Expect(err.Error()).To(Equal("Timed out waiting for PostgreSQL to connect"))
@@ -90,6 +100,10 @@ var _ = Describe("TokenPG Extension", func() {
 					Should(BeEquivalentTo(query))
 				Eventually(func() interface{} { return mockClient.Execs[1][1] }).
 					Should(BeEquivalentTo([]interface{}{token}))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeleted] }).
+					Should(Equal(1))
 			})
 
 			It("should not break if token does not exist in db", func() {
@@ -105,6 +119,10 @@ var _ = Describe("TokenPG Extension", func() {
 					Should(BeEquivalentTo(query))
 				Eventually(func() interface{} { return mockClient.Execs[1][1] }).
 					Should(BeEquivalentTo([]interface{}{token}))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeleted] }).
+					Should(Equal(1))
 			})
 
 			It("should return an error if pg error occurred", func() {
@@ -113,6 +131,10 @@ var _ = Describe("TokenPG Extension", func() {
 
 				Eventually(func() []*logrus.Entry { return hook.Entries }).
 					Should(testing.ContainLogMessage(deletionError))
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(1))
+				Eventually(func() int { return mockStatsDClient.Count[MetricsTokensDeletionError] }).
+					Should(Equal(1))
 			})
 		})
 
@@ -139,11 +161,15 @@ var _ = Describe("TokenPG Extension", func() {
 					To(testing.ContainLogMessage("waiting the worker to finish"))
 				Expect(hook.Entries).
 					To(testing.ContainLogMessage("tokenPG closed"))
+				Expect(hook.Entries).
+					To(testing.ContainLogMessage(fmt.Sprintf("%d tokens haven't been processed", len(tokenPG.tokensToDelete))))
 
 				err := tokenPG.HandleToken(token, "test", "apns")
 				Expect(err).To(HaveOccurred())
 				Expect(err.Error()).To(Equal("can't handle more tokens. The TokenPG has been stopped"))
 				Consistently(tokenPG.HasJob).Should(BeTrue())
+
+				Expect(mockStatsDClient.Count[MetricsTokensHandleAfterClosing]).To(Equal(1))
 			})
 
 			It("waiting finish jobs to stop", func() {
@@ -170,6 +196,9 @@ var _ = Describe("TokenPG Extension", func() {
 					Expect(mockClient.Execs[i+1][0]).To(BeIdenticalTo(query))
 					Expect(mockClient.Execs[i+1][1]).To(BeEquivalentTo([]interface{}{tokens[i]}))
 				}
+
+				Expect(mockStatsDClient.Count[MetricsTokensToDelete]).To(Equal(size))
+				Expect(mockStatsDClient.Count[MetricsTokensDeleted]).To(Equal(size))
 			})
 		})
 	})

pusher/apns.go

@@ -160,7 +160,7 @@ func (a *APNSPusher) configureStatsReporters(clientOrNil interfaces.StatsDClient
 }
 
 func (a *APNSPusher) configureInvalidTokenHandlers(dbOrNil interfaces.DB) error {
-	invalidTokenHandlers, err := configureInvalidTokenHandlers(a.Config, a.Logger, dbOrNil)
+	invalidTokenHandlers, err := configureInvalidTokenHandlers(a.Config, a.Logger, a.StatsReporters, dbOrNil)
 	if err != nil {
 		return err
 	}

pusher/gcm.go

@@ -154,7 +154,7 @@ func (g *GCMPusher) configureFeedbackReporters() error {
 }
 
 func (g *GCMPusher) configureInvalidTokenHandlers(dbOrNil interfaces.DB) error {
-	invalidTokenHandlers, err := configureInvalidTokenHandlers(g.Config, g.Logger, dbOrNil)
+	invalidTokenHandlers, err := configureInvalidTokenHandlers(g.Config, g.Logger, g.StatsReporters, dbOrNil)
 	if err != nil {
 		return err
 	}

pusher/handlers.go

@@ -31,16 +31,16 @@ import (
 	"github.com/topfreegames/pusher/interfaces"
 )
 
-type invalidTokenHandlerInitializer func(*viper.Viper, *logrus.Logger, interfaces.DB) (interfaces.InvalidTokenHandler, error)
+type invalidTokenHandlerInitializer func(*viper.Viper, *logrus.Logger, []interfaces.StatsReporter, interfaces.DB) (interfaces.InvalidTokenHandler, error)
 
 // AvailableInvalidTokenHandlers contains functions to initialize all invalid token handlers
 var AvailableInvalidTokenHandlers = map[string]invalidTokenHandlerInitializer{
-	"pg": func(config *viper.Viper, logger *logrus.Logger, dbOrNil interfaces.DB) (interfaces.InvalidTokenHandler, error) {
-		return extensions.NewTokenPG(config, logger, dbOrNil)
+	"pg": func(config *viper.Viper, logger *logrus.Logger, statsReporter []interfaces.StatsReporter, dbOrNil interfaces.DB) (interfaces.InvalidTokenHandler, error) {
+		return extensions.NewTokenPG(config, logger, statsReporter, dbOrNil)
 	},
 }
 
-func configureInvalidTokenHandlers(config *viper.Viper, logger *logrus.Logger, dbOrNil interfaces.DB) ([]interfaces.InvalidTokenHandler, error) {
+func configureInvalidTokenHandlers(config *viper.Viper, logger *logrus.Logger, statsReporter []interfaces.StatsReporter, dbOrNil interfaces.DB) ([]interfaces.InvalidTokenHandler, error) {
 	invalidTokenHandlers := []interfaces.InvalidTokenHandler{}
 	invalidTokenHandlerNames := config.GetStringSlice("invalidToken.handlers")
 	for _, invalidTokenHandlerName := range invalidTokenHandlerNames {
@@ -49,7 +49,7 @@ func configureInvalidTokenHandlers(config *viper.Viper, logger *logrus.Logger, d
 			return nil, fmt.Errorf("Failed to initialize %s. Invalid Token Handler not available.", invalidTokenHandlerName)
 		}
 
-		r, err := invalidTokenHandlerFunc(config, logger, dbOrNil)
+		r, err := invalidTokenHandlerFunc(config, logger, statsReporter, dbOrNil)
 		if err != nil {
 			return nil, fmt.Errorf("Failed to initialize %s. %s", invalidTokenHandlerName, err.Error())
 		}