Cancel current request if retried.

Signed-off-by: Panu Horsmalahti <phorsmalahti@mirantis.com>

src/common/k8s-api/__tests__/kube-api.test.ts

@@ -23,6 +23,8 @@ import type { Request } from "node-fetch";
 import { forRemoteCluster, KubeApi } from "../kube-api";
 import { KubeJsonApi } from "../kube-json-api";
 import { KubeObject } from "../kube-object";
+import AbortController from "abort-controller";
+import { delay } from "../../utils/delay";
 
 class TestKubeObject extends KubeObject {
   static kind = "Pod";
@@ -325,5 +327,126 @@ describe("KubeApi", () => {
       api.watch({ namespace: "kube-system", timeout: 60 });
       expect(spy).toHaveBeenCalledWith("/api/v1/namespaces/kube-system/pods?watch=1&resourceVersion=", { query: { timeoutSeconds: 60 }}, expect.anything());
     });
+
+    it("aborts watch using abortController", async (done) => {
+      const spy = jest.spyOn(request, "getResponse");
+
+      (fetch as any).mockResponse(async (request: Request) => {
+        (request as any).signal.addEventListener("abort", () => {
+          done();
+        });
+
+        return {};
+      });
+
+      const abortController = new AbortController();
+      
+      api.watch({
+        namespace: "kube-system",
+        timeout: 60,
+        abortController,
+      });
+
+      expect(spy).toHaveBeenCalledWith("/api/v1/namespaces/kube-system/pods?watch=1&resourceVersion=", { query: { timeoutSeconds: 60 }}, expect.anything());
+
+      await delay(100);
+
+      abortController.abort();
+    });
+
+    describe("retries", () => {
+      it("if request ended", (done) => {
+        const spy = jest.spyOn(request, "getResponse");
+
+        // we need to mock using jest as jest-fetch-mock doesn't support mocking the body completely
+        jest.spyOn(global, "fetch").mockImplementation(async () => {
+          return {
+            ok: true,
+            body: {
+              on: (eventName: string, callback: Function) => {
+                // End the request in 100ms.
+                if (eventName === "end") {
+                  setTimeout(() => {
+                    callback();
+                  }, 100);
+                }
+              },
+            },
+          } as any;
+        });
+      
+        api.watch({
+          namespace: "kube-system",
+        });
+
+        expect(spy).toHaveBeenCalledTimes(1);
+
+        setTimeout(() => {  
+          expect(spy).toHaveBeenCalledTimes(2);
+          done();
+        }, 2000);
+      });
+
+      it("if request not closed after timeout", (done) => {
+        const spy = jest.spyOn(request, "getResponse");
+
+        (fetch as any).mockResponse(async () => {
+          return {};
+        });
+
+        const timeoutSeconds = 1;
+      
+        api.watch({
+          namespace: "kube-system",
+          timeout: timeoutSeconds,
+        });
+
+        expect(spy).toHaveBeenCalledTimes(1);
+
+        setTimeout(() => {  
+          expect(spy).toHaveBeenCalledTimes(2);
+          done();
+        }, timeoutSeconds * 1000 * 1.2);
+      });
+
+      it("retries only once if request ends and timeout is set", (done) => {
+        const spy = jest.spyOn(request, "getResponse");
+
+        // we need to mock using jest as jest-fetch-mock doesn't support mocking the body completely
+        jest.spyOn(global, "fetch").mockImplementation(async () => {
+          return {
+            ok: true,
+            body: {
+              on: (eventName: string, callback: Function) => {
+                // End the request in 100ms
+                if (eventName === "end") {
+                  setTimeout(() => {
+                    callback();
+                  }, 100);
+                }
+              },
+            },
+          } as any;
+        });
+      
+        const timeoutSeconds = 0.5;
+      
+        api.watch({
+          namespace: "kube-system",
+          timeout: timeoutSeconds,
+        });
+
+        expect(spy).toHaveBeenCalledTimes(1);
+
+        setTimeout(() => {  
+          expect(spy).toHaveBeenCalledTimes(2);
+          done();
+        }, 2000);
+      });
+
+      afterEach(() => {
+        jest.clearAllMocks();
+      });
+    });
   });
 });

src/common/k8s-api/kube-api.ts

@@ -208,6 +208,8 @@ export type KubeApiWatchOptions = {
   abortController?: AbortController
   watchId?: string;
   retry?: boolean;
+
+  // timeout in seconds
   timeout?: number;
 };
 
@@ -529,9 +531,18 @@ export class KubeApi<T extends KubeObject> {
     let errorReceived = false;
     let timedRetry: NodeJS.Timeout;
     const { namespace, callback = noop, retry, timeout } = opts;
-    const abortController = opts.abortController ?? new AbortController();
     const { watchId = `${this.kind.toLowerCase()}-${this.watchId++}` } = opts;
 
+    // Create AbortController for this request
+    const abortController = new AbortController();
+
+    // If caller aborts, abort using request's abortController
+    if (opts.abortController) {
+      opts.abortController.signal.addEventListener("abort", () => {
+        abortController.abort();
+      });
+    }
+
     abortController.signal.addEventListener("abort", () => {
       logger.info(`[KUBE-API] watch (${watchId}) aborted ${watchUrl}`);
       clearTimeout(timedRetry);
@@ -563,16 +574,11 @@ export class KubeApi<T extends KubeObject> {
           setTimeout(() => {
             // We only retry if we haven't retried, haven't aborted and haven't received k8s error
             if (retried || abortController.signal.aborted || errorReceived) {
-              // TODO: Reduce logging here
-              logger.info(`[KUBE-API] Watch timeout set, returning due to retried: ${retried} aborted: ${abortController.signal.aborted} errorReceived: ${errorReceived}`);
-
               return;
             }
 
-            // Close current request, how?
-            // If we call abortController.abort();, that will abort the next request too,
-            // as abortController is passed down
-            // abortController.abort();
+            // Close current request
+            abortController.abort();
 
             logger.info(`[KUBE-API] Watch timeout set, but not retried, retrying now`);
 
@@ -590,9 +596,6 @@ export class KubeApi<T extends KubeObject> {
             // We only retry if we haven't retried, haven't aborted and haven't received k8s error
             // kubernetes errors (=errorReceived set) should be handled in a callback
             if (retried || abortController.signal.aborted || errorReceived) {
-              // TODO: Reduce logging here
-              logger.info(`[KUBE-API] response event ${eventName} returning due to retried: ${retried} signal.aborted: ${abortController.signal.aborted} errorReceived: ${errorReceived}`);
-
               return;
             }