fix(kuma-cp): make store changes processing more reliable (backport of #6728)

pkg/events/eventbus.go

@@ -2,36 +2,45 @@ package events
 
 import (
 	"sync"
+
+	"github.com/kumahq/kuma/pkg/core"
 )
 
 func NewEventBus() *EventBus {
-	return &EventBus{}
+	return &EventBus{
+		subscribers: map[string]chan Event{},
+	}
 }
 
 type EventBus struct {
 	mtx         sync.RWMutex
-	subscribers []chan Event
+	subscribers map[string]chan Event
 }
 
-func (b *EventBus) New() Listener {
+func (b *EventBus) Subscribe() Listener {
+	id := core.NewUUID()
 	b.mtx.Lock()
 	defer b.mtx.Unlock()
 
 	events := make(chan Event, 10)
-	b.subscribers = append(b.subscribers, events)
+	b.subscribers[id] = events
 	return &reader{
 		events: events,
+		close: func() {
+			b.mtx.Lock()
+			defer b.mtx.Unlock()
+			delete(b.subscribers, id)
+		},
 	}
 }
 
 func (b *EventBus) Send(event Event) {
 	b.mtx.RLock()
 	defer b.mtx.RUnlock()
-
 	switch e := event.(type) {
 	case ResourceChangedEvent:
-		for _, s := range b.subscribers {
-			s <- ResourceChangedEvent{
+		for _, channel := range b.subscribers {
+			channel <- ResourceChangedEvent{
 				Operation: e.Operation,
 				Type:      e.Type,
 				Key:       e.Key,
@@ -42,8 +51,13 @@ func (b *EventBus) Send(event Event) {
 
 type reader struct {
 	events chan Event
+	close  func()
 }
 
 func (k *reader) Recv() <-chan Event {
 	return k.events
 }
+
+func (k *reader) Close() {
+	k.close()
+}

pkg/events/interfaces.go

@@ -26,12 +26,13 @@ var ListenerStoppedErr = errors.New("listener closed")
 
 type Listener interface {
 	Recv() <-chan Event
+	Close()
 }
 
 type Emitter interface {
 	Send(Event)
 }
 
 type ListenerFactory interface {
-	New() Listener
+	Subscribe() Listener
 }

pkg/insights/resyncer.go

@@ -133,7 +133,8 @@ func (r *resyncer) Start(stop <-chan struct{}) error {
 		}
 	}(stop)
 
-	eventReader := r.eventFactory.New()
+	eventReader := r.eventFactory.Subscribe()
+	defer eventReader.Close()
 	for {
 		select {
 		case <-stop:

pkg/insights/test/test_event_reader.go

@@ -10,10 +10,17 @@ func (t *TestEventReader) Recv() <-chan events.Event {
 	return t.Ch
 }
 
+// This method should be called only once. In tests, we
+// have only one call to subscribe, and we want to avoid
+// closing the channel twice, as it may lead to a panic.
+func (t *TestEventReader) Close() {
+	close(t.Ch)
+}
+
 type TestEventReaderFactory struct {
 	Reader *TestEventReader
 }
 
-func (t *TestEventReaderFactory) New() events.Listener {
+func (t *TestEventReaderFactory) Subscribe() events.Listener {
 	return t.Reader
 }

pkg/kds/global/components.go

@@ -20,6 +20,7 @@ import (
 	"github.com/kumahq/kuma/pkg/core/resources/registry"
 	"github.com/kumahq/kuma/pkg/core/resources/store"
 	"github.com/kumahq/kuma/pkg/core/runtime"
+	"github.com/kumahq/kuma/pkg/core/runtime/component"
 	"github.com/kumahq/kuma/pkg/core/user"
 	"github.com/kumahq/kuma/pkg/kds/client"
 	"github.com/kumahq/kuma/pkg/kds/mux"
@@ -140,7 +141,7 @@ func Setup(rt runtime.Runtime) error {
 			}
 		}()
 	})
-	return rt.Add(mux.NewServer(
+	return rt.Add(component.NewResilientComponent(kdsGlobalLog.WithName("kds-mux-client"), mux.NewServer(
 		onSessionStarted,
 		rt.KDSContext().GlobalServerFilters,
 		*rt.Config().Multizone.Global.KDS,
@@ -151,7 +152,7 @@ func Setup(rt runtime.Runtime) error {
 			onZoneToGlobalSyncConnect,
 			rt.KDSContext().GlobalServerFiltersV2,
 		),
-	))
+	)))
 }
 
 func createZoneIfAbsent(log logr.Logger, name string, resManager core_manager.ResourceManager) error {

pkg/plugins/common/postgres/listener.go

@@ -2,7 +2,7 @@ package postgres
 
 type Listener interface {
 	Notify() chan *Notification
-	Error() error
+	Error() <-chan error
 	Close() error
 }
 

pkg/plugins/common/postgres/pgx_listener.go

@@ -15,7 +15,7 @@ import (
 // PgxListener will listen for NOTIFY commands on a channel.
 type PgxListener struct {
 	notificationsCh chan *Notification
-	err             error
+	err             chan error
 	mu              sync.Mutex
 
 	logger logr.Logger
@@ -25,9 +25,7 @@ type PgxListener struct {
 	stopFn func()
 }
 
-func (l *PgxListener) Error() error {
-	l.mu.Lock()
-	defer l.mu.Unlock()
+func (l *PgxListener) Error() <-chan error {
 	return l.err
 }
 
@@ -46,6 +44,7 @@ func NewPgxListener(config postgres.PostgresStoreConfig, logger logr.Logger) (Li
 	}
 	l := &PgxListener{
 		notificationsCh: make(chan *Notification, 32),
+		err:             make(chan error),
 		logger:          logger,
 		db:              db,
 	}
@@ -73,8 +72,8 @@ func (l *PgxListener) run(ctx context.Context) {
 	if err != nil {
 		l.mu.Lock()
 		defer l.mu.Unlock()
-		l.err = err
 		close(l.notificationsCh)
+		l.err <- err
 	}
 }
 

pkg/plugins/common/postgres/pq_listener.go

@@ -1,8 +1,6 @@
 package postgres
 
 import (
-	"sync"
-
 	"github.com/go-logr/logr"
 	"github.com/lib/pq"
 
@@ -12,13 +10,11 @@ import (
 type pqListener struct {
 	listener      *pq.Listener
 	notifications chan *Notification
-	err           error
-	mu            sync.Mutex
+	err           chan error
+	stop          chan struct{}
 }
 
-func (p *pqListener) Error() error {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+func (p *pqListener) Error() <-chan error {
 	return p.err
 }
 
@@ -29,6 +25,7 @@ func (p *pqListener) Notify() chan *Notification {
 }
 
 func (p *pqListener) Close() error {
+	close(p.stop)
 	return p.listener.Close()
 }
 
@@ -39,15 +36,17 @@ func NewListener(cfg config.PostgresStoreConfig, log logr.Logger) (Listener, err
 	}
 
 	notificationCh := make(chan *Notification)
+	errCh := make(chan error)
+	stopCh := make(chan struct{})
 	l := &pqListener{
 		notifications: notificationCh,
+		err:           errCh,
+		stop:          stopCh,
 	}
 
 	reportProblem := func(ev pq.ListenerEventType, err error) {
 		if err != nil {
-			l.mu.Lock()
-			defer l.mu.Unlock()
-			l.err = err
+			l.err <- err
 			// notifications channel is already closed via Close()
 			return
 		}
@@ -63,12 +62,17 @@ func NewListener(cfg config.PostgresStoreConfig, log logr.Logger) (Listener, err
 
 	go func() {
 		for {
-			pqNotification, more := <-pqNotificationCh
-			if more {
-				notification := toNotification(pqNotification)
-				notificationCh <- notification
-			} else {
-				break
+			select {
+			case pqNotification, more := <-pqNotificationCh:
+				if more {
+					notification := toNotification(pqNotification)
+					notificationCh <- notification
+				} else {
+					return
+				}
+			case <-stopCh:
+				log.Info("stop")
+				return
 			}
 		}
 	}()

pkg/plugins/resources/postgres/events/listener.go

@@ -51,10 +51,10 @@ func (k *listener) Start(stop <-chan struct{}) error {
 	log.Info("start monitoring")
 	for {
 		select {
+		case err := <-listener.Error():
+			log.Error(err, "failed to listen on events")
+			return err
 		case n := <-listener.Notify():
-			if err := listener.Error(); err != nil {
-				return err
-			}
 			if n == nil {
 				continue
 			}

pkg/plugins/resources/postgres/listener_test.go

@@ -124,7 +124,7 @@ func setupStore(cfg postgres_config.PostgresStoreConfig, driverName string) stor
 func setupListeners(cfg postgres_config.PostgresStoreConfig, driverName string, listenerErrCh chan error, listenerStopCh chan struct{}) kuma_events.Listener {
 	cfg.DriverName = driverName
 	eventsBus := kuma_events.NewEventBus()
-	listener := eventsBus.New()
+	listener := eventsBus.Subscribe()
 	l := postgres_events.NewListener(cfg, eventsBus)
 	resilientListener := component.NewResilientComponent(core.Log.WithName("postgres-event-listener-component"), l)
 	go func() {