From 53d74816949e66f0c7b2acbdba787b952f4138e0 Mon Sep 17 00:00:00 2001 From: Yuriy Losev Date: Tue, 3 Nov 2026 15:13:14 +0400 Subject: [PATCH] [shell-operator] feat: let an embedding application transform objects before caching Add the WithObjectTransform option. The function is applied to every Kubernetes object before the shared informers cache it and the hooks filter it, so an embedding application can drop fields no hook reads, such as metadata.managedFields, and cut the informer cache size. The transform is set on every shared informer and also applied to the initial list in loadExistedObjects, so the first snapshot checksums match the later watch events. Without the option objects are kept as is. Signed-off-by: Yuriy Losev --- pkg/kube_events_manager/factory.go | 48 ++++++++++++++- pkg/kube_events_manager/factory_test.go | 60 +++++++++++++++++++ .../kube_events_manager.go | 7 +++ pkg/kube_events_manager/resource_informer.go | 10 ++-- pkg/shell-operator/bootstrap.go | 6 +- pkg/shell-operator/operator.go | 15 +++++ 6 files changed, 139 insertions(+), 7 deletions(-) diff --git a/pkg/kube_events_manager/factory.go b/pkg/kube_events_manager/factory.go index 7ec9f3c5..4c92638d 100644 --- a/pkg/kube_events_manager/factory.go +++ b/pkg/kube_events_manager/factory.go @@ -10,6 +10,7 @@ import ( "github.com/deckhouse/deckhouse/pkg/log" metricsstorage "github.com/deckhouse/deckhouse/pkg/metrics-storage" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/client-go/dynamic" @@ -54,6 +55,8 @@ type FactoryStore struct { stoppedCh map[FactoryIndex]chan struct{} // optional, only for the factory_informer_dead_total counter metricStorage metricsstorage.Storage + // optional, applied to every object before it is cached and filtered + transform cache.TransformFunc } func NewFactoryStore(ctx context.Context) *FactoryStore { @@ -77,6 +80,40 @@ func (c *FactoryStore) WithMetricStorage(mstor metricsstorage.Storage) { c.metricStorage = mstor } +// WithTransform sets a function applied to every object before the informers +// cache it and the hooks filter it, e.g. to drop fields no hook reads. It must be +// idempotent, as client-go may apply it to an object again. Set it before the +// first Start: informers that are already running keep their objects as is. +func (c *FactoryStore) WithTransform(fn cache.TransformFunc) { + c.mu.Lock() + defer c.mu.Unlock() + c.transform = fn +} + +// transformObject applies the transform to an object listed outside the informers, +// so its filter result and checksum match the ones of later watch events. +func (c *FactoryStore) transformObject(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) { + c.mu.Lock() + transform := c.transform + c.mu.Unlock() + + if transform == nil { + return obj, nil + } + + res, err := transform(obj) + if err != nil { + return nil, err + } + + transformed, ok := res.(*unstructured.Unstructured) + if !ok { + return nil, fmt.Errorf("transform returned %T instead of *unstructured.Unstructured", res) + } + + return transformed, nil +} + func (c *FactoryStore) Reset() { c.mu.Lock() defer c.mu.Unlock() @@ -125,7 +162,16 @@ func (c *FactoryStore) get(client dynamic.Interface, index FactoryIndex) *Factor factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory( client, resyncPeriod, index.Namespace, tweakListOptions) - factory.ForResource(index.GVR) + informer := factory.ForResource(index.GVR).Informer() + + if c.transform != nil { + // Fails only on a started informer, and this one is brand new. + if err := informer.SetTransform(c.transform); err != nil { + log.Warn("Factory store: transform wasn't set", + slog.String(pkg.LogKeyNamespace, index.Namespace), slog.String(pkg.LogKeyGVR, index.GVR.String()), + log.Err(err)) + } + } c.add(index, factory) diff --git a/pkg/kube_events_manager/factory_test.go b/pkg/kube_events_manager/factory_test.go index 217d2a51..0d43b4c0 100644 --- a/pkg/kube_events_manager/factory_test.go +++ b/pkg/kube_events_manager/factory_test.go @@ -11,6 +11,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" "github.com/flant/kube-client/fake" + kemtypes "github.com/flant/shell-operator/pkg/kube_events_manager/types" "github.com/flant/shell-operator/pkg/metric" "github.com/flant/shell-operator/pkg/metrics" ) @@ -156,3 +157,62 @@ func TestFactoryStore_RecreatesDeadFactory(t *testing.T) { "namespace": index.Namespace, })) } + +// TestFactoryStore_TransformsObjects checks that the store transform reaches both +// the initial list and the watch events, so hooks never see the dropped fields. +func TestFactoryStore_TransformsObjects(t *testing.T) { + g := NewWithT(t) + + fc := fake.NewFakeCluster(fake.ClusterVersionV121) + createNsWithLabels(fc, "default", map[string]string{}) + createCM(fc, "default", testCM("listed")) + + store := NewFactoryStore(context.Background()) + store.WithTransform(func(obj interface{}) (interface{}, error) { + if u, ok := obj.(*unstructured.Unstructured); ok { + unstructured.RemoveNestedField(u.Object, "data") + } + return obj, nil + }) + + metricStorage := metric.NewStorageMock(t) + metricStorage.HistogramObserveMock.Set(func(string, float64, map[string]string, []float64) {}) + metricStorage.GaugeSetMock.Set(func(string, float64, map[string]string) {}) + + monitorCfg := &MonitorConfig{ + ApiVersion: "v1", + Kind: "ConfigMap", + EventTypes: []kemtypes.WatchEventType{kemtypes.WatchEventAdded}, + NamespaceSelector: &kemtypes.NamespaceSelector{ + NameSelector: &kemtypes.NameSelector{MatchNames: []string{"default"}}, + }, + KeepFullObjectsInMemory: true, + } + + var mu sync.Mutex + var fromEvents []kemtypes.ObjectAndFilterResult + mon := NewMonitor(context.Background(), fc.Client, metricStorage, store, monitorCfg, func(ev kemtypes.KubeEvent) { + mu.Lock() + fromEvents = append(fromEvents, ev.Objects...) + mu.Unlock() + }, log.NewNop()) + + // CreateInformers lists existing objects itself, before the informers start. + g.Expect(mon.CreateInformers()).To(Succeed()) + snapshot := mon.Snapshot() + g.Expect(snapshot).To(HaveLen(1)) + g.Expect(snapshot[0].Object.Object).NotTo(HaveKey("data"), "initial list must be transformed") + + mon.Start(context.TODO()) + mon.EnableKubeEventCb() + createCM(fc, "default", testCM("watched")) + + g.Eventually(func() []kemtypes.ObjectAndFilterResult { + mu.Lock() + defer mu.Unlock() + return append([]kemtypes.ObjectAndFilterResult(nil), fromEvents...) + }, "5s", "10ms").Should(ContainElement(Satisfy(func(o kemtypes.ObjectAndFilterResult) bool { + _, hasData := o.Object.Object["data"] + return o.Object.GetName() == "watched" && !hasData + })), "watch events must be transformed") +} diff --git a/pkg/kube_events_manager/kube_events_manager.go b/pkg/kube_events_manager/kube_events_manager.go index 383d803e..1d5a80cd 100644 --- a/pkg/kube_events_manager/kube_events_manager.go +++ b/pkg/kube_events_manager/kube_events_manager.go @@ -9,6 +9,7 @@ import ( "github.com/deckhouse/deckhouse/pkg/log" metricsstorage "github.com/deckhouse/deckhouse/pkg/metrics-storage" + "k8s.io/client-go/tools/cache" klient "github.com/flant/kube-client/client" "github.com/flant/shell-operator/pkg" @@ -89,6 +90,12 @@ func (mgr *kubeEventsManager) WithMetricStorage(mstor metricsstorage.Storage) { mgr.factoryStore.WithMetricStorage(mstor) } +// WithObjectTransform sets a function applied to every object before the +// informers cache it and the hooks filter it. Call it before adding monitors. +func (mgr *kubeEventsManager) WithObjectTransform(fn cache.TransformFunc) { + mgr.factoryStore.WithTransform(fn) +} + // AddMonitor creates a monitor with informers and return a KubeEvent with existing objects. // TODO cleanup informers in case of error // TODO use Context to stop informers diff --git a/pkg/kube_events_manager/resource_informer.go b/pkg/kube_events_manager/resource_informer.go index 5a417768..6f7cf8f0 100644 --- a/pkg/kube_events_manager/resource_informer.go +++ b/pkg/kube_events_manager/resource_informer.go @@ -227,16 +227,18 @@ func (ei *resourceInformer) loadExistedObjects() error { filteredObjects := make(map[string]*kemtypes.ObjectAndFilterResult) for _, item := range objList.Items { - // copy loop var to avoid duplication of pointer in filteredObjects - obj := item + // item is a per-iteration variable, so each pointer in filteredObjects is unique + obj, err := ei.factoryStore.transformObject(&item) + if err != nil { + return fmt.Errorf("transform '%s/%s': %w", ei.Monitor.Metadata.DebugName, item.GetName(), err) + } var objFilterRes *kemtypes.ObjectAndFilterResult - var err error func() { defer measure.Duration(func(d time.Duration) { ei.metricStorage.HistogramObserve(metrics.KubeJqFilterDurationSeconds, d.Seconds(), ei.Monitor.Metadata.MetricLabels, nil) })() - objFilterRes, err = applyFilter(ei.Monitor.CompiledJqFilter, ei.Monitor.JqFilter, ei.Monitor.FilterFunc, &obj) + objFilterRes, err = applyFilter(ei.Monitor.CompiledJqFilter, ei.Monitor.JqFilter, ei.Monitor.FilterFunc, obj) }() if err != nil { diff --git a/pkg/shell-operator/bootstrap.go b/pkg/shell-operator/bootstrap.go index 3cc3139d..93fe5398 100644 --- a/pkg/shell-operator/bootstrap.go +++ b/pkg/shell-operator/bootstrap.go @@ -282,8 +282,10 @@ func (op *ShellOperator) SetupEventManagers() { op.ScheduleManager = schedulemanager.NewScheduleManager(op.ctx, op.logger.Named("schedule-manager")) // Initialize kubernetes events manager. - op.KubeEventsManager = kubeeventsmanager.NewKubeEventsManager(op.ctx, op.KubeClient, op.logger.Named("kube-events-manager")) - op.KubeEventsManager.WithMetricStorage(op.MetricStorage) + kubeEventsManager := kubeeventsmanager.NewKubeEventsManager(op.ctx, op.KubeClient, op.logger.Named("kube-events-manager")) + kubeEventsManager.WithMetricStorage(op.MetricStorage) + kubeEventsManager.WithObjectTransform(op.objectTransform) + op.KubeEventsManager = kubeEventsManager // Initialize events handler that emit tasks to run hooks cfg := &managerEventsHandlerConfig{ diff --git a/pkg/shell-operator/operator.go b/pkg/shell-operator/operator.go index 7ca4a302..04940d4b 100644 --- a/pkg/shell-operator/operator.go +++ b/pkg/shell-operator/operator.go @@ -26,6 +26,7 @@ import ( metricsstorage "github.com/deckhouse/deckhouse/pkg/metrics-storage" "github.com/gofrs/uuid/v5" "go.opentelemetry.io/otel" + "k8s.io/client-go/tools/cache" klient "github.com/flant/kube-client/client" pkg "github.com/flant/shell-operator/pkg" @@ -100,6 +101,10 @@ type ShellOperator struct { // hookRunner runs hooks for events; nil starts each hook as a process. hookRunner executor.Runner + // objectTransform is applied to every Kubernetes object before it is + // cached and filtered; nil keeps objects as is. + objectTransform cache.TransformFunc + // MetricNames is a per-instance snapshot of resolved metric names. // Library callers may consult it to avoid relying on the package-level // metrics globals. NewShellOperator populates it from cfg's prefix; the @@ -157,6 +162,16 @@ func WithHookMetricStorage(ms metricsstorage.Storage) Option { } } +// WithObjectTransform sets a function applied to every Kubernetes object before +// the informers cache it and the hooks filter it. Use it to drop fields no hook +// reads, e.g. metadata.managedFields, to save memory. The function must be +// idempotent. By default objects are kept as is. +func WithObjectTransform(fn cache.TransformFunc) Option { + return func(operator *ShellOperator) { + operator.objectTransform = fn + } +} + // NewBareShellOperator constructs an empty *ShellOperator without wiring any // of the heavy components (HookManager, webhook managers, debug server, etc.). // It is intended for tests and downstream tooling that need to assemble a