chore: add labels to redis failure metric

config/default.yaml

@@ -1,7 +1,7 @@
 ---
 gracefulShutdownTimeout: 30
 apns:
-  rateLimit.rpm: 100
+  rateLimit.rpm: 20
   concurrentWorkers: 300
   connectionPoolSize: 1
   pushQueueSize: 100
@@ -15,7 +15,7 @@ apns:
       teamID: "ABC123DEFG"
       topic: "com.game.test"
 gcm:
-  rateLimit.rpm: 100
+  rateLimit.rpm: 20
   pingInterval: 30
   pingTimeout: 10
   maxPendingMessages: 100

extensions/apns_message_handler.go

@@ -231,7 +231,7 @@ func (a *APNSMessageHandler) HandleMessages(ctx context.Context, message interfa
 		return
 	}
 
-	allowed := a.rateLimiter.Allow(ctx, notification.DeviceToken)
+	allowed := a.rateLimiter.Allow(ctx, notification.DeviceToken, a.appName, "apns")
 	if !allowed {
 		statsReporterNotificationRateLimitReached(a.StatsReporters, a.appName, "apns")
 		l.WithField("message", message).Warn("rate limit reached")

extensions/common.go

@@ -94,9 +94,9 @@ func statsReporterNotificationRateLimitReached(statsReporters []interfaces.Stats
 	}
 }
 
-func statsReporterNotificationRateLimitFailed(statsReporters []interfaces.StatsReporter) {
+func statsReporterNotificationRateLimitFailed(statsReporters []interfaces.StatsReporter, game string, platform string) {
 	for _, statsReporter := range statsReporters {
-		statsReporter.NotificationRateLimitFailed()
+		statsReporter.NotificationRateLimitFailed(game, platform)
 	}
 }
 

extensions/datadog_statsd.go

@@ -121,10 +121,10 @@ func (s *StatsD) NotificationRateLimitReached(game string, platform string) {
 }
 
 // NotificationRateLimitFailed stores how many times rate limits failed to be calculated
-func (s *StatsD) NotificationRateLimitFailed() {
+func (s *StatsD) NotificationRateLimitFailed(game string, platform string) {
 	s.Client.Incr(
 		"rate_limit_failed",
-		[]string{},
+		[]string{fmt.Sprintf("platform:%s", platform), fmt.Sprintf("game:%s", game)},
 		1,
 	)
 }

extensions/gcm_message_handler.go

@@ -345,7 +345,7 @@ func (g *GCMMessageHandler) sendMessage(message interfaces.KafkaMessage) error {
 
 	l = l.WithField("message", km)
 
-	allowed := g.rateLimiter.Allow(context.Background(), km.To)
+	allowed := g.rateLimiter.Allow(context.Background(), km.To, message.Game, "gcm")
 	if !allowed {
 		statsReporterNotificationRateLimitReached(g.StatsReporters, message.Game, "gcm")
 		l.WithField("message", message).Warn("rate limit reached")

extensions/handler/message_handler.go

@@ -91,7 +91,7 @@ func (h *messageHandler) HandleMessages(ctx context.Context, msg interfaces.Kafk
 		return
 	}
 
-	allowed := h.rateLimiter.Allow(ctx, km.To)
+	allowed := h.rateLimiter.Allow(ctx, km.To, msg.Game, "gcm")
 	if !allowed {
 		h.reportRateLimitReached(msg.Game)
 		l.WithField("message", msg).Warn("rate limit reached")

extensions/rate_limiter.go

@@ -55,7 +55,7 @@ func NewRateLimiter(limit int, config *viper.Viper, statsReporters []interfaces.
 // Allow checks Redis for the current rate a given device has in the current minute
 // If the rate is lower than the limit, the message is allowed. Otherwise, it is not allowed.
 // Reference: https://redis.io/glossary/rate-limiting/
-func (r rateLimiter) Allow(ctx context.Context, device string) bool {
+func (r rateLimiter) Allow(ctx context.Context, device string, game string, platform string) bool {
 	deviceKey := fmt.Sprintf("%s:%d", device, time.Now().Minute())
 	l := r.l.WithFields(logrus.Fields{
 		"deviceKey": deviceKey,
@@ -66,7 +66,7 @@ func (r rateLimiter) Allow(ctx context.Context, device string) bool {
 	if err != nil && !errors.Is(err, redis.Nil) {
 		// Something went wrong, return true to avoid blocking notifications.
 		l.WithError(err).Error("could not get current rate in redis")
-		statsReporterNotificationRateLimitFailed(r.statsReporters)
+		statsReporterNotificationRateLimitFailed(r.statsReporters, game, platform)
 		return true
 	}
 	if errors.Is(err, redis.Nil) {
@@ -78,7 +78,7 @@ func (r rateLimiter) Allow(ctx context.Context, device string) bool {
 	if err != nil {
 		// Something went wrong, return true to avoid blocking notifications.
 		l.WithError(err).Error("current rate is invalid")
-		statsReporterNotificationRateLimitFailed(r.statsReporters)
+		statsReporterNotificationRateLimitFailed(r.statsReporters, game, platform)
 		return true
 	}
 
@@ -94,7 +94,7 @@ func (r rateLimiter) Allow(ctx context.Context, device string) bool {
 	if err != nil {
 		// Allow the operation even if the transaction fails, to avoid blocking notifications.
 		l.WithError(err).Error("increment to current rate failed")
-		statsReporterNotificationRateLimitFailed(r.statsReporters)
+		statsReporterNotificationRateLimitFailed(r.statsReporters, game, platform)
 	}
 
 	l.WithField("currentRate", current).Debug("current rate allows message")

extensions/rate_limiter_test.go

@@ -27,19 +27,20 @@ var _ = FDescribe("Rate Limiter", func() {
 		config, err := util.NewViperWithConfigFile(configFile)
 		Expect(err).NotTo(HaveOccurred())
 		hook.Reset()
-
+		game := "test"
+		platform := "test"
 		statsClients := []interfaces.StatsReporter{}
 
 		Describe("Rate limiting", func() {
 			It("should return not-allowed when rate limit is reached", func() {
 				rl := NewRateLimiter(1, config, statsClients, logger)
 				ctx := context.Background()
 				device := uuid.NewString()
-				allowed := rl.Allow(ctx, device)
+				allowed := rl.Allow(ctx, device, game, platform)
 				Expect(allowed).To(BeTrue())
 
 				// Should not allow due to reaching limit of 1
-				allowed = rl.Allow(ctx, device)
+				allowed = rl.Allow(ctx, device, game, platform)
 				Expect(allowed).To(BeFalse())
 			})
 
@@ -49,7 +50,7 @@ var _ = FDescribe("Rate Limiter", func() {
 				device := uuid.NewString()
 				currMin := time.Now().Minute()
 
-				allowed := rl.Allow(ctx, device)
+				allowed := rl.Allow(ctx, device, game, platform)
 				Expect(allowed).To(BeTrue())
 
 				key := fmt.Sprintf("%s:%d", device, currMin)
@@ -66,7 +67,7 @@ var _ = FDescribe("Rate Limiter", func() {
 				ctx := context.Background()
 				device := uuid.NewString()
 
-				allowed := rl.Allow(ctx, device)
+				allowed := rl.Allow(ctx, device, game, platform)
 				Expect(allowed).To(BeTrue())
 			})
 

interfaces/rate_limiter.go

@@ -26,5 +26,5 @@ import "context"
 
 // RateLimiter interface for rate limiting notifications per device.
 type RateLimiter interface {
-	Allow(ctx context.Context, device string) bool
+	Allow(ctx context.Context, device, game, platform string) bool
 }

interfaces/stats_reporter.go

@@ -38,7 +38,7 @@ type StatsReporter interface {
 	ReportMetricGauge(metric string, value float64, game string, platform string)
 	ReportMetricCount(metric string, value int64, game string, platform string)
 	NotificationRateLimitReached(game string, platform string)
-	NotificationRateLimitFailed()
+	NotificationRateLimitFailed(game string, platform string)
 	ReportSendNotificationLatency(latencyMs time.Duration, game string, platform string, labels ...string)
 	ReportFirebaseLatency(latencyMs time.Duration, game string, labels ...string)
 }

mocks/rate_limiter.go

@@ -9,6 +9,6 @@ func NewRateLimiterMock() *rateLimiterMock {
 	return &rateLimiterMock{}
 }
 
-func (rl *rateLimiterMock) Allow(ctx context.Context, device string) bool {
+func (rl *rateLimiterMock) Allow(ctx context.Context, device, game, platform string) bool {
 	return true
 }