[shell-operator] fix: wait for kube events to finish processing

pkg/kube_events_manager/kube_events_manager.go

@@ -169,6 +169,20 @@ func (mgr *kubeEventsManager) Ch() chan kemtypes.KubeEvent {
 	return mgr.KubeEventCh
 }
 
+// Synced reports whether all monitors have handled every watch event emitted so far.
+// It lists objects on each call and is meant for the fake cluster in tests,
+// so it is not a part of the KubeEventsManager interface.
+func (mgr *kubeEventsManager) Synced(ctx context.Context) bool {
+	mgr.m.RLock()
+	defer mgr.m.RUnlock()
+	for _, mon := range mgr.Monitors {
+		if m, ok := mon.(*monitor); ok && !m.synced(ctx) {
+			return false
+		}
+	}
+	return true
+}
+
 // Stop the kube events manager and all the informers inside monitors.
 func (mgr *kubeEventsManager) Stop() {
 	mgr.cancel()

pkg/kube_events_manager/monitor.go

@@ -296,6 +296,20 @@ func (m *monitor) Snapshot() []kemtypes.ObjectAndFilterResult {
 	return objects
 }
 
+// synced reports whether all informers have handled every watch event emitted so far.
+func (m *monitor) synced(ctx context.Context) bool {
+	synced := true
+	for _, informer := range m.ResourceInformers {
+		synced = synced && informer.synced(ctx)
+	}
+	m.VaryingInformers.RangeValue(func(value []*resourceInformer) {
+		for _, informer := range value {
+			synced = synced && informer.synced(ctx)
+		}
+	})
+	return synced
+}
+
 // EnableKubeEventCb allows execution of event callback for all informers.
 // Also executes eventCb for events accumulated during "Synchronization" phase.
 func (m *monitor) EnableKubeEventCb() {

pkg/kube_events_manager/resource_informer.go

@@ -6,13 +6,15 @@ import (
 	"log/slog"
 	"runtime/trace"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 	metricsstorage "github.com/deckhouse/deckhouse/pkg/metrics-storage"
 	"github.com/gofrs/uuid/v5"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/cache"
 
@@ -65,6 +67,9 @@ type resourceInformer struct {
 	// a flag to stop handle events after Stop()
 	stopped bool
 
+	// Number of watch events being handled right now, see synced.
+	inFlight atomic.Int32
+
 	logger *log.Logger
 }
 
@@ -280,6 +285,9 @@ func (ei *resourceInformer) OnDelete(obj interface{}) {
 // TODO add delay to merge Added and Modified events (node added and then labels applied — one hook run on Added+Modified is enough)
 // func (ei *resourceInformer) HandleKubeEvent(obj *unstructured.Unstructured, objectId string, filterResult string, newChecksum string, eventType WatchEventType) {
 func (ei *resourceInformer) handleWatchEvent(object interface{}, eventType kemtypes.WatchEventType) {
+	ei.inFlight.Add(1)
+	defer ei.inFlight.Add(-1)
+
 	// check if stop
 	if ei.stopped {
 		log.Debug("received WATCH for stopped informer",
@@ -317,7 +325,20 @@ func (ei *resourceInformer) handleWatchEvent(object interface{}, eventType kemty
 			slog.String(pkg.LogKeyDebugName, ei.Monitor.Metadata.DebugName),
 			slog.String(pkg.LogKeyEventType, string(eventType)),
 			log.Err(err))
-		return
+		if eventType != kemtypes.WatchEventDeleted {
+			return
+		}
+		// A deleted object must leave the cache even if the filter fails on its last state:
+		// reuse the cached filter result.
+		ei.cacheLock.RLock()
+		cached, ok := ei.cachedObjects[resourceId]
+		ei.cacheLock.RUnlock()
+		if !ok {
+			return
+		}
+		res := *cached
+		res.Object = obj
+		objFilterRes = &res
 	}
 
 	if !ei.Monitor.KeepFullObjectsInMemory {
@@ -472,6 +493,56 @@ func (ei *resourceInformer) start() {
 	log.Debug("informer is ready", slog.String(pkg.LogKeyDebugName, ei.Monitor.Metadata.DebugName))
 }
 
+// synced reports whether the informer has handled every watch event emitted so far:
+// the cache matches the objects in the cluster and no event is being handled.
+// It lists objects on each call and is meant for the fake cluster in tests.
+// The fake cluster ignores selectors in watches, so objects are listed without them.
+func (ei *resourceInformer) synced(ctx context.Context) bool {
+	if ei.ctx == nil || ei.ctx.Err() != nil {
+		// Not started or stopped: no events to wait for.
+		return true
+	}
+	if ei.inFlight.Load() > 0 {
+		return false
+	}
+	selector, err := labels.Parse(ei.ListOptions.LabelSelector)
+	if err != nil {
+		return true
+	}
+	list, err := ei.KubeClient.Dynamic().Resource(ei.GroupVersionResource).Namespace(ei.Namespace).List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return true
+	}
+
+	ei.cacheLock.RLock()
+	defer ei.cacheLock.RUnlock()
+	found := 0
+	for i := range list.Items {
+		obj := &list.Items[i]
+		cached, ok := ei.cachedObjects[resourceId(obj)]
+		if !ok && !selector.Matches(labels.Set(obj.GetLabels())) {
+			// Objects outside the label selector are cached only after their first watch event.
+			continue
+		}
+		if ok {
+			found++
+		}
+		res, err := applyFilter(ei.Monitor.CompiledJqFilter, ei.Monitor.JqFilter, ei.Monitor.FilterFunc, obj)
+		if err != nil {
+			// handleWatchEvent drops objects the filter fails on: never cached, or cached in an old state.
+			continue
+		}
+		if !ok || res.Metadata.Checksum != cached.Metadata.Checksum {
+			// Pending Added or Modified event.
+			return false
+		}
+	}
+
+	// A cached object missing from the cluster is a pending Deleted event.
+	// Check inFlight after the cache: the cache is updated before the event is sent.
+	return found == len(ei.cachedObjects) && ei.inFlight.Load() == 0
+}
+
 // wait blocks until the underlying shared informer for this FactoryIndex is stopped
 func (ei *resourceInformer) wait() {
 	ei.factoryStore.WaitStopped(ei.FactoryIndex)

test/hook/context/generator.go

@@ -71,7 +71,7 @@ func NewBindingContextController(config string, logger *log.Logger, version ...f
 
 	b.ScheduleManager = schedulemanager.NewScheduleManager(ctx, b.logger.Named("schedule-manager"))
 
-	b.Controller = NewStateController(fc, b.KubeEventsManager)
+	b.Controller = NewStateController(fc)
 
 	return b
 }
@@ -140,6 +140,12 @@ func (b *BindingContextController) ChangeState(newState string) (GeneratedBindin
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	// Synced is not a part of the KubeEventsManager interface: it is meant for the fake cluster only.
+	mgr, ok := b.KubeEventsManager.(interface{ Synced(context.Context) bool })
+	if !ok {
+		return GeneratedBindingContexts{}, fmt.Errorf("KubeEventsManager %T cannot report handled events", b.KubeEventsManager)
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
 	defer cancel()
 
@@ -149,23 +155,23 @@ func (b *BindingContextController) ChangeState(newState string) (GeneratedBindin
 		return GeneratedBindingContexts{}, fmt.Errorf("error while changing BindingContextGenerator state: %v", err)
 	}
 
-outer:
+	// Collect events until informers have handled every change and the last sent event is received.
+	tick := time.NewTicker(10 * time.Millisecond)
+	defer tick.Stop()
 	for {
 		select {
 		case ev := <-b.KubeEventsManager.Ch():
-			switch ev.MonitorId {
-			case "STOP_EVENTS":
-				break outer
-			default:
-				b.HookCtrl.HandleKubeEvent(context.TODO(), ev, func(info controller.BindingExecutionInfo) {
-					cc.AddBindingContext(types.OnKubernetesEvent, info)
-				})
+			b.HookCtrl.HandleKubeEvent(context.TODO(), ev, func(info controller.BindingExecutionInfo) {
+				cc.AddBindingContext(types.OnKubernetesEvent, info)
+			})
+		case <-tick.C:
+			if mgr.Synced(ctx) && len(b.KubeEventsManager.Ch()) == 0 {
+				return cc.CombinedAndUpdated(b.HookCtrl)
 			}
 		case <-ctx.Done():
 			return GeneratedBindingContexts{}, fmt.Errorf("timeout occurred while waiting for binding contexts")
 		}
 	}
-	return cc.CombinedAndUpdated(b.HookCtrl)
 }
 
 func (b *BindingContextController) RunSchedule(ctx context.Context, crontab string) (GeneratedBindingContexts, error) {

test/hook/context/generator_test.go

@@ -4,10 +4,13 @@ import (
 	"context"
 	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 
+	"github.com/flant/shell-operator/pkg/hook"
 	bctx "github.com/flant/shell-operator/pkg/hook/binding_context"
 )
 
@@ -476,3 +479,136 @@ metadata:
 	parsedBindingContexts = parseContexts(contexts.Rendered)
 	g.Expect(parsedBindingContexts[0].Snapshots["selected_pods"]).To(HaveLen(2))
 }
+
+// A slow filter stands in for a loaded CI runner: events must not be cut off by a timer.
+func Test_ChangeState_SlowEvents(t *testing.T) {
+	g := NewWithT(t)
+
+	h, err := hook.NewHook("test", "test", false, false, "", log.NewNop()).LoadConfig([]byte(`
+configVersion: v1
+kubernetes:
+- apiVersion: v1
+  kind: Pod
+  name: pods
+  includeSnapshotsFrom: [pods]
+`))
+	g.Expect(err).ShouldNot(HaveOccurred())
+	h.GetConfig().OnKubernetesEvents[0].Monitor.FilterFunc = func(obj *unstructured.Unstructured) (interface{}, error) {
+		time.Sleep(50 * time.Millisecond)
+		return obj.GetName(), nil
+	}
+
+	c := NewBindingContextController("", log.NewNop())
+	c.WithHook(h)
+	defer c.Stop()
+
+	_, err = c.Run(``)
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	contexts, err := c.ChangeState(`
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: pod1
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: pod2
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: pod3
+`)
+	g.Expect(err).ShouldNot(HaveOccurred())
+	parsedBindingContexts := parseContexts(contexts.Rendered)
+	g.Expect(parsedBindingContexts).To(HaveLen(3))
+	g.Expect(parsedBindingContexts[2].Snapshots["pods"]).To(HaveLen(3))
+
+	contexts, err = c.ChangeState(``)
+	g.Expect(err).ShouldNot(HaveOccurred())
+	parsedBindingContexts = parseContexts(contexts.Rendered)
+	g.Expect(parsedBindingContexts).To(HaveLen(3))
+	g.Expect(parsedBindingContexts[2].Snapshots["pods"]).To(BeEmpty())
+}
+
+// Objects the filter fails on are never cached, ChangeState must not wait for them.
+func Test_ChangeState_FilterError(t *testing.T) {
+	g := NewWithT(t)
+
+	c := NewBindingContextController(`
+configVersion: v1
+kubernetes:
+- apiVersion: v1
+  kind: Pod
+  name: pods
+  includeSnapshotsFrom: [pods]
+  jqFilter: 'if .metadata.name == "bad" then error("boom") else .metadata.name end'
+`, log.NewNop())
+	defer c.Stop()
+
+	_, err := c.Run(``)
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	contexts, err := c.ChangeState(`
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: bad
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: good
+`)
+	g.Expect(err).ShouldNot(HaveOccurred())
+	parsedBindingContexts := parseContexts(contexts.Rendered)
+	g.Expect(parsedBindingContexts).To(HaveLen(1))
+	g.Expect(parsedBindingContexts[0].Snapshots["pods"]).To(HaveLen(1))
+}
+
+// A cached object must leave the snapshot on delete even if the filter fails on its last state.
+func Test_ChangeState_FilterErrorOnDelete(t *testing.T) {
+	g := NewWithT(t)
+
+	c := NewBindingContextController(`
+configVersion: v1
+kubernetes:
+- apiVersion: v1
+  kind: Pod
+  name: pods
+  includeSnapshotsFrom: [pods]
+  jqFilter: 'if .metadata.labels.state == "bad" then error("boom") else .metadata.name end'
+`, log.NewNop())
+	defer c.Stop()
+
+	_, err := c.Run(`
+apiVersion: v1
+kind: Pod
+metadata:
+  name: p1
+  labels:
+    state: good
+`)
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	_, err = c.ChangeState(`
+apiVersion: v1
+kind: Pod
+metadata:
+  name: p1
+  labels:
+    state: bad
+`)
+	g.Expect(err).ShouldNot(HaveOccurred())
+
+	contexts, err := c.ChangeState(``)
+	g.Expect(err).ShouldNot(HaveOccurred())
+	parsedBindingContexts := parseContexts(contexts.Rendered)
+	g.Expect(parsedBindingContexts).To(HaveLen(1))
+	g.Expect(parsedBindingContexts[0].WatchEvent).To(BeEquivalentTo("Deleted"))
+	g.Expect(parsedBindingContexts[0].Snapshots["pods"]).To(BeEmpty())
+}