Revert "Try a workaround to reduce cpu usage"

This reverts commit 2eebadb.

Author: thokra-nav

internal/cmd/api/api.go

@@ -204,10 +204,8 @@ func run(ctx context.Context, cfg *Config, log logrus.FieldLogger) error {
 			}
 		}
 
-		log := log.WithField("subsystem", "event-watcher")
-		log.Level = logrus.DebugLevel
 		log.WithField("envs", len(k8sClients)).Info("Start event watcher")
-		eventWatcher := event.NewWatcher(pool, k8sClients, log.WithField("subsystem", "event-watcher"))
+		eventWatcher := event.NewWatcher(pool, k8sClients, log)
 		go eventWatcher.Run(ctx)
 	}
 

internal/kubernetes/event/watcher.go

@@ -3,7 +3,6 @@ package event
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"regexp"
 	"strings"
@@ -29,7 +28,8 @@ type Watcher struct {
 	log     logrus.FieldLogger
 	wg      *pool.ContextPool
 
-	cancel context.CancelFunc
+	// State returns true when the watcher should be started/continue running and false when it should stop.
+	state []chan bool
 }
 
 func NewWatcher(pool *pgxpool.Pool, clients map[string]kubernetes.Interface, log logrus.FieldLogger) *Watcher {
@@ -42,75 +42,78 @@ func NewWatcher(pool *pgxpool.Pool, clients map[string]kubernetes.Interface, log
 		events:  make(chan eventsql.UpsertParams, 20),
 		queries: eventsql.New(pool),
 		log:     log,
+		state:   chs,
 	}
 }
 
 func (w *Watcher) Run(ctx context.Context) {
-	w.wg = pool.New().WithContext(ctx)
+	w.wg = pool.New().WithErrors().WithContext(ctx)
 
 	leaderelection.RegisterOnStartedLeading(w.onStartedLeading)
 	leaderelection.RegisterOnStoppedLeading(w.onStoppedLeading)
 	if leaderelection.IsLeader() {
-		w.log.Debug("Is already leader, force start")
 		w.onStartedLeading(ctx)
 	}
 
 	w.wg.Go(func(ctx context.Context) error {
 		return w.batchInsert(ctx)
 	})
 
+	i := 0
+	for env, client := range w.clients {
+		ch := w.state[i]
+		i++
+		w.wg.Go(func(ctx context.Context) error {
+			return w.run(ctx, env, client, ch)
+		})
+	}
+
 	if err := w.wg.Wait(); err != nil {
 		w.log.WithError(err).Error("error running events watcher")
 	}
 }
 
 func (w *Watcher) onStoppedLeading() {
-	w.log.Debug("onStoppedLeading...")
-	if w.cancel != nil {
-		w.cancel()
-		w.cancel = nil
-
-		w.log.Debug("cancelling")
+	for _, ch := range w.state {
+		select {
+		case ch <- false:
+		default:
+			w.log.WithField("state", "stopped").Error("failed to send state")
+		}
 	}
 }
 
-func (w *Watcher) onStartedLeading(ctx context.Context) {
-	if w.cancel != nil {
-		w.cancel()
-	}
-
-	go func() {
-		time.Sleep(5 * time.Second)
-		w.onStoppedLeading()
-	}()
-
-	ctx, cancel := context.WithCancel(ctx)
-	w.cancel = cancel
-
-	for env, client := range w.clients {
-		w.wg.Go(func(_ context.Context) error {
-			w.log.WithField("env", env).Debug("starting watcher")
-			return w.run(ctx, env, client)
-		})
+func (w *Watcher) onStartedLeading(_ context.Context) {
+	for _, ch := range w.state {
+		select {
+		case ch <- true:
+		default:
+			w.log.WithField("state", "started").Error("failed to send state")
+		}
 	}
 }
 
 var regHorizontalPodAutoscaler = regexp.MustCompile(`New size: (\d+); reason: (\w+).*(below|above) target`)
 
-func (w *Watcher) run(ctx context.Context, env string, client kubernetes.Interface) error {
+func (w *Watcher) run(ctx context.Context, env string, client kubernetes.Interface, state chan bool) error {
 	for {
-		if err := w.watch(ctx, env, client); err != nil {
-			if errors.Is(err, context.Canceled) {
-				return nil
+		select {
+		case <-ctx.Done():
+			return nil
+		case s := <-state:
+			w.log.WithField("env", env).WithField("state", s).Info("state change")
+			if s {
+				if err := w.watch(ctx, env, client, state); err != nil {
+					w.log.WithError(err).Error("failed to watch events")
+				}
+				w.log.WithField("env", env).Info("stopped watching")
+
 			}
-			w.log.WithError(err).Error("failed to watch events")
 		}
-		w.log.WithField("env", env).Info("stopped watching")
-		time.Sleep(2 * time.Second)
 	}
 }
 
-func (w *Watcher) watch(ctx context.Context, env string, client kubernetes.Interface) error {
+func (w *Watcher) watch(ctx context.Context, env string, client kubernetes.Interface, state chan bool) error {
 	// Events we want to watch for
 	// SuccessfulRescale - Check for successful rescale events
 	// Killing - Check for liveness failures
@@ -166,6 +169,10 @@ func (w *Watcher) watch(ctx context.Context, env string, client kubernetes.Inter
 		select {
 		case <-ctx.Done():
 			return nil
+		case s := <-state:
+			if !s {
+				return nil
+			}
 		case event := <-rescale.ResultChan():
 			handleEvent(event, func(e *eventv1.Event) (eventsql.UpsertParams, bool) {
 				if !strings.HasPrefix(e.Note, "New size") {