feat: add resourceVersion tracking for watches & better recovery (#453)

package.json

@@ -32,7 +32,7 @@
     "@types/ramda": "0.29.9",
     "express": "4.18.2",
     "fast-json-patch": "3.1.1",
-    "kubernetes-fluent-client": "1.10.0",
+    "kubernetes-fluent-client": "2.0.1",
     "pino": "8.17.2",
     "pino-pretty": "10.3.1",
     "prom-client": "15.1.0",

src/lib/controller/index.ts

@@ -45,12 +45,12 @@ export class Controller {
     this.#capabilities = capabilities;
 
     // Initialize the Pepr store for each capability
-    new PeprControllerStore(config, capabilities, `pepr-${config.uuid}-store`, () => {
+    new PeprControllerStore(capabilities, `pepr-${config.uuid}-store`, () => {
       this.#bindEndpoints();
       onReady && onReady();
       Log.info("✅ Controller startup complete");
       // Initialize the schedule store for each capability
-      new PeprControllerStore(config, capabilities, `pepr-${config.uuid}-schedule`, () => {
+      new PeprControllerStore(capabilities, `pepr-${config.uuid}-schedule`, () => {
         Log.info("✅ Scheduling processed");
       });
     });

src/lib/controller/store.ts

@@ -8,7 +8,6 @@ import { startsWith } from "ramda";
 import { Capability } from "../capability";
 import { PeprStore } from "../k8s";
 import Log from "../logger";
-import { ModuleConfig } from "../module";
 import { DataOp, DataSender, DataStore, Storage } from "../storage";
 
 const namespace = "pepr-system";
@@ -20,7 +19,7 @@ export class PeprControllerStore {
   #sendDebounce: NodeJS.Timeout | undefined;
   #onReady?: () => void;
 
-  constructor(config: ModuleConfig, capabilities: Capability[], name: string, onReady?: () => void) {
+  constructor(capabilities: Capability[], name: string, onReady?: () => void) {
     this.#onReady = onReady;
 
     // Setup Pepr State bindings
@@ -71,7 +70,8 @@ export class PeprControllerStore {
   }
 
   #setupWatch = () => {
-    void K8s(PeprStore, { name: this.#name, namespace }).Watch(this.#receive);
+    const watcher = K8s(PeprStore, { name: this.#name, namespace }).Watch(this.#receive);
+    watcher.start().catch(e => Log.error(e, "Error starting Pepr store watch"));
   };
 
   #receive = (store: PeprStore) => {

src/lib/module.ts

@@ -9,6 +9,7 @@ import { ValidateError } from "./errors";
 import { AdmissionRequest, MutateResponse, ValidateResponse, WebhookIgnore } from "./k8s";
 import { CapabilityExport } from "./types";
 import { setupWatch } from "./watch-processor";
+import { Log } from "../lib";
 
 /** Global configuration for the Pepr runtime. */
 export type ModuleConfig = {
@@ -102,7 +103,10 @@ export class PeprModule {
     this.#controller = new Controller(config, capabilities, opts.beforeHook, opts.afterHook, () => {
       // Wait for the controller to be ready before setting up watches
       if (isWatchMode() || isDevMode()) {
-        setupWatch(capabilities);
+        setupWatch(config.uuid, capabilities).catch(e => {
+          Log.error(e, "Error setting up watch");
+          process.exit(1);
+        });
       }
     });