Improve cluster connection visibility (#4266)

src/main/__test__/kube-auth-proxy.test.ts

@@ -55,7 +55,7 @@ import { ChildProcess, spawn } from "child_process";
 import { bundledKubectlPath, Kubectl } from "../kubectl";
 import { mock, MockProxy } from "jest-mock-extended";
 import { waitUntilUsed } from "tcp-port-used";
-import type { Readable } from "stream";
+import { EventEmitter, Readable } from "stream";
 import { UserStore } from "../../common/user-store";
 import { Console } from "console";
 import { stdout, stderr } from "process";
@@ -134,30 +134,73 @@ describe("kube auth proxy tests", () => {
 
   describe("spawn tests", () => {
     let mockedCP: MockProxy<ChildProcess>;
-    let listeners: Record<string, (...args: any[]) => void>;
+    let listeners: EventEmitter;
     let proxy: KubeAuthProxy;
 
     beforeEach(async () => {
       mockedCP = mock<ChildProcess>();
-      listeners = {};
+      listeners = new EventEmitter();
 
       jest.spyOn(Kubectl.prototype, "checkBinary").mockReturnValueOnce(Promise.resolve(true));
       jest.spyOn(Kubectl.prototype, "ensureKubectl").mockReturnValueOnce(Promise.resolve(false));
       mockedCP.on.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): ChildProcess => {
-        listeners[event] = listener;
+        listeners.on(event, listener);
 
         return mockedCP;
       });
       mockedCP.stderr = mock<Readable>();
       mockedCP.stderr.on.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
-        listeners[`stderr/${event}`] = listener;
+        listeners.on(`stderr/${event}`, listener);
+
+        return mockedCP.stderr;
+      });
+      mockedCP.stderr.off.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
+        listeners.off(`stderr/${event}`, listener);
+
+        return mockedCP.stderr;
+      });
+      mockedCP.stderr.removeListener.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
+        listeners.off(`stderr/${event}`, listener);
+
+        return mockedCP.stderr;
+      });
+      mockedCP.stderr.once.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
+        listeners.once(`stderr/${event}`, listener);
+
+        return mockedCP.stderr;
+      });
+      mockedCP.stderr.removeAllListeners.mockImplementation((event?: string): Readable => {
+        listeners.removeAllListeners(event ?? `stderr/${event}`);
 
         return mockedCP.stderr;
       });
       mockedCP.stdout = mock<Readable>();
       mockedCP.stdout.on.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
-        listeners[`stdout/${event}`] = listener;
-        listeners[`stdout/${event}`]("Starting to serve on 127.0.0.1:9191");
+        listeners.on(`stdout/${event}`, listener);
+
+        if (event === "data") {
+          listeners.emit("stdout/data", "Starting to serve on 127.0.0.1:9191");
+        }
+
+        return mockedCP.stdout;
+      });
+      mockedCP.stdout.once.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
+        listeners.once(`stdout/${event}`, listener);
+
+        return mockedCP.stdout;
+      });
+      mockedCP.stdout.off.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
+        listeners.off(`stdout/${event}`, listener);
+
+        return mockedCP.stdout;
+      });
+      mockedCP.stdout.removeListener.mockImplementation((event: string, listener: (message: any, sendHandle: any) => void): Readable => {
+        listeners.off(`stdout/${event}`, listener);
+
+        return mockedCP.stdout;
+      });
+      mockedCP.stdout.removeAllListeners.mockImplementation((event?: string): Readable => {
+        listeners.removeAllListeners(event ?? `stdout/${event}`);
 
         return mockedCP.stdout;
       });
@@ -175,36 +218,36 @@ describe("kube auth proxy tests", () => {
 
     it("should call spawn and broadcast errors", async () => {
       await proxy.run();
-      listeners["error"]({ message: "foobarbat" });
+      listeners.emit("error", { message: "foobarbat" });
 
-      expect(mockBroadcastIpc).toBeCalledWith("kube-auth:foobar", { data: "foobarbat", error: true });
+      expect(mockBroadcastIpc).toBeCalledWith("cluster:foobar:connection-update", { message: "foobarbat", isError: true });
     });
 
     it("should call spawn and broadcast exit", async () => {
       await proxy.run();
-      listeners["exit"](0);
+      listeners.emit("exit", 0);
 
-      expect(mockBroadcastIpc).toBeCalledWith("kube-auth:foobar", { data: "proxy exited with code: 0", error: false });
+      expect(mockBroadcastIpc).toBeCalledWith("cluster:foobar:connection-update", { message: "proxy exited with code: 0", isError: false });
     });
 
     it("should call spawn and broadcast errors from stderr", async () => {
       await proxy.run();
-      listeners["stderr/data"]("an error");
+      listeners.emit("stderr/data", "an error");
 
-      expect(mockBroadcastIpc).toBeCalledWith("kube-auth:foobar", { data: "an error", error: true });
+      expect(mockBroadcastIpc).toBeCalledWith("cluster:foobar:connection-update", { message: "an error", isError: true });
     });
 
     it("should call spawn and broadcast stdout serving info", async () => {
       await proxy.run();
 
-      expect(mockBroadcastIpc).toBeCalledWith("kube-auth:foobar", { data: "Authentication proxy started\n" });
+      expect(mockBroadcastIpc).toBeCalledWith("cluster:foobar:connection-update", { message: "Authentication proxy started", isError: false });
     });
 
     it("should call spawn and broadcast stdout other info", async () => {
       await proxy.run();
-      listeners["stdout/data"]("some info");
+      listeners.emit("stdout/data", "some info");
 
-      expect(mockBroadcastIpc).toBeCalledWith("kube-auth:foobar", { data: "some info" });
+      expect(mockBroadcastIpc).toBeCalledWith("cluster:foobar:connection-update", { message: "some info", isError: false });
     });
   });
 });

src/main/cluster.ts

@@ -32,7 +32,7 @@ import logger from "./logger";
 import { VersionDetector } from "./cluster-detectors/version-detector";
 import { DetectorRegistry } from "./cluster-detectors/detector-registry";
 import plimit from "p-limit";
-import type { ClusterState, ClusterRefreshOptions, ClusterMetricsResourceType, ClusterId, ClusterMetadata, ClusterModel, ClusterPreferences, ClusterPrometheusPreferences, UpdateClusterModel } from "../common/cluster-types";
+import type { ClusterState, ClusterRefreshOptions, ClusterMetricsResourceType, ClusterId, ClusterMetadata, ClusterModel, ClusterPreferences, ClusterPrometheusPreferences, UpdateClusterModel, KubeAuthUpdate } from "../common/cluster-types";
 import { ClusterMetadataKey, initialNodeShellImage, ClusterStatus } from "../common/cluster-types";
 import { storedKubeConfigFolder, toJS } from "../common/utils";
 import type { Response } from "request";
@@ -117,12 +117,6 @@ export class Cluster implements ClusterModel, ClusterState {
    * @observable
    */
   @observable disconnected = true;
-  /**
-   * Connection failure reason
-   *
-   * @observable
-   */
-  @observable failureReason: string;
   /**
    * Does user have admin like access
    *
@@ -358,14 +352,20 @@ export class Cluster implements ClusterModel, ClusterState {
     if (this.disconnected || !this.accessible) {
       await this.reconnect();
     }
+
+    this.broadcastConnectUpdate("Refreshing connection status ...");
     await this.refreshConnectionStatus();
 
     if (this.accessible) {
+      this.broadcastConnectUpdate("Refreshing cluster accessibility ...");
       await this.refreshAccessibility();
-      this.ensureKubectl(); // download kubectl in background, so it's not blocking dashboard
+      // download kubectl in background, so it's not blocking dashboard
+      this.ensureKubectl()
+        .catch(error => logger.warn(`[CLUSTER]: failed to download kubectl for clusterId=${this.id}`, error));
+      this.broadcastConnectUpdate("Connected, waiting for view to load ...");
     }
-    this.activated = true;
 
+    this.activated = true;
     this.pushState();
   }
 
@@ -445,9 +445,8 @@ export class Cluster implements ClusterModel, ClusterState {
   private async refreshAccessibility(): Promise<void> {
     this.isAdmin = await this.isClusterAdmin();
     this.isGlobalWatchEnabled = await this.canUseWatchApi({ resource: "*" });
-
-    await this.refreshAllowedResources();
-
+    this.allowedNamespaces = await this.getAllowedNamespaces();
+    this.allowedResources = await this.getAllowedResources();
     this.ready = true;
   }
 
@@ -462,15 +461,6 @@ export class Cluster implements ClusterModel, ClusterState {
     this.accessible = connectionStatus == ClusterStatus.AccessGranted;
   }
 
-  /**
-   * @internal
-   */
-  @action
-  async refreshAllowedResources() {
-    this.allowedNamespaces = await this.getAllowedNamespaces();
-    this.allowedResources = await this.getAllowedResources();
-  }
-
   async getKubeconfig(): Promise<KubeConfig> {
     const { config } = await loadConfigFromFile(this.kubeConfigPath);
 
@@ -501,34 +491,35 @@ export class Cluster implements ClusterModel, ClusterState {
 
       this.metadata.version = versionData.value;
 
-      this.failureReason = null;
-
       return ClusterStatus.AccessGranted;
     } catch (error) {
-      logger.error(`Failed to connect cluster "${this.contextName}": ${error}`);
+      logger.error(`[CLUSTER]: Failed to connect to "${this.contextName}": ${error}`);
 
       if (error.statusCode) {
         if (error.statusCode >= 400 && error.statusCode < 500) {
-          this.failureReason = "Invalid credentials";
-
-          return ClusterStatus.AccessDenied;
-        } else {
-          this.failureReason = error.error || error.message;
-
-          return ClusterStatus.Offline;
-        }
-      } else if (error.failed === true) {
-        if (error.timedOut === true) {
-          this.failureReason = "Connection timed out";
-
-          return ClusterStatus.Offline;
-        } else {
-          this.failureReason = "Failed to fetch credentials";
+          this.broadcastConnectUpdate("Invalid credentials", true);
 
           return ClusterStatus.AccessDenied;
         }
+
+        this.broadcastConnectUpdate(error.error || error.message, true);
+
+        return ClusterStatus.Offline;
       }
-      this.failureReason = error.message;
+
+      if (error.failed === true) {
+        if (error.timedOut === true) {
+          this.broadcastConnectUpdate("Connection timed out", true);
+
+          return ClusterStatus.Offline;
+        }
+
+        this.broadcastConnectUpdate("Failed to fetch credentials", true);
+
+        return ClusterStatus.AccessDenied;
+      }
+
+      this.broadcastConnectUpdate(error.message, true);
 
       return ClusterStatus.Offline;
     }
@@ -579,7 +570,7 @@ export class Cluster implements ClusterModel, ClusterState {
   }
 
   toJSON(): ClusterModel {
-    const model: ClusterModel = {
+    return toJS({
       id: this.id,
       contextName: this.contextName,
       kubeConfigPath: this.kubeConfigPath,
@@ -589,29 +580,24 @@ export class Cluster implements ClusterModel, ClusterState {
       metadata: this.metadata,
       accessibleNamespaces: this.accessibleNamespaces,
       labels: this.labels,
-    };
-
-    return toJS(model);
+    });
   }
 
   /**
    * Serializable cluster-state used for sync btw main <-> renderer
    */
   getState(): ClusterState {
-    const state: ClusterState = {
+    return toJS({
       apiUrl: this.apiUrl,
       online: this.online,
       ready: this.ready,
       disconnected: this.disconnected,
       accessible: this.accessible,
-      failureReason: this.failureReason,
       isAdmin: this.isAdmin,
       allowedNamespaces: this.allowedNamespaces,
       allowedResources: this.allowedResources,
       isGlobalWatchEnabled: this.isGlobalWatchEnabled,
-    };
-
-    return toJS(state);
+    });
   }
 
   /**
@@ -643,6 +629,17 @@ export class Cluster implements ClusterModel, ClusterState {
     };
   }
 
+  /**
+   * broadcast an authentication update concerning this cluster
+   * @internal
+   */
+  broadcastConnectUpdate(message: string, isError = false): void {
+    const update: KubeAuthUpdate = { message, isError };
+
+    logger.debug(`[CLUSTER]: broadcasting connection update`, { ...update, meta: this.getMeta() });
+    broadcastMessage(`cluster:${this.id}:connection-update`, update);
+  }
+
   protected async getAllowedNamespaces() {
     if (this.accessibleNamespaces.length) {
       return this.accessibleNamespaces;

src/main/context-handler.ts

@@ -146,7 +146,7 @@ export class ContextHandler {
         proxyEnv.HTTPS_PROXY = this.cluster.preferences.httpsProxy;
       }
       this.kubeAuthProxy = new KubeAuthProxy(this.cluster, proxyEnv);
-      this.kubeAuthProxy.run();
+      await this.kubeAuthProxy.run();
     }
 
     await this.kubeAuthProxy.whenReady;
@@ -157,8 +157,4 @@ export class ContextHandler {
     this.kubeAuthProxy = undefined;
     this.apiTarget = undefined;
   }
-
-  get proxyLastError(): string {
-    return this.kubeAuthProxy?.lastError || "";
-  }
 }

src/main/kube-auth-proxy.ts

@@ -22,7 +22,6 @@
 import { ChildProcess, spawn } from "child_process";
 import { waitUntilUsed } from "tcp-port-used";
 import { randomBytes } from "crypto";
-import { broadcastMessage } from "../common/ipc";
 import type { Cluster } from "./cluster";
 import { Kubectl } from "./kubectl";
 import logger from "./logger";
@@ -30,22 +29,16 @@ import * as url from "url";
 import { getPortFrom } from "./utils/get-port";
 import { makeObservable, observable, when } from "mobx";
 
-export interface KubeAuthProxyLog {
-  data: string;
-  error?: boolean; // stream=stderr
-}
-
 const startingServeRegex = /^starting to serve on (?<address>.+)/i;
 
 export class KubeAuthProxy {
-  public lastError: string;
   public readonly apiPrefix: string;
 
   public get port(): number {
     return this._port;
   }
 
-  protected _port: number;
+  protected _port?: number;
   protected cluster: Cluster;
   protected env: NodeJS.ProcessEnv = null;
   protected proxyProcess: ChildProcess;
@@ -92,67 +85,56 @@ export class KubeAuthProxy {
 
     this.proxyProcess = spawn(proxyBin, args, { env: this.env });
     this.proxyProcess.on("error", (error) => {
-      this.sendIpcLogMessage({ data: error.message, error: true });
+      this.cluster.broadcastConnectUpdate(error.message, true);
       this.exit();
     });
 
     this.proxyProcess.on("exit", (code) => {
-      this.sendIpcLogMessage({ data: `proxy exited with code: ${code}`, error: code > 0 });
+      this.cluster.broadcastConnectUpdate(`proxy exited with code: ${code}`, code > 0);
+      this.exit();
+    });
+
+    this.proxyProcess.on("disconnect", () => {
+      this.cluster.broadcastConnectUpdate("Proxy disconnected communications", true );
       this.exit();
     });
 
     this.proxyProcess.stderr.on("data", (data) => {
-      this.lastError = this.parseError(data.toString());
-      this.sendIpcLogMessage({ data: data.toString(), error: true });
+      this.cluster.broadcastConnectUpdate(data.toString(), true);
+    });
+
+    this.proxyProcess.stdout.on("data", (data: any) => {
+      if (typeof this._port === "number") {
+        this.cluster.broadcastConnectUpdate(data.toString());
+      }
     });
 
     this._port = await getPortFrom(this.proxyProcess.stdout, {
       lineRegex: startingServeRegex,
-      onFind: () => this.sendIpcLogMessage({ data: "Authentication proxy started\n" }),
+      onFind: () => this.cluster.broadcastConnectUpdate("Authentication proxy started"),
     });
 
-    this.proxyProcess.stdout.on("data", (data: any) => {
-      this.sendIpcLogMessage({ data: data.toString() });
-    });
+    try {
+      await waitUntilUsed(this.port, 500, 10000);
+      this.ready = true;
+    } catch (error) {
+      this.cluster.broadcastConnectUpdate("Proxy port failed to be used within timelimit, restarting...", true);
+      this.exit();
 
-    await waitUntilUsed(this.port, 500, 10000);
-
-    this.ready = true;
-  }
-
-  protected parseError(data: string) {
-    const error = data.split("http: proxy error:").slice(1).join("").trim();
-    let errorMsg = error;
-    const jsonError = error.split("Response: ")[1];
-
-    if (jsonError) {
-      try {
-        const parsedError = JSON.parse(jsonError);
-
-        errorMsg = parsedError.error_description || parsedError.error || jsonError;
-      } catch (_) {
-        errorMsg = jsonError.trim();
-      }
+      return this.run();
     }
-
-    return errorMsg;
-  }
-
-  protected sendIpcLogMessage(res: KubeAuthProxyLog) {
-    const channel = `kube-auth:${this.cluster.id}`;
-
-    logger.info(`[KUBE-AUTH]: out-channel "${channel}"`, { ...res, meta: this.cluster.getMeta() });
-    broadcastMessage(channel, res);
   }
 
   public exit() {
     this.ready = false;
-    if (!this.proxyProcess) return;
-    logger.debug("[KUBE-AUTH]: stopping local proxy", this.cluster.getMeta());
-    this.proxyProcess.kill();
-    this.proxyProcess.removeAllListeners();
-    this.proxyProcess.stderr.removeAllListeners();
-    this.proxyProcess.stdout.removeAllListeners();
-    this.proxyProcess = null;
+
+    if (this.proxyProcess) {
+      logger.debug("[KUBE-AUTH]: stopping local proxy", this.cluster.getMeta());
+      this.proxyProcess.removeAllListeners();
+      this.proxyProcess.stderr.removeAllListeners();
+      this.proxyProcess.stdout.removeAllListeners();
+      this.proxyProcess.kill();
+      this.proxyProcess = null;
+    }
   }
 }

src/main/kubeconfig-manager.ts

@@ -49,7 +49,7 @@ export class KubeconfigManager {
       try {
         this.tempFile = await this.createProxyKubeconfig();
       } catch (err) {
-        logger.error(`Failed to created temp config for auth-proxy`, { err });
+        logger.error(`[KUBECONFIG-MANAGER]: Failed to created temp config for auth-proxy`, { err });
       }
     }
 
@@ -61,7 +61,7 @@ export class KubeconfigManager {
       return;
     }
 
-    logger.info(`Deleting temporary kubeconfig: ${this.tempFile}`);
+    logger.info(`[KUBECONFIG-MANAGER]: Deleting temporary kubeconfig: ${this.tempFile}`);
     await fs.unlink(this.tempFile);
   }
 
@@ -70,7 +70,7 @@ export class KubeconfigManager {
       return;
     }
 
-    logger.info(`Deleting temporary kubeconfig: ${this.tempFile}`);
+    logger.info(`[KUBECONFIG-MANAGER]: Deleting temporary kubeconfig: ${this.tempFile}`);
     await fs.unlink(this.tempFile);
     this.tempFile = undefined;
   }
@@ -80,8 +80,7 @@ export class KubeconfigManager {
       await this.contextHandler.ensureServer();
       this.tempFile = await this.createProxyKubeconfig();
     } catch (err) {
-      console.log(err);
-      logger.error(`Failed to created temp config for auth-proxy`, { err });
+      logger.error(`[KUBECONFIG-MANAGER]: Failed to created temp config for auth-proxy`, err);
     }
   }
 
@@ -124,7 +123,7 @@ export class KubeconfigManager {
 
     await fs.ensureDir(path.dirname(tempFile));
     await fs.writeFile(tempFile, configYaml, { mode: 0o600 });
-    logger.debug(`Created temp kubeconfig "${contextName}" at "${tempFile}": \n${configYaml}`);
+    logger.debug(`[KUBECONFIG-MANAGER]: Created temp kubeconfig "${contextName}" at "${tempFile}": \n${configYaml}`);
 
     return tempFile;
   }

src/main/window-manager.ts

@@ -119,7 +119,7 @@ export class WindowManager extends Singleton {
           logger.info("[WINDOW-MANAGER]: Main window loaded");
         })
         .on("will-attach-webview", (event, webPreferences, params) => {
-          logger.info("[WINDOW-MANAGER]: Attaching webview");
+          logger.debug("[WINDOW-MANAGER]: Attaching webview");
           // Following is security recommendations because we allow webview tag (webviewTag: true)
           // suggested by https://www.electronjs.org/docs/tutorial/security#11-verify-webview-options-before-creation
           // and https://www.electronjs.org/docs/tutorial/security#10-do-not-use-allowpopups