Refactor watches to use native k8s api (#2095)

* fix lint

Signed-off-by: Roman <ixrock@gmail.com>

* fixes & refactoring

Signed-off-by: Roman <ixrock@gmail.com>

* fix lint, micro-refactoring

Signed-off-by: Roman <ixrock@gmail.com>

* more refactoring, clean up, responding to comments

Signed-off-by: Roman <ixrock@gmail.com>

* fix: remove extra check for cluster.allowedApi from processing buffered watch-api events

Signed-off-by: Roman <ixrock@gmail.com>

* refactoring, detaching NamespaceStore from KubeObjectStore

Signed-off-by: Roman <ixrock@gmail.com>

* fix: wait for contextReady in NamespaceStore

Signed-off-by: Roman <ixrock@gmail.com>

* refactoring & fixes

Signed-off-by: Roman <ixrock@gmail.com>

* fix lint

Signed-off-by: Roman <ixrock@gmail.com>

* fixes: reloading context stores on NamespaceSelect-change

Signed-off-by: Roman <ixrock@gmail.com>

* optimize loading all resources when "all namespaces" selected -> single request per resource (when have rights)

Signed-off-by: Roman <ixrock@gmail.com>

* use native k8s api watches

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

* retry watch when it makes sense

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

* workaround for browser connection limits

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

* cleanup

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

* cleanup

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

* use always random subdomain for getResponse

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

* resubscribe stores on contextNamespace change

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

* fix

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

* modify watch event before calling callback

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

Co-authored-by: Roman <ixrock@gmail.com>

src/renderer/api/json-api.ts

@@ -3,7 +3,7 @@
 import { stringify } from "querystring";
 import { EventEmitter } from "../../common/event-emitter";
 import { cancelableFetch } from "../utils/cancelableFetch";
-
+import { randomBytes } from "crypto";
 export interface JsonApiData {
 }
 
@@ -55,6 +55,34 @@ export class JsonApi<D = JsonApiData, P extends JsonApiParams = JsonApiParams> {
     return this.request<T>(path, params, { ...reqInit, method: "get" });
   }
 
+  getResponse(path: string, params?: P, init: RequestInit = {}): Promise<Response> {
+    const reqPath = `${this.config.apiBase}${path}`;
+    const subdomain = randomBytes(2).toString("hex");
+    let reqUrl = `http://${subdomain}.${window.location.host}${reqPath}`; // hack around browser connection limits (chromium allows 6 per domain)
+    const reqInit: RequestInit = { ...init };
+    const { query } = params || {} as P;
+
+    if (!reqInit.method) {
+      reqInit.method = "get";
+    }
+
+    if (query) {
+      const queryString = stringify(query);
+
+      reqUrl += (reqUrl.includes("?") ? "&" : "?") + queryString;
+    }
+
+    const infoLog: JsonApiLog = {
+      method: reqInit.method.toUpperCase(),
+      reqUrl: reqPath,
+      reqInit,
+    };
+
+    this.writeLog({ ...infoLog });
+
+    return fetch(reqUrl, reqInit);
+  }
+
   post<T = D>(path: string, params?: P, reqInit: RequestInit = {}) {
     return this.request<T>(path, params, { ...reqInit, method: "post" });
   }

src/renderer/api/kube-api.ts

@@ -9,7 +9,9 @@ import { apiKube } from "./index";
 import { createKubeApiURL, parseKubeApi } from "./kube-api-parse";
 import { KubeJsonApi, KubeJsonApiData, KubeJsonApiDataList } from "./kube-json-api";
 import { IKubeObjectConstructor, KubeObject } from "./kube-object";
-import { kubeWatchApi } from "./kube-watch-api";
+import byline from "byline";
+import { ReadableWebToNodeStream } from "readable-web-to-node-stream";
+import { IKubeWatchEvent } from "./kube-watch-api";
 
 export interface IKubeApiOptions<T extends KubeObject> {
   /**
@@ -91,6 +93,12 @@ export function ensureObjectSelfLink(api: KubeApi, object: KubeJsonApiData) {
   }
 }
 
+type KubeApiWatchOptions = {
+  namespace: string;
+  callback?: (data: IKubeWatchEvent) => void;
+  abortController?: AbortController
+};
+
 export class KubeApi<T extends KubeObject = any> {
   readonly kind: string;
   readonly apiBase: string;
@@ -104,6 +112,7 @@ export class KubeApi<T extends KubeObject = any> {
   public objectConstructor: IKubeObjectConstructor<T>;
   protected request: KubeJsonApi;
   protected resourceVersions = new Map<string, string>();
+  protected watchDisposer: () => void;
 
   constructor(protected options: IKubeApiOptions<T>) {
     const {
@@ -357,8 +366,88 @@ export class KubeApi<T extends KubeObject = any> {
     });
   }
 
-  watch(): () => void {
-    return kubeWatchApi.subscribeApi(this);
+  watch(opts: KubeApiWatchOptions = { namespace: "" }): () => void {
+    if (!opts.abortController) {
+      opts.abortController = new AbortController();
+    }
+    const { abortController, namespace, callback } = opts;
+
+    const watchUrl = this.getWatchUrl(namespace);
+    const responsePromise = this.request.getResponse(watchUrl, null, {
+      signal: abortController.signal
+    });
+
+    responsePromise.then((response) => {
+      if (!response.ok && !abortController.signal.aborted) {
+        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);
+
+      stream.on("data", (line) => {
+        try {
+          const event: IKubeWatchEvent = JSON.parse(line);
+
+          this.modifyWatchEvent(event);
+
+          if (callback) {
+            callback(event);
+          }
+        } catch (ignore) {
+          // ignore parse errors
+        }
+      });
+
+      stream.on("close", () => {
+        setTimeout(() => {
+          if (!abortController.signal.aborted) this.watch({...opts, namespace, callback});
+        }, 1000);
+      });
+    }, (error) => {
+      if (error instanceof DOMException) return; // AbortController rejects, we can ignore it
+
+      console.error("watch rejected", error);
+    }).catch((error) => {
+      console.error("watch error", error);
+    });
+
+    const disposer = () => {
+      abortController.abort();
+    };
+
+    return disposer;
+  }
+
+  protected modifyWatchEvent(event: IKubeWatchEvent) {
+
+    switch (event.type) {
+      case "ADDED":
+      case "DELETED":
+
+      case "MODIFIED": {
+        ensureObjectSelfLink(this, event.object);
+
+        const { namespace, resourceVersion } = event.object.metadata;
+
+        this.setResourceVersion(namespace, resourceVersion);
+        this.setResourceVersion("", resourceVersion);
+
+        break;
+      }
+    }
   }
 }
 

src/renderer/api/kube-watch-api.ts

@@ -1,143 +1,63 @@
 // Kubernetes watch-api client
 // API: https://developer.mozilla.org/en-US/docs/Web/API/Streams_API/Using_readable_streams
 
-import type { Cluster } from "../../main/cluster";
-import type { IKubeWatchEvent, IKubeWatchEventStreamEnd, IWatchRoutePayload } from "../../main/routes/watch-route";
-import type { KubeObject } from "./kube-object";
 import type { KubeObjectStore } from "../kube-object.store";
+import type { ClusterContext } from "../components/context";
 
 import plimit from "p-limit";
-import debounce from "lodash/debounce";
-import { autorun, comparer, computed, IReactionDisposer, observable, reaction } from "mobx";
-import { autobind, EventEmitter, noop } from "../utils";
-import { ensureObjectSelfLink, KubeApi, parseKubeApi } from "./kube-api";
-import { KubeJsonApiData, KubeJsonApiError } from "./kube-json-api";
-import { apiPrefix, isDebugging, isProduction } from "../../common/vars";
-import { apiManager } from "./api-manager";
+import { comparer, IReactionDisposer, observable, reaction, when } from "mobx";
+import { autobind, noop } from "../utils";
+import { KubeApi } from "./kube-api";
+import { KubeJsonApiData } from "./kube-json-api";
+import { isDebugging, isProduction } from "../../common/vars";
 
-export { IKubeWatchEvent, IKubeWatchEventStreamEnd };
-
-export interface IKubeWatchMessage<T extends KubeObject = any> {
-  namespace?: string;
-  data?: IKubeWatchEvent<KubeJsonApiData>
-  error?: IKubeWatchEvent<KubeJsonApiError>;
-  api?: KubeApi<T>;
-  store?: KubeObjectStore<T>;
+export interface IKubeWatchEvent<T = KubeJsonApiData> {
+  type: "ADDED" | "MODIFIED" | "DELETED";
+  object?: T;
 }
 
 export interface IKubeWatchSubscribeStoreOptions {
+  namespaces?: string[]; // default: all accessible namespaces
   preload?: boolean; // preload store items, default: true
   waitUntilLoaded?: boolean; // subscribe only after loading all stores, default: true
   loadOnce?: boolean; // check store.isLoaded to skip loading if done already, default: false
 }
 
-export interface IKubeWatchReconnectOptions {
-  reconnectAttempts: number;
-  timeout: number;
-}
-
 export interface IKubeWatchLog {
-  message: string | Error;
+  message: string | string[] | Error;
   meta?: object;
+  cssStyle?: string;
 }
 
 @autobind()
 export class KubeWatchApi {
-  private requestId = 0;
-  private reader: ReadableStreamReader<string>;
-  public onMessage = new EventEmitter<[IKubeWatchMessage]>();
-
-  @observable.ref private cluster: Cluster;
-  @observable.ref private namespaces: string[] = [];
+  @observable context: ClusterContext = null;
   @observable subscribers = observable.map<KubeApi, number>();
   @observable isConnected = false;
 
-  @computed get isReady(): boolean {
-    return Boolean(this.cluster && this.namespaces);
+  contextReady = when(() => Boolean(this.context));
+
+  constructor() {
+    this.init();
   }
 
-  @computed get isActive(): boolean {
-    return this.apis.length > 0;
-  }
-
-  @computed get apis(): string[] {
-    if (!this.isReady) {
-      return [];
-    }
-
-    return Array.from(this.subscribers.keys()).map(api => {
-      if (!this.isAllowedApi(api)) {
-        return [];
-      }
-
-      // TODO: optimize - check when all namespaces are selected and then request all in one
-      if (api.isNamespaced && !this.cluster.isGlobalWatchEnabled) {
-        return this.namespaces.map(namespace => api.getWatchUrl(namespace));
-      }
-
-      return api.getWatchUrl();
-    }).flat();
-  }
-
-  async init({ getCluster, getNamespaces }: {
-    getCluster: () => Cluster,
-    getNamespaces: () => string[],
-  }): Promise<void> {
-    autorun(() => {
-      this.cluster = getCluster();
-      this.namespaces = getNamespaces();
-    });
-    this.bindAutoConnect();
-  }
-
-  private bindAutoConnect() {
-    const connect = debounce(() => this.connect(), 1000);
-
-    reaction(() => this.apis, connect, {
-      fireImmediately: true,
-      equals: comparer.structural,
-    });
-
-    window.addEventListener("online", () => this.connect());
-    window.addEventListener("offline", () => this.disconnect());
-    setInterval(() => this.connectionCheck(), 60000 * 5); // every 5m
-  }
-
-  getSubscribersCount(api: KubeApi) {
-    return this.subscribers.get(api) || 0;
+  private async init() {
+    await this.contextReady;
   }
 
   isAllowedApi(api: KubeApi): boolean {
-    return Boolean(this?.cluster.isAllowedResource(api.kind));
+    return Boolean(this.context?.cluster.isAllowedResource(api.kind));
   }
 
-  subscribeApi(api: KubeApi | KubeApi[]): () => void {
-    const apis: KubeApi[] = [api].flat();
-
-    apis.forEach(api => {
-      if (!this.isAllowedApi(api)) return; // skip
-      this.subscribers.set(api, this.getSubscribersCount(api) + 1);
-    });
-
-    return () => {
-      apis.forEach(api => {
-        const count = this.getSubscribersCount(api) - 1;
-
-        if (count <= 0) this.subscribers.delete(api);
-        else this.subscribers.set(api, count);
-      });
-    };
-  }
-
-  preloadStores(stores: KubeObjectStore[], { loadOnce = false } = {}) {
+  preloadStores(stores: KubeObjectStore[], opts: { namespaces?: string[], loadOnce?: boolean } = {}) {
     const limitRequests = plimit(1); // load stores one by one to allow quick skipping when fast clicking btw pages
     const preloading: Promise<any>[] = [];
 
     for (const store of stores) {
       preloading.push(limitRequests(async () => {
-        if (store.isLoaded && loadOnce) return; // skip
+        if (store.isLoaded && opts.loadOnce) return; // skip
 
-        return store.loadAll(this.namespaces);
+        return store.loadAll({ namespaces: opts.namespaces });
       }));
     }
 
@@ -147,19 +67,22 @@ export class KubeWatchApi {
     };
   }
 
-  subscribeStores(stores: KubeObjectStore[], options: IKubeWatchSubscribeStoreOptions = {}): () => void {
-    const { preload = true, waitUntilLoaded = true, loadOnce = false } = options;
-    const apis = new Set(stores.map(store => store.getSubscribeApis()).flat());
-    const unsubscribeList: (() => void)[] = [];
+  subscribeStores(stores: KubeObjectStore[], opts: IKubeWatchSubscribeStoreOptions = {}): () => void {
+    const { preload = true, waitUntilLoaded = true, loadOnce = false, } = opts;
+    const subscribingNamespaces = opts.namespaces ?? this.context?.allNamespaces ?? [];
+    const unsubscribeList: Function[] = [];
     let isUnsubscribed = false;
 
-    const load = () => this.preloadStores(stores, { loadOnce });
+    const load = (namespaces = subscribingNamespaces) => this.preloadStores(stores, { namespaces, loadOnce });
     let preloading = preload && load();
     let cancelReloading: IReactionDisposer = noop;
 
     const subscribe = () => {
       if (isUnsubscribed) return;
-      apis.forEach(api => unsubscribeList.push(this.subscribeApi(api)));
+
+      stores.forEach((store) => {
+        unsubscribeList.push(store.subscribe());
+      });
     };
 
     if (preloading) {
@@ -167,17 +90,20 @@ export class KubeWatchApi {
         preloading.loading.then(subscribe, error => {
           this.log({
             message: new Error("Loading stores has failed"),
-            meta: { stores, error, options },
+            meta: { stores, error, options: opts },
           });
         });
       } else {
         subscribe();
       }
 
-      // reload when context namespaces changes
-      cancelReloading = reaction(() => this.namespaces, () => {
+      // reload stores only for context namespaces change
+      cancelReloading = reaction(() => this.context?.contextNamespaces, namespaces => {
         preloading?.cancelLoading();
-        preloading = load();
+        unsubscribeList.forEach(unsubscribe => unsubscribe());
+        unsubscribeList.length = 0;
+        preloading = load(namespaces);
+        preloading.loading.then(subscribe);
       }, {
         equals: comparer.shallow,
       });
@@ -190,184 +116,25 @@ export class KubeWatchApi {
       cancelReloading();
       preloading?.cancelLoading();
       unsubscribeList.forEach(unsubscribe => unsubscribe());
+      unsubscribeList.length = 0;
     };
   }
 
-  protected async connectionCheck() {
-    if (!this.isConnected) {
-      this.log({ message: "Offline: reconnecting.." });
-      await this.connect();
-    }
-
-    this.log({
-      message: `Connection check: ${this.isConnected ? "online" : "offline"}`,
-      meta: { connected: this.isConnected },
-    });
-  }
-
-  protected async connect(apis = this.apis) {
-    this.disconnect(); // close active connections first
-
-    if (!navigator.onLine || !apis.length) {
-      this.isConnected = false;
-
-      return;
-    }
-
-    this.log({
-      message: "Connecting",
-      meta: { apis }
-    });
-
-    try {
-      const requestId = ++this.requestId;
-      const abortController = new AbortController();
-
-      const request = await fetch(`${apiPrefix}/watch`, {
-        method: "POST",
-        body: JSON.stringify({ apis } as IWatchRoutePayload),
-        signal: abortController.signal,
-        headers: {
-          "content-type": "application/json"
-        }
-      });
-
-      // request above is stale since new request-id has been issued
-      if (this.requestId !== requestId) {
-        abortController.abort();
-
-        return;
-      }
-
-      let jsonBuffer = "";
-      const stream = request.body.pipeThrough(new TextDecoderStream());
-      const reader = stream.getReader();
-
-      this.isConnected = true;
-      this.reader = reader;
-
-      while (true) {
-        const { done, value } = await reader.read();
-
-        if (done) break; // exit
-
-        const events = (jsonBuffer + value).split("\n");
-
-        jsonBuffer = this.processBuffer(events);
-      }
-    } catch (error) {
-      this.log({ message: error });
-    } finally {
-      this.isConnected = false;
-    }
-  }
-
-  protected disconnect() {
-    this.reader?.cancel();
-    this.reader = null;
-    this.isConnected = false;
-  }
-
-  // process received stream events, returns unprocessed buffer chunk if any
-  protected processBuffer(events: string[]): string {
-    for (const json of events) {
-      try {
-        const kubeEvent: IKubeWatchEvent = JSON.parse(json);
-        const message = this.getMessage(kubeEvent);
-
-        if (!this.namespaces.includes(message.namespace)) {
-          continue; // skip updates from non-watching resources context
-        }
-
-        this.onMessage.emit(message);
-      } catch (error) {
-        return json;
-      }
-    }
-
-    return "";
-  }
-
-  protected getMessage(event: IKubeWatchEvent): IKubeWatchMessage {
-    const message: IKubeWatchMessage = {};
-
-    switch (event.type) {
-      case "ADDED":
-      case "DELETED":
-
-      case "MODIFIED": {
-        const data = event as IKubeWatchEvent<KubeJsonApiData>;
-        const api = apiManager.getApiByKind(data.object.kind, data.object.apiVersion);
-
-        message.data = data;
-
-        if (api) {
-          ensureObjectSelfLink(api, data.object);
-
-          const { namespace, resourceVersion } = data.object.metadata;
-
-          api.setResourceVersion(namespace, resourceVersion);
-          api.setResourceVersion("", resourceVersion);
-
-          message.api = api;
-          message.store = apiManager.getStore(api);
-          message.namespace = namespace;
-        }
-        break;
-      }
-
-      case "ERROR":
-        message.error = event as IKubeWatchEvent<KubeJsonApiError>;
-        break;
-
-      case "STREAM_END": {
-        this.onServerStreamEnd(event as IKubeWatchEventStreamEnd, {
-          reconnectAttempts: 5,
-          timeout: 1000,
-        });
-        break;
-      }
-    }
-
-    return message;
-  }
-
-  protected async onServerStreamEnd(event: IKubeWatchEventStreamEnd, opts?: IKubeWatchReconnectOptions) {
-    const { apiBase, namespace } = parseKubeApi(event.url);
-    const api = apiManager.getApi(apiBase);
-
-    if (!api) return;
-
-    try {
-      await api.refreshResourceVersion({ namespace });
-      this.connect();
-    } catch (error) {
-      this.log({
-        message: new Error(`Failed to connect on single stream end: ${error}`),
-        meta: { event, error },
-      });
-
-      if (this.isActive && opts?.reconnectAttempts > 0) {
-        opts.reconnectAttempts--;
-        setTimeout(() => this.onServerStreamEnd(event, opts), opts.timeout); // repeat event
-      }
-    }
-  }
-
-  protected log({ message, meta = {} }: IKubeWatchLog) {
+  protected log({ message, cssStyle = "", meta = {} }: IKubeWatchLog) {
     if (isProduction && !isDebugging) {
       return;
     }
 
-    const logMessage = `%c[KUBE-WATCH-API]: ${String(message).toUpperCase()}`;
-    const isError = message instanceof Error;
-    const textStyle = `font-weight: bold;`;
-    const time = new Date().toLocaleString();
+    const logInfo = [`%c[KUBE-WATCH-API]:`, `font-weight: bold; ${cssStyle}`, message].flat().map(String);
+    const logMeta = {
+      time: new Date().toLocaleString(),
+      ...meta,
+    };
 
-    if (isError) {
-      console.error(logMessage, textStyle, { time, ...meta });
+    if (message instanceof Error) {
+      console.error(...logInfo, logMeta);
     } else {
-      console.info(logMessage, textStyle, { time, ...meta });
+      console.info(...logInfo, logMeta);
     }
   }
 }