chore: add pr suggestions

config/default.yaml

@@ -95,9 +95,12 @@ feedbackListeners:
       connectionTimeout: 100
 rateLimiter:
   limit:
-    rpm: 100
+    rpm:
+      apns: 100
+      gcm: 100
   redis: 
     host: "localhost"
     port: 6379
     password: ""
-  test: false
+  tls:
+    disabled: false

config/docker_test.yaml

@@ -95,9 +95,12 @@ feedbackListeners:
       connectionTimeout: 100
 rateLimiter:
   limit:
-    rpm: 100
+    rpm:
+      apns: 100
+      gcm: 100
   redis: 
     host: "redis"
     port: 6379
     password: ""
-  test: true
+  tls:
+    disabled: true

config/test.yaml

@@ -93,9 +93,12 @@ feedbackListeners:
       connectionTimeout: 100
 rateLimiter:
   limit:
-    rpm: 100
+    rpm:
+      apns: 100
+      gcm: 100
   redis: 
     host: "localhost"
     port: 6379
     password: ""
-  test: true
+  tls:
+    disabled: true

e2e/fcm_e2e_test.go

@@ -71,7 +71,8 @@ func (s *FcmE2ETestSuite) setupFcmPusher(appName string) (*firebaseMock.MockPush
 	statsReport, err := extensions.NewStatsD(s.vConfig, logger, statsdClientMock)
 	s.Require().NoError(err)
 
-	rateLimiter := extensions.NewRateLimiter(s.vConfig, logger)
+	limit := s.vConfig.GetInt("rateLimiter.limit.rpm.gcm")
+	rateLimiter := extensions.NewRateLimiter(limit, s.vConfig, []interfaces.StatsReporter{statsReport}, logger)
 
 	pushClient := firebaseMock.NewMockPushClient(ctrl)
 	gcmPusher.MessageHandler = map[string]interfaces.MessageHandler{

extensions/apns_message_handler.go

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

extensions/common.go

@@ -94,6 +94,12 @@ func statsReporterNotificationRateLimitReached(statsReporters []interfaces.Stats
 	}
 }
 
+func statsReporterNotificationRateLimitFailed(statsReporters []interfaces.StatsReporter) {
+	for _, statsReporter := range statsReporters {
+		statsReporter.NotificationRateLimitFailed()
+	}
+}
+
 func statsReporterReportSendNotificationLatency(statsReporters []interfaces.StatsReporter, latencyMs time.Duration, game string, platform string, labels ...string) {
 	for _, statsReporter := range statsReporters {
 		statsReporter.ReportSendNotificationLatency(latencyMs, game, platform, labels...)

extensions/datadog_statsd.go

@@ -120,6 +120,15 @@ func (s *StatsD) NotificationRateLimitReached(game string, platform string) {
 	)
 }
 
+// NotificationRateLimitFailed stores how many times rate limits failed to be calculated
+func (s *StatsD) NotificationRateLimitFailed() {
+	s.Client.Incr(
+		"rate_limit_failed",
+		[]string{},
+		1,
+	)
+}
+
 // InitializeFailure notifu error when is impossible tho initilizer an app
 func (s *StatsD) InitializeFailure(game string, platform string) {
 	s.Client.Incr("initialize_failure", []string{fmt.Sprintf("platform:%s", platform), fmt.Sprintf("game:%s", game)}, 1)

extensions/gcm_message_handler.go

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

extensions/handler/message_handler.go

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

extensions/rate_limiter.go

@@ -11,36 +11,40 @@ import (
 	"github.com/redis/go-redis/v9"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
+	"github.com/topfreegames/pusher/interfaces"
 )
 
 type rateLimiter struct {
-	redis    *redis.Client
-	rpmLimit int
-	l        *logrus.Entry
+	redis          *redis.Client
+	rpmLimit       int
+	statsReporters []interfaces.StatsReporter
+	l              *logrus.Entry
 }
 
-func NewRateLimiter(config *viper.Viper, logger *logrus.Logger) rateLimiter {
+func NewRateLimiter(limit int, config *viper.Viper, statsReporters []interfaces.StatsReporter, logger *logrus.Logger) rateLimiter {
 	host := config.GetString("rateLimiter.redis.host")
 	port := config.GetInt("rateLimiter.redis.port")
 	pwd := config.GetString("rateLimiter.redis.password")
-	limit := config.GetInt("rateLimiter.limit.rpm")
-	isTest := config.GetBool("rateLimiter.test")
+	disableTLS := config.GetBool("rateLimiter.tls.disabled")
 
 	addr := fmt.Sprintf("%s:%d", host, port)
 	opts := &redis.Options{
 		Addr:     addr,
 		Password: pwd,
 	}
-	// Setting TLSConfig only for production due to not being able to enable TLS in the integration test container.
-	if !isTest {
+
+	// TLS for integration tests running in containers can raise connection errors.
+	// Not recommended to disable TLS for production.
+	if !disableTLS {
 		opts.TLSConfig = &tls.Config{}
 	}
 
 	rdb := redis.NewClient(opts)
 
 	return rateLimiter{
-		redis:    rdb,
-		rpmLimit: limit,
+		redis:          rdb,
+		rpmLimit:       limit,
+		statsReporters: statsReporters,
 		l: logger.WithFields(logrus.Fields{
 			"extension": "RateLimiter",
 			"rpmLimit":  limit,
@@ -62,6 +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)
 		return true
 	}
 	if errors.Is(err, redis.Nil) {
@@ -73,6 +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)
 		return true
 	}
 
@@ -82,12 +88,13 @@ func (r rateLimiter) Allow(ctx context.Context, device string) bool {
 
 	_, err = r.redis.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
 		pipe.Incr(ctx, deviceKey)
-		pipe.Expire(ctx, deviceKey, 1*time.Minute)
+		pipe.Expire(ctx, deviceKey, time.Minute)
 		return nil
 	})
 	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)
 	}
 
 	l.WithField("currentRate", current).Debug("current rate allows message")

extensions/rate_limiter_test.go

@@ -8,6 +8,7 @@ import (
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"github.com/topfreegames/pusher/interfaces"
 	"github.com/topfreegames/pusher/util"
 
 	"github.com/google/uuid"
@@ -27,10 +28,11 @@ var _ = FDescribe("Rate Limiter", func() {
 		Expect(err).NotTo(HaveOccurred())
 		hook.Reset()
 
+		statsClients := []interfaces.StatsReporter{}
+
 		Describe("Rate limiting", func() {
 			It("should return not-allowed when rate limit is reached", func() {
-				rl := NewRateLimiter(config, logger)
-				rl.rpmLimit = 1
+				rl := NewRateLimiter(1, config, statsClients, logger)
 				ctx := context.Background()
 				device := uuid.NewString()
 				allowed := rl.Allow(ctx, device)
@@ -42,7 +44,7 @@ var _ = FDescribe("Rate Limiter", func() {
 			})
 
 			It("should increment current rate if limit is not reached", func() {
-				rl := NewRateLimiter(config, logger)
+				rl := NewRateLimiter(10, config, statsClients, logger)
 				ctx := context.Background()
 				device := uuid.NewString()
 				currMin := time.Now().Minute()
@@ -56,6 +58,18 @@ var _ = FDescribe("Rate Limiter", func() {
 				Expect(actual).To(BeEquivalentTo("1"))
 			})
 
+			It("should return allowed if redis fails", func() {
+				wrongConfig, err := util.NewViperWithConfigFile(configFile)
+				Expect(err).NotTo(HaveOccurred())
+				wrongConfig.Set("rateLimiter.redis.host", "unreachable")
+				rl := NewRateLimiter(10, wrongConfig, statsClients, logger)
+				ctx := context.Background()
+				device := uuid.NewString()
+
+				allowed := rl.Allow(ctx, device)
+				Expect(allowed).To(BeTrue())
+			})
+
 		})
 	})
 })

interfaces/stats_reporter.go

@@ -38,6 +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()
 	ReportSendNotificationLatency(latencyMs time.Duration, game string, platform string, labels ...string)
 	ReportFirebaseLatency(latencyMs time.Duration, game string, labels ...string)
 }

pusher/apns.go

@@ -98,6 +98,7 @@ func (a *APNSPusher) configure(queue interfaces.APNSPushQueue, db interfaces.DB,
 		keyID := a.ViperConfig.GetString("apns.certs." + k + ".keyID")
 		teamID := a.ViperConfig.GetString("apns.certs." + k + ".teamID")
 		topic := a.ViperConfig.GetString("apns.certs." + k + ".topic")
+		rateLimit := a.ViperConfig.GetInt("rateLimiter.limit.apns.rpm")
 
 		l.WithFields(logrus.Fields{
 			"app":         k,
@@ -120,7 +121,7 @@ func (a *APNSPusher) configure(queue interfaces.APNSPushQueue, db interfaces.DB,
 			a.feedbackReporters,
 			queue,
 			interfaces.ConsumptionManager(q),
-			extensions.NewRateLimiter(a.ViperConfig, l.Logger),
+			extensions.NewRateLimiter(rateLimit, a.ViperConfig, a.StatsReporters, l.Logger),
 		)
 		if err == nil {
 			a.MessageHandler[k] = handler

pusher/gcm.go

@@ -98,6 +98,8 @@ func (g *GCMPusher) createMessageHandlerForApps(ctx context.Context) error {
 	g.MessageHandler = make(map[string]interfaces.MessageHandler)
 	for _, app := range g.Config.GetGcmAppsArray() {
 		credentials := g.ViperConfig.GetString("gcm.firebaseCredentials." + app)
+		rateLimit := g.ViperConfig.GetInt("rateLimiter.limit.rpm.gcm")
+
 		l = l.WithField("app", app)
 		if credentials != "" { // Firebase is configured, use new handler
 			pushClient, err := client.NewFirebaseClient(ctx, credentials, g.Logger)
@@ -111,7 +113,7 @@ func (g *GCMPusher) createMessageHandlerForApps(ctx context.Context) error {
 				pushClient,
 				g.feedbackReporters,
 				g.StatsReporters,
-				extensions.NewRateLimiter(g.ViperConfig, l.Logger),
+				extensions.NewRateLimiter(rateLimit, g.ViperConfig, g.StatsReporters, l.Logger),
 				g.Logger,
 				g.Config.GCM.ConcurrentWorkers,
 			)
@@ -124,7 +126,7 @@ func (g *GCMPusher) createMessageHandlerForApps(ctx context.Context) error {
 				g.Queue.PendingMessagesWaitGroup(),
 				g.StatsReporters,
 				g.feedbackReporters,
-				extensions.NewRateLimiter(g.ViperConfig, l.Logger),
+				extensions.NewRateLimiter(rateLimit, g.ViperConfig, g.StatsReporters, l.Logger),
 			)
 			if err != nil {
 				l.WithError(err).Error("could not create gcm message handler")