retry watch when it makes sense

Signed-off-by: Jari Kolehmainen <jari.kolehmainen@gmail.com>

src/renderer/api/kube-api.ts

@@ -96,6 +96,7 @@ export function ensureObjectSelfLink(api: KubeApi, object: KubeJsonApiData) {
 type KubeApiWatchOptions = {
   namespace: string;
   callback?: (data: IKubeWatchEvent) => void;
+  abortController?: AbortController
 };
 
 export class KubeApi<T extends KubeObject = any> {
@@ -366,13 +367,31 @@ export class KubeApi<T extends KubeObject = any> {
   }
 
   watch(opts: KubeApiWatchOptions = { namespace: "" }): () => void {
-    const { namespace, callback } = opts;
+    if (!opts.abortController) {
+      opts.abortController = new AbortController();
+    }
+    const { abortController, namespace, callback } = opts;
     const watchUrl = this.getWatchUrl(namespace);
-    const abortController = new AbortController();
     const responsePromise = this.request.getReadableStream(watchUrl, null, { signal: abortController.signal });
     let disposed = false;
 
     responsePromise.then((response) => {
+      if (!response.ok && !disposed) {
+        if (response.status === 410) { // resourceVersion has gone
+          setTimeout(() => {
+            this.refreshResourceVersion().then(() => {
+              this.watch({...opts, abortController});
+            });
+          }, 1000);
+
+        } else if (response.status >= 500) { // k8s is having hard time
+          setTimeout(() => {
+            this.watch({...opts, abortController});
+          }, 5000);
+        }
+
+        return;
+      }
       const nodeStream = new ReadableWebToNodeStream(response.body);
       const stream = byline(nodeStream);
 
@@ -392,13 +411,9 @@ export class KubeApi<T extends KubeObject = any> {
 
       stream.on("close", () => {
         setTimeout(() => {
-          if (!disposed) this.watch({namespace, callback});
+          if (!disposed) this.watch({...opts, namespace, callback});
         }, 1000);
       });
-
-      stream.on("error", (error) => {
-        console.error("stream error", error);
-      });
     }, (error) => {
       if (error instanceof DOMException) return; // AbortController rejects, we can ignore it
 

src/renderer/kube-object.store.ts

@@ -267,6 +267,7 @@ export abstract class KubeObjectStore<T extends KubeObject = any> extends ItemSt
   }
 
   subscribe(apis = this.getSubscribeApis()) {
+    const abortController = new AbortController();
     let disposers: {(): void}[] = [];
 
     const callback = (data: IKubeWatchEvent) => {
@@ -275,13 +276,15 @@ export abstract class KubeObjectStore<T extends KubeObject = any> extends ItemSt
 
     if (this.context.cluster.isGlobalWatchEnabled) {
       disposers = apis.map(api => api.watch({
+        abortController,
         namespace: "",
-        callback: (data) => callback(data)
+        callback: (data) => callback(data),
       }));
     } else {
       apis.map(api => {
         this.loadedNamespaces.forEach((namespace) => {
           disposers.push(api.watch({
+            abortController,
             namespace,
             callback: (data) => callback(data)
           }));