[shell-operator] Let an embedding application transform objects before caching

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)
 

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")
+}

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

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 {

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{

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