feat: support learning mode for event reporter

changelog/CHANGELOG.md

@@ -0,0 +1,2 @@
+### Event reporter v2.0.4
+1. Support learning mode for event reporter.

cmd/event-reporter-server/commands/event_reporter_server.go

@@ -95,9 +95,10 @@ func NewCommand() *cobra.Command {
 		rootpath                 string
 		useGrpc                  bool
 
-		rateLimiterEnabled    bool
-		rateLimiterBucketSize int
-		rateLimiterDuration   time.Duration
+		rateLimiterEnabled      bool
+		rateLimiterBucketSize   int
+		rateLimiterDuration     time.Duration
+		rateLimiterLearningMode bool
 	)
 	var command = &cobra.Command{
 		Use:               cliName,
@@ -178,9 +179,10 @@ func NewCommand() *cobra.Command {
 					AuthToken: codefreshToken,
 				},
 				RateLimiterOpts: &reporter.RateLimiterOpts{
-					Enabled:  rateLimiterEnabled,
-					Rate:     rateLimiterDuration,
-					Capacity: rateLimiterBucketSize,
+					Enabled:      rateLimiterEnabled,
+					Rate:         rateLimiterDuration,
+					Capacity:     rateLimiterBucketSize,
+					LearningMode: rateLimiterLearningMode,
 				},
 			}
 
@@ -231,6 +233,7 @@ func NewCommand() *cobra.Command {
 	command.Flags().BoolVar(&rateLimiterEnabled, "rate-limiter-enabled", env.ParseBoolFromEnv("RATE_LIMITER_ENABLED", false), "Use rate limiter for prevent queue to be overflowed")
 	command.Flags().IntVar(&rateLimiterBucketSize, "rate-limiter-bucket-size", env.ParseNumFromEnv("RATE_LIMITER_BUCKET_SIZE", math.MaxInt, 0, math.MaxInt), "The maximum amount of requests allowed per window.")
 	command.Flags().DurationVar(&rateLimiterDuration, "rate-limiter-period", env.ParseDurationFromEnv("RATE_LIMITER_DURATION", 24*time.Hour, 0, math.MaxInt64), "The rate limit window size.")
+	command.Flags().BoolVar(&rateLimiterLearningMode, "rate-limiter-learning-mode", env.ParseBoolFromEnv("RATE_LIMITER_LEARNING_MODE_ENABLED", false), "The rate limit enabled in learning mode ( not blocking sending to queue but logging it )")
 	cacheSrc = servercache.AddCacheFlagsToCmd(command, func(client *redis.Client) {
 		redisClient = client
 	})

event_reporter/metrics/metrics.go

@@ -70,7 +70,7 @@ var (
 			Name: "cf_e_reporter_app_events_size",
 			Help: "Size of specific application events queue of taked shard.",
 		},
-		[]string{"reporter_shard", "application", "got_in_queue"},
+		[]string{"reporter_shard", "application", "got_in_queue", "error_in_learning_mode"},
 	)
 	erroredEventsCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
@@ -145,8 +145,8 @@ func (m *MetricsServer) SetQueueSizeCounter(size int) {
 	m.queueSizeCounter.WithLabelValues(m.shard).Set(float64(size))
 }
 
-func (m *MetricsServer) IncAppEventsCounter(application string, gotToProcessingQueue bool) {
-	m.appEventsCounter.WithLabelValues(m.shard, application, strconv.FormatBool(gotToProcessingQueue)).Inc()
+func (m *MetricsServer) IncAppEventsCounter(application string, gotToProcessingQueue bool, errorInLearningMode bool) {
+	m.appEventsCounter.WithLabelValues(m.shard, application, strconv.FormatBool(gotToProcessingQueue), strconv.FormatBool(errorInLearningMode)).Inc()
 }
 
 func (m *MetricsServer) IncErroredEventsCounter(metricEventType MetricEventType, errorType MetricEventErrorType, application string) {

event_reporter/reporter/broadcaster.go

@@ -78,23 +78,27 @@ func (b *broadcasterHandler) notify(event *appv1.ApplicationWatchEvent) {
 	for _, s := range subscribers {
 		if s.matches(event) {
 
-			duration, err := b.rateLimiter.Limit(event.Application.Name)
+			duration, err, learningMode := b.rateLimiter.Limit(event.Application.Name)
+			errorInLearningMode := learningMode && err != nil
 			if err != nil {
-				log.Infof("adding application '%s' to channel failed, due to rate limit, duration left %s", event.Application.Name, duration.String())
-				b.metricsServer.IncAppEventsCounter(event.Application.Name, false)
-				continue
+				log.Errorf("adding application '%s' to channel failed, due to rate limit, duration left %s, learningMode %t", event.Application.Name, duration.String(), learningMode)
+				// if learning mode is enabled, we will continue to send events
+				if !learningMode {
+					b.metricsServer.IncAppEventsCounter(event.Application.Name, false, false)
+					continue
+				}
 			}
 
 			select {
 			case s.ch <- event:
 				{
 					log.Infof("adding application '%s' to channel", event.Application.Name)
-					b.metricsServer.IncAppEventsCounter(event.Application.Name, true)
+					b.metricsServer.IncAppEventsCounter(event.Application.Name, true, errorInLearningMode)
 				}
 			default:
 				// drop event if cannot send right away
 				log.WithField("application", event.Application.Name).Warn("unable to send event notification")
-				b.metricsServer.IncAppEventsCounter(event.Application.Name, false)
+				b.metricsServer.IncAppEventsCounter(event.Application.Name, false, errorInLearningMode)
 			}
 		}
 	}

event_reporter/reporter/rate_limiter.go

@@ -7,9 +7,10 @@ import (
 )
 
 type RateLimiterOpts struct {
-	Enabled  bool
-	Rate     time.Duration
-	Capacity int
+	Enabled      bool
+	Rate         time.Duration
+	Capacity     int
+	LearningMode bool
 }
 
 type RateLimiter struct {
@@ -21,9 +22,9 @@ func NewRateLimiter(opts *RateLimiterOpts) *RateLimiter {
 	return &RateLimiter{opts: opts, limiters: make(map[string]*limiters.FixedWindow)}
 }
 
-func (rl *RateLimiter) Limit(applicationName string) (time.Duration, error) {
+func (rl *RateLimiter) Limit(applicationName string) (time.Duration, error, bool) {
 	if !rl.opts.Enabled {
-		return time.Duration(0), nil
+		return time.Duration(0), nil, rl.opts.LearningMode
 	}
 
 	limiter := rl.limiters[applicationName]
@@ -32,5 +33,6 @@ func (rl *RateLimiter) Limit(applicationName string) (time.Duration, error) {
 		rl.limiters[applicationName] = limiter
 	}
 
-	return limiter.Limit(context.Background())
+	duration, err := limiter.Limit(context.Background())
+	return duration, err, rl.opts.LearningMode
 }

event_reporter/reporter/rate_limiter_test.go

@@ -10,7 +10,7 @@ func TestRateLimiter(t *testing.T) {
 		rl := NewRateLimiter(&RateLimiterOpts{
 			Enabled: false,
 		})
-		d, err := rl.Limit("foo")
+		d, err, _ := rl.Limit("foo")
 		if err != nil {
 			t.Errorf("Expected no error, got %v", err)
 		}
@@ -24,7 +24,7 @@ func TestRateLimiter(t *testing.T) {
 			Rate:     time.Second,
 			Capacity: 1,
 		})
-		d, err := rl.Limit("foo")
+		d, err, _ := rl.Limit("foo")
 		if err != nil {
 			t.Errorf("Expected no error, got %v", err)
 		}
@@ -38,7 +38,7 @@ func TestRateLimiter(t *testing.T) {
 			Rate:     time.Second,
 			Capacity: 0,
 		})
-		_, err := rl.Limit("foo")
+		_, err, _ := rl.Limit("foo")
 		if err == nil {
 			t.Errorf("Expected error, got nil")
 		}