fix: re-use the maxQueuedEvents (kyverno#10024) (kyverno#10031)

* fix: re-use the maxQueuedEvents * fix: use the apierrors.IsNotFound instead of checking a specfic error msg --------- Signed-off-by: Mariam Fahmy <[email protected]> Co-authored-by: Mariam Fahmy <[email protected]> Co-authored-by: shuting <[email protected]>
nirmata · Apr 11, 2024 · 6c71685 · 6c71685
1 parent f8c058c
commit 6c71685
Show file tree

Hide file tree

Showing 6 changed files with 14 additions and 3 deletions.
diff --git a/cmd/background-controller/main.go b/cmd/background-controller/main.go
@@ -147,6 +147,7 @@ func main() {
 	eventGenerator := event.NewEventGenerator(
 		setup.EventsClient,
 		logging.WithName("EventGenerator"),
+		maxQueuedEvents,
 		strings.Split(omitEvents, ",")...,
 	)
 	eventController := internal.NewController(

diff --git a/cmd/cleanup-controller/main.go b/cmd/cleanup-controller/main.go
@@ -154,6 +154,7 @@ func main() {
 	eventGenerator := event.NewEventGenerator(
 		setup.EventsClient,
 		logging.WithName("EventGenerator"),
+		maxQueuedEvents,
 	)
 	eventController := internal.NewController(
 		event.ControllerName,

diff --git a/cmd/kyverno/main.go b/cmd/kyverno/main.go
@@ -354,6 +354,7 @@ func main() {
 	eventGenerator := event.NewEventGenerator(
 		setup.EventsClient,
 		logging.WithName("EventGenerator"),
+		maxQueuedEvents,
 		strings.Split(omitEvents, ",")...,
 	)
 	gcstore := store.New()

diff --git a/cmd/reports-controller/main.go b/cmd/reports-controller/main.go
@@ -265,6 +265,7 @@ func main() {
 	eventGenerator := event.NewEventGenerator(
 		setup.EventsClient,
 		logging.WithName("EventGenerator"),
+		maxQueuedEvents,
 		strings.Split(omitEvents, ",")...,
 	)
 	eventController := internal.NewController(

diff --git a/pkg/event/controller.go b/pkg/event/controller.go
@@ -13,6 +13,7 @@ import (
 	"go.opentelemetry.io/otel/metric"
 	corev1 "k8s.io/api/core/v1"
 	eventsv1 "k8s.io/api/events/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -44,10 +45,11 @@ type controller struct {
 	clock                clock.Clock
 	hostname             string
 	droppedEventsCounter metric.Int64Counter
+	maxQueuedEvents      int
 }
 
 // NewEventGenerator to generate a new event controller
-func NewEventGenerator(eventsClient v1.EventsV1Interface, logger logr.Logger, omitEvents ...string) *controller {
+func NewEventGenerator(eventsClient v1.EventsV1Interface, logger logr.Logger, maxQueuedEvents int, omitEvents ...string) *controller {
 	clock := clock.RealClock{}
 	hostname, _ := os.Hostname()
 	meter := otel.GetMeterProvider().Meter(metrics.MeterName)
@@ -66,13 +68,18 @@ func NewEventGenerator(eventsClient v1.EventsV1Interface, logger logr.Logger, om
 		clock:                clock,
 		hostname:             hostname,
 		droppedEventsCounter: droppedEventsCounter,
+		maxQueuedEvents:      maxQueuedEvents,
 	}
 }
 
 // Add queues an event for generation
 func (gen *controller) Add(infos ...Info) {
 	logger := gen.logger
 	logger.V(3).Info("generating events", "count", len(infos))
+	if gen.maxQueuedEvents == 0 || gen.queue.Len() > gen.maxQueuedEvents {
+		logger.V(2).Info("exceeds the event queue limit, dropping the event", "maxQueuedEvents", gen.maxQueuedEvents, "current size", gen.queue.Len())
+		return
+	}
 	for _, info := range infos {
 		// don't create event for resources with generateName as the name is not generated yet
 		if info.Regarding.Name == "" {
@@ -119,7 +126,7 @@ func (gen *controller) processNextWorkItem(ctx context.Context) bool {
 		return true
 	}
 	_, err := gen.eventsClient.Events(event.Namespace).Create(ctx, event, metav1.CreateOptions{})
-	if err != nil {
+	if err != nil && !apierrors.IsNotFound(err) {
 		if gen.queue.NumRequeues(key) < workQueueRetryLimit {
 			logger.Error(err, "failed to create event", "key", key)
 			gen.queue.AddRateLimited(key)

diff --git a/pkg/event/controller_test.go b/pkg/event/controller_test.go
@@ -27,7 +27,7 @@ func TestEventGenerator(t *testing.T) {
 	logger := logr.Discard()
 
 	eventsClient := clientset.EventsV1()
-	eventGenerator := NewEventGenerator(eventsClient, logger)
+	eventGenerator := NewEventGenerator(eventsClient, logger, 1000)
 
 	go eventGenerator.Run(ctx, Workers)
 	time.Sleep(1 * time.Second)