diff --git a/pkg/kube_events_manager/kube_events_manager.go b/pkg/kube_events_manager/kube_events_manager.go index 171218bc..383d803e 100644 --- a/pkg/kube_events_manager/kube_events_manager.go +++ b/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() diff --git a/pkg/kube_events_manager/monitor.go b/pkg/kube_events_manager/monitor.go index 91f6c58d..fcdb2ea5 100644 --- a/pkg/kube_events_manager/monitor.go +++ b/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() { diff --git a/pkg/kube_events_manager/resource_informer.go b/pkg/kube_events_manager/resource_informer.go index d27a0dc8..5a417768 100644 --- a/pkg/kube_events_manager/resource_informer.go +++ b/pkg/kube_events_manager/resource_informer.go @@ -6,6 +6,7 @@ import ( "log/slog" "runtime/trace" "sync" + "sync/atomic" "time" "github.com/deckhouse/deckhouse/pkg/log" @@ -13,6 +14,7 @@ import ( "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) diff --git a/test/hook/context/generator.go b/test/hook/context/generator.go index 254ebd86..a9338b15 100644 --- a/test/hook/context/generator.go +++ b/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) { diff --git a/test/hook/context/generator_test.go b/test/hook/context/generator_test.go index e52fa8e2..3df7320d 100644 --- a/test/hook/context/generator_test.go +++ b/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()) +} diff --git a/test/hook/context/state.go b/test/hook/context/state.go index d99a965d..ad263050 100644 --- a/test/hook/context/state.go +++ b/test/hook/context/state.go @@ -3,12 +3,9 @@ package context import ( "fmt" "reflect" - "time" "github.com/flant/kube-client/fake" "github.com/flant/kube-client/manifest" - kubeeventsmanager "github.com/flant/shell-operator/pkg/kube_events_manager" - "github.com/flant/shell-operator/pkg/kube_events_manager/types" ) // if we use default, then we are not able to emulate global resources due to fake cluster limitations @@ -19,19 +16,14 @@ var defaultNamespace = "" //nolint:gochecknoglobals type StateController struct { CurrentState map[string]manifest.Manifest fakeCluster *fake.Cluster - - stopCh chan<- types.KubeEvent } // NewStateController creates controller to apply state changes -func NewStateController(fc *fake.Cluster, ev kubeeventsmanager.KubeEventsManager) *StateController { - c := &StateController{ +func NewStateController(fc *fake.Cluster) *StateController { + return &StateController{ CurrentState: make(map[string]manifest.Manifest), fakeCluster: fc, - stopCh: ev.Ch(), } - - return c } func (c *StateController) SetInitialState(initialState string) error { @@ -94,10 +86,5 @@ func (c *StateController) ChangeState(newRawState string) error { } } - go func() { - time.Sleep(100 * time.Millisecond) - c.stopCh <- types.KubeEvent{MonitorId: "STOP_EVENTS"} - }() - return nil }