[shell-operator] add dedup client

go.mod

@@ -41,10 +41,12 @@ require (
 	github.com/goccy/go-json v0.10.6
 	github.com/gojuno/minimock/v3 v3.4.7
 	github.com/itchyny/gojq v0.12.17
+	github.com/ldmonster/kubeclient v0.0.0-20260522082709-ed73652c723f
 	github.com/muesli/termenv v0.16.0
 	github.com/onsi/ginkgo/v2 v2.27.5
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/pflag v1.0.9
+	sigs.k8s.io/controller-runtime v0.20.4
 )
 
 require (
@@ -133,7 +135,6 @@ require (
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20250318190949-c8a335a9a2ff // indirect
 	k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 // indirect
-	sigs.k8s.io/controller-runtime v0.20.4 // indirect
 	sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect
 	sigs.k8s.io/kustomize/api v0.19.0 // indirect
 	sigs.k8s.io/kustomize/kyaml v0.19.0 // indirect

go.sum

@@ -223,6 +223,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/ldmonster/kubeclient v0.0.0-20260522082709-ed73652c723f h1:hTyM8+nWGxBczLaa0HzjXbKJuMjdbZQa9ZB2F0wdO04=
+github.com/ldmonster/kubeclient v0.0.0-20260522082709-ed73652c723f/go.mod h1:Q8GOTVz5hMCvWJjTmeLRQ79yp+AkX76yuNL/R66gybk=
 github.com/lucasb-eyer/go-colorful v1.2.0 h1:1nnpGOrhyZZuNyfu1QjKiUICQ74+3FNCN69Aj6K7nkY=
 github.com/lucasb-eyer/go-colorful v1.2.0/go.mod h1:R4dSotOR9KMtayYi1e77YzuveK+i7ruzyGqttikkLy0=
 github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=

pkg/app/README.md

@@ -315,11 +315,16 @@ err := so.AssembleCommonOperatorFromConfig(cfg, []string{
 
 The new `AssembleCommonOperatorFromConfig(cfg, labels)` method on
 `*ShellOperator` is what makes this clean — it derives both
-`KubeClientConfig`s (main + object-patcher), the HTTP listen address/port,
-and the metric prefix from `cfg`, so the consumer does not have to unpack
-fields by hand. The older primitive-taking
+`KubeClientConfig`s (main + object-patcher), the deduplicated-kubeclient
+configuration (`DedupClientConfig`), the HTTP listen address/port, and the
+metric prefix from `cfg`, so the consumer does not have to unpack fields by
+hand. The older primitive-taking
 `AssembleCommonOperator(listenAddress, listenPort, labels, mainKubeCfg, patcherKubeCfg)`
-is still available for callers that need finer control.
+is still available for callers that need finer control; it preserves its
+original signature and disables the dedup cache. To enable the dedup cache
+without going through `*app.Config`, use the new
+`AssembleCommonOperatorWithDedupClient(...)` variant which adds a final
+`DedupClientConfig` parameter.
 
 If you also want env-var parsing on top of your own values, call
 `ParseEnv(cfg)` between steps 1 and 2 — env values will overlay the fields you

pkg/app/app_config.go

@@ -34,6 +34,32 @@ type ObjectPatcherSettings struct {
 	KubeClientTimeout time.Duration `env:"KUBE_CLIENT_TIMEOUT"`
 }
 
+// DedupClientSettings configures the deduplicated kubeclient cache provided by
+// github.com/ldmonster/kubeclient. The cache stores a single canonical copy of
+// each repeated value and subtree across watched objects, dramatically lowering
+// in-memory footprint for clusters with many similar resources (e.g.
+// templated Deployments). All settings here are optional; when Enabled is
+// false the client is not constructed at all. List-typed env vars use a comma
+// separator: GVK strings follow the form "<group>/<version>/<kind>" (the
+// group may be empty for core resources, e.g. "/v1/Pod").
+type DedupClientSettings struct {
+	Enabled            bool          `env:"ENABLED"`
+	Namespaces         []string      `env:"NAMESPACES" envSeparator:","`
+	WatchGVKs          []string      `env:"WATCH_GVKS" envSeparator:","`
+	ReconstructLRUSize int           `env:"RECONSTRUCT_LRU_SIZE"`
+	GCInterval         time.Duration `env:"GC_INTERVAL"`
+
+	// SnapshotStore enables a process-wide deduplicated SnapshotStore that
+	// backs every kubernetes-binding monitor's per-object cache. When on,
+	// `*Unstructured` bodies live exactly once in memory across all
+	// resourceInformers (refcounted), trading a small per-snapshot-read CPU
+	// cost for a substantial drop in RSS for workloads with many similar
+	// objects. Independent of the runtime DedupClient (Enabled flag): the
+	// snapshot store can be turned on without spinning up any kubeclient
+	// informers.
+	SnapshotStore bool `env:"SNAPSHOT_STORE"`
+}
+
 // AdmissionSettings holds settings for the validating-webhook server.
 type AdmissionSettings struct {
 	ConfigurationName string   `env:"CONFIGURATION_NAME"`
@@ -83,6 +109,7 @@ type Config struct {
 	App           AppSettings           `envPrefix:"SHELL_OPERATOR_"`
 	Kube          KubeSettings          `envPrefix:"KUBE_"`
 	ObjectPatcher ObjectPatcherSettings `envPrefix:"OBJECT_PATCHER_"`
+	DedupClient   DedupClientSettings   `envPrefix:"DEDUP_CLIENT_"`
 	Admission     AdmissionSettings     `envPrefix:"VALIDATING_WEBHOOK_"`
 	Conversion    ConversionSettings    `envPrefix:"CONVERSION_WEBHOOK_"`
 	Debug         DebugSettings         `envPrefix:"DEBUG_"`

pkg/app/flags.go

@@ -22,11 +22,13 @@ func BindFlags(cfg *Config, rootCmd *cobra.Command, cmd *cobra.Command) func() {
 	bindLogFlags(cfg, cmd)
 	applyAdmission := bindAdmissionWebhookFlags(cfg, cmd)
 	applyConversion := bindConversionWebhookFlags(cfg, cmd)
+	applyDedup := bindDedupClientFlags(cfg, cmd)
 	bindDebugFlags(cfg, rootCmd, cmd)
 
 	return func() {
 		applyAdmission()
 		applyConversion()
+		applyDedup()
 	}
 }
 
@@ -106,6 +108,55 @@ func bindConversionWebhookFlags(cfg *Config, cmd *cobra.Command) func() {
 	}
 }
 
+// bindDedupClientFlags registers flags for the deduplicated kubeclient cache.
+// The two []string fields (Namespaces and WatchGVKs) follow the same pattern
+// used by the validating-webhook ClientCA flag: any explicit CLI invocation
+// fully replaces the env-var derived slice; otherwise the env value is kept.
+func bindDedupClientFlags(cfg *Config, cmd *cobra.Command) func() {
+	f := cmd.Flags()
+	f.BoolVar(&cfg.DedupClient.Enabled, "dedup-client-enabled", cfg.DedupClient.Enabled,
+		"Enable the deduplicated kubeclient cache (github.com/ldmonster/kubeclient). "+
+			"When set, shell-operator builds a controller-runtime compatible client backed "+
+			"by a deduplicated store. Can be set with $DEDUP_CLIENT_ENABLED.")
+	f.BoolVar(&cfg.DedupClient.SnapshotStore, "dedup-client-snapshot-store", cfg.DedupClient.SnapshotStore,
+		"Back per-monitor object snapshots with a process-wide deduplicated store "+
+			"(github.com/ldmonster/kubeclient/store). Trades a small per-snapshot-read CPU "+
+			"cost for a substantial drop in RSS when many monitors observe similar objects. "+
+			"Independent of --dedup-client-enabled. Can be set with $DEDUP_CLIENT_SNAPSHOT_STORE.")
+	f.IntVar(&cfg.DedupClient.ReconstructLRUSize, "dedup-client-reconstruct-lru-size",
+		cfg.DedupClient.ReconstructLRUSize,
+		"Size of the LRU that memoises reconstructed Unstructured objects in the dedup cache. "+
+			"Zero disables reconstruction caching. Can be set with $DEDUP_CLIENT_RECONSTRUCT_LRU_SIZE.")
+	f.DurationVar(&cfg.DedupClient.GCInterval, "dedup-client-gc-interval",
+		cfg.DedupClient.GCInterval,
+		"How often the deduplicated store reclaims unused interned values and subtrees. "+
+			"Zero leaves the kubeclient default in place. Can be set with $DEDUP_CLIENT_GC_INTERVAL.")
+
+	envNamespaces := cfg.DedupClient.Namespaces
+	envGVKs := cfg.DedupClient.WatchGVKs
+	var cliNamespaces, cliGVKs []string
+	f.StringArrayVar(&cliNamespaces, "dedup-client-namespace", nil,
+		"Namespace to restrict the dedup cache to. Repeat the flag to add more, or pass a "+
+			"comma-separated list via $DEDUP_CLIENT_NAMESPACES. Empty means all namespaces.")
+	f.StringArrayVar(&cliGVKs, "dedup-client-watch-gvk", nil,
+		"GroupVersionKind to pre-register with the dedup cache, formatted as "+
+			"\"<group>/<version>/<kind>\" (the group is empty for core resources, e.g. \"/v1/Pod\"). "+
+			"Repeat the flag to add more, or pass a comma-separated list via $DEDUP_CLIENT_WATCH_GVKS.")
+
+	return func() {
+		if len(cliNamespaces) > 0 {
+			cfg.DedupClient.Namespaces = cliNamespaces
+		} else {
+			cfg.DedupClient.Namespaces = envNamespaces
+		}
+		if len(cliGVKs) > 0 {
+			cfg.DedupClient.WatchGVKs = cliGVKs
+		} else {
+			cfg.DedupClient.WatchGVKs = envGVKs
+		}
+	}
+}
+
 func bindLogFlags(cfg *Config, cmd *cobra.Command) {
 	f := cmd.Flags()
 	f.StringVar(&cfg.Log.Level, "log-level", cfg.Log.Level, "Logging level: debug, info, error. Default is info. Can be set with $LOG_LEVEL.")

pkg/kube/dedupclient/README.md

@@ -0,0 +1,131 @@
+# pkg/kube/dedupclient
+
+Two-part integration of
+[`github.com/ldmonster/kubeclient`](https://github.com/ldmonster/kubeclient)
+into shell-operator. Both parts can be enabled independently — the one that
+moves the RSS needle for typical workloads is the **SnapshotStore**.
+
+| Component        | Type                       | Purpose                                                                                                  | Flag                                |
+| ---------------- | -------------------------- | -------------------------------------------------------------------------------------------------------- | ----------------------------------- |
+| `Client`         | `*kubeclient.DedupClient` wrapper | Controller-runtime compatible Kubernetes client for hooks/extensions, with its own deduplicated cache. | `--dedup-client-enabled`            |
+| `SnapshotStore`  | `*store.DedupStore` wrapper       | Process-wide, reference-counted cache that backs every kube-events-manager monitor's per-object snapshot. **This is what reduces memory.** | `--dedup-client-snapshot-store`     |
+
+For clusters with thousands of similar resources (e.g. templated
+`Deployment`s) the upstream store reports **60–90 %** lower cache memory
+usage thanks to value interning and subtree deduplication.
+
+## Quick start
+
+```go
+import (
+    klient "github.com/flant/kube-client/client"
+    "github.com/flant/shell-operator/pkg/kube/dedupclient"
+)
+
+func newDedup(main *klient.Client, logger *log.Logger) (*dedupclient.Client, error) {
+    mapper, _ := main.ToRESTMapper()
+    return dedupclient.New(dedupclient.Config{
+        RESTConfig: main.RestConfig(),
+        RESTMapper: mapper,
+        Namespaces: []string{"kube-system", "default"}, // empty = all
+        WatchGVKs: []schema.GroupVersionKind{
+            {Group: "", Version: "v1", Kind: "Pod"},
+            {Group: "apps", Version: "v1", Kind: "Deployment"},
+        },
+        ReconstructLRUSize: 4096, // 0 disables reconstruction caching
+    }, logger)
+}
+```
+
+`Start(ctx)` spins up the cache run loop in a single dedicated goroutine and
+returns immediately. `Shutdown(ctx)` cancels the loop and waits for it to
+exit (or for `ctx` to expire).
+
+## How shell-operator wires it up
+
+When `app.Config.DedupClient.Enabled` is `true`,
+`AssembleCommonOperatorFromConfig` calls `initDedupClient`, which hands the
+main `klient.Client`'s `rest.Config` and `RESTMapper` to
+`dedupclient.New`. The resulting `*Client` is stored on
+`shell_operator.ShellOperator.DedupClient`, started during `op.Start()`, and
+stopped from `op.Shutdown()`.
+
+Configuration knobs (env vars / CLI flags):
+
+| Env var                              | Flag                                 | Meaning                                                      |
+| ------------------------------------ | ------------------------------------ | ------------------------------------------------------------ |
+| `DEDUP_CLIENT_ENABLED`               | `--dedup-client-enabled`             | Construct the deduplicated client at all.                    |
+| `DEDUP_CLIENT_SNAPSHOT_STORE`        | `--dedup-client-snapshot-store`      | Back per-monitor snapshots with the shared dedup store. Independent of `--dedup-client-enabled`. |
+| `DEDUP_CLIENT_NAMESPACES`            | `--dedup-client-namespace`           | Comma-separated (env) or repeated (flag) namespace allow-list. Empty = all. |
+| `DEDUP_CLIENT_WATCH_GVKS`            | `--dedup-client-watch-gvk`           | GVKs to pre-register, formatted as `<group>/<version>/<kind>` (group empty for core). |
+| `DEDUP_CLIENT_RECONSTRUCT_LRU_SIZE`  | `--dedup-client-reconstruct-lru-size`| LRU size for reconstructed Unstructured objects. 0 disables. |
+| `DEDUP_CLIENT_GC_INTERVAL`           | `--dedup-client-gc-interval`         | GC interval for unused interned values/subtrees. 0 = upstream default. |
+
+When both features are off the wrappers are **not** constructed at all, so
+this integration adds zero runtime overhead to existing deployments.
+
+## SnapshotStore — the memory win
+
+`SnapshotStore` plugs into shell-operator's `pkg/kube_events_manager` so that
+every monitor's `cachedObjects` map stops holding `*Unstructured` pointers
+and instead stores `(resourceId → store.ObjectKey)` references into a
+process-wide, reference-counted dedup store.
+
+What changes when the flag is on (`MonitorConfig.KeepFullObjectsInMemory == true`):
+
+- Each `resourceInformer` calls `SnapshotStore.Acquire(ownerID, key, obj)` on
+  initial-list and on Add/Modified events. The store de-duplicates field
+  values and subtrees across every object it holds.
+- The per-monitor `*ObjectAndFilterResult` keeps `Object == nil`; the
+  authoritative body lives once in the store.
+- `monitor.Snapshot()` reconstitutes `Object` lazily by calling
+  `SnapshotStore.Get(key)`. Reconstitution is a fresh allocation per call,
+  which trades a small CPU cost for the memory drop.
+- On informer shutdown, all keys held by that informer are released. The
+  underlying object is removed from the store only when the last owner
+  releases it, so overlapping watches are correctly handled.
+
+When `KeepFullObjectsInMemory == false`, the existing "no full body kept"
+path takes precedence and the store is bypassed for that monitor — there is
+no benefit to deduplicating bodies you've already chosen to discard.
+
+### When does it actually save memory?
+
+The win scales with two factors:
+
+1. **Cross-factory duplication.** Each unique
+   `(GVR, namespace, fieldSelector, labelSelector)` gets its own client-go
+   informer cache today. When several monitors observe overlapping object
+   sets through *different* selectors, every cache holds its own copy. Once
+   `SnapshotStore` is on, the bodies converge to a single deduplicated copy
+   regardless of how many monitors observe them.
+2. **Intra-object subtree duplication.** Even within a single GVR, similar
+   objects share substantial structure — e.g. a thousand Pods generated
+   from one template share `securityContext`, `tolerations`, `resources`,
+   and most label/annotation keys. Value interning + subtree dedup encode
+   those shared parts once.
+
+If your hooks rarely call `Snapshot()` on each event the CPU cost of
+reconstruction is negligible; if they do (and operate on huge snapshots),
+benchmark before turning it on.
+
+## Debug endpoint
+
+Once registered (automatically in `bootstrap.go`), the debug server exposes:
+
+```
+GET /dedup-client/status.{json|yaml|text}
+```
+
+The response carries the status of both components:
+
+```json
+{
+  "client":        { "enabled": true,  "cacheSyncedHint": false },
+  "snapshotStore": { "enabled": true,  "liveObjects": 1284, "totalAcquires": 5012, "totalReleases": 3728, "totalDeletes": 211 }
+}
+```
+
+Each component reports `enabled: false` with a clear `reason` when its flag
+is not set, so liveness probes can distinguish "not configured" from
+"errored".