cluster-status -- part 2

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

src/main/cluster-manager.ts

@@ -1,61 +1,92 @@
 import type http from "http"
-import { autorun } from "mobx";
+import { autorun, reaction } from "mobx";
 import { apiKubePrefix } from "../common/vars";
 import { ClusterId, clusterStore } from "../common/cluster-store"
-import { handleMessage } from "../common/ipc";
-import { tracker } from "../common/tracker";
-import { Cluster, ClusterIpcEvent } from "./cluster"
+import { handleIpc } from "../common/ipc";
+import { Cluster, ClusterIpcChannel } from "./cluster"
 import logger from "./logger";
+import { tracker } from "../common/tracker";
 
 export class ClusterManager {
+  protected activeClusterId: ClusterId;
+
   constructor(public readonly port: number) {
+    this.activeClusterId = clusterStore.activeClusterId;
+
     // auto-init clusters
     autorun(() => {
       clusterStore.clusters.forEach(cluster => {
-        if (cluster.initialized) return;
-        cluster.init(port);
-        logger.info(`[CLUSTER-MANAGER]: initializing cluster`, cluster.getMeta());
+        if (!cluster.initialized) {
+          logger.info(`[CLUSTER-MANAGER]: initializing cluster`, cluster.getMeta());
+          cluster.init(port);
+        }
       });
     });
 
+    // auto-bind events for active cluster
+    reaction(() => clusterStore.activeCluster, activeCluster => {
+      const prevCluster = clusterStore.getById(this.activeClusterId);
+      if (prevCluster) {
+        prevCluster.unbindEvents();
+      }
+      if (activeCluster) {
+        this.activeClusterId = activeCluster.id;
+        activeCluster.bindEvents();
+        activeCluster.refreshStatus();
+      }
+    }, {
+      fireImmediately: true
+    });
+
     // auto-stop removed clusters
     autorun(() => {
       const { removedClusters } = clusterStore;
-      const meta = Array.from(removedClusters.values()).map(cluster => cluster.getMeta());
-      logger.info(`[CLUSTER-MANAGER]: removing clusters`, meta);
-      removedClusters.forEach(cluster => cluster.destroy());
-      removedClusters.clear();
+      if (removedClusters.size > 0) {
+        const meta = Array.from(removedClusters.values()).map(cluster => cluster.getMeta());
+        logger.info(`[CLUSTER-MANAGER]: removing clusters`, meta);
+        removedClusters.forEach(cluster => cluster.disconnect());
+        removedClusters.clear();
+      }
     }, {
       delay: 250
     });
 
-    // listen for ipc-events that must be handled *only* in main-process (nodeIntegration=true)
-    handleMessage(ClusterIpcEvent.STOP, this.stopCluster.bind(this));
+    // listen for ipc-events that must/can be handled *only* in main-process (nodeIntegration=true)
+    handleIpc(ClusterIpcChannel.INIT, this.onClusterInit);
+    handleIpc(ClusterIpcChannel.DISCONNECT, this.onClusterDisconnect);
+    handleIpc(ClusterIpcChannel.RECONNECT, this.onClusterReconnect);
   }
 
   stop() {
     clusterStore.clusters.forEach((cluster: Cluster) => {
-      cluster.stop();
+      cluster.disconnect();
     })
   }
 
+  protected onClusterInit = async (id = clusterStore.activeClusterId) => {
+    const cluster = this.getCluster(id);
+    if (cluster) {
+      logger.info(`[CLUSTER-MANAGER]: init cluster`, cluster.getMeta());
+      tracker.event("cluster", "activate");
+      await cluster.refreshStatus();
+      cluster.pushState();
+    }
+  }
+
+  protected onClusterDisconnect = (id: ClusterId) => {
+    tracker.event("cluster", "stop");
+    this.getCluster(id)?.disconnect();
+  }
+
+  protected onClusterReconnect = (id: ClusterId) => {
+    tracker.event("cluster", "reconnect");
+    this.getCluster(id)?.reconnect();
+  }
+
   protected getCluster(id: ClusterId) {
     return clusterStore.getById(id);
   }
 
-  protected stopCluster(clusterId: ClusterId) {
-    tracker.event("cluster", "stop");
-    this.getCluster(clusterId)?.destroy();
-  }
-
-  // todo
-  protected reconnectCluster(clusterId: ClusterId) {
-    tracker.event("cluster", "reconnect");
-    logger.info(`[CLUSTER-MANAGER]: reconnect cluster`, {
-      meta: this.getCluster(clusterId)?.getMeta()
-    });
-  }
-
   getClusterForRequest(req: http.IncomingMessage): Cluster {
     let cluster: Cluster = null
 

src/main/cluster.ts

@@ -1,9 +1,9 @@
 import type { ClusterId, ClusterModel, ClusterPreferences } from "../common/cluster-store"
 import type { FeatureStatusMap } from "./feature"
 import type { WorkspaceId } from "../common/workspace-store";
-import { action, observable, reaction, toJS, when } from "mobx";
+import { action, computed, observable, reaction, toJS, when } from "mobx";
 import { apiKubePrefix } from "../common/vars";
-import { sendMessage } from "../common/ipc";
+import { broadcastIpc } from "../common/ipc";
 import { ContextHandler } from "./context-handler"
 import { AuthorizationV1Api, CoreV1Api, KubeConfig, V1ResourceAttributes } from "@kubernetes/client-node"
 import { Kubectl } from "./kubectl";
@@ -13,8 +13,9 @@ import { getFeatures, installFeature, uninstallFeature, upgradeFeature } from ".
 import request, { RequestPromiseOptions } from "request-promise-native"
 import logger from "./logger"
 
-export enum ClusterIpcEvent {
-  STOP = "cluster:stop",
+export enum ClusterIpcChannel {
+  INIT = "cluster:init",
+  DISCONNECT = "cluster:disconnect",
   RECONNECT = "cluster:reconnect",
 }
 
@@ -43,8 +44,6 @@ export class Cluster implements ClusterModel {
   public kubeCtl: Kubectl
   public contextHandler: ContextHandler;
   protected kubeconfigManager: KubeconfigManager;
-
-  public whenReady = when(() => this.initialized);
   protected disposers: Function[] = [];
 
   @observable initialized = false;
@@ -69,6 +68,10 @@ export class Cluster implements ClusterModel {
     this.updateModel(model);
   }
 
+  @computed get isReady() {
+    return this.initialized && this.accessible === true;
+  }
+
   @action
   updateModel(model: ClusterModel) {
     Object.assign(this, model);
@@ -98,8 +101,7 @@ export class Cluster implements ClusterModel {
     }
   }
 
-  bindEvents(viewId: number) {
-    if (!this.initialized) return;
+  bindEvents() {
     logger.info(`[CLUSTER]: bind events`, this.getMeta());
     const refreshStatusTimer = setInterval(() => this.refreshStatus(), 30000); // every 30s
     const refreshEventsTimer = setInterval(() => this.refreshEvents(), 3000); // every 3s
@@ -107,43 +109,35 @@ export class Cluster implements ClusterModel {
     this.disposers.push(
       () => clearInterval(refreshStatusTimer),
       () => clearInterval(refreshEventsTimer),
-
-      reaction(() => this.getState(), clusterState => {
-        sendMessage({
-          channel: "cluster:state",
-          webContentId: viewId,
-          args: [clusterState],
-        })
-      }, {
+      reaction(() => this.getState(), this.pushState, {
         fireImmediately: true
       })
     );
   }
 
   unbindEvents() {
-    if (!this.initialized) return;
     logger.info(`[CLUSTER]: unbind events`, this.getMeta());
     this.disposers.forEach(dispose => dispose());
     this.disposers.length = 0;
   }
 
-  stop() {
-    this.contextHandler.stopServer();
+  // fixme: possibly doesn't work as expected
+  async reconnect() {
+    logger.info(`[CLUSTER]: reconnect`, this.getMeta());
+    await this.contextHandler.stopServer();
+    await this.contextHandler.ensureServer();
   }
 
-  destroy() {
-    try {
-      this.stop();
-      this.unbindEvents();
-      this.kubeconfigManager.unlink();
-    } catch (err) {
-      logger.error(`[CLUSTER]: destroy() throws: ${err}`, this.getMeta());
-    }
+  disconnect() {
+    logger.info(`[CLUSTER]: disconnect`, this.getMeta());
+    this.contextHandler.stopServer();
+    this.unbindEvents();
   }
 
   @action
   async refreshStatus() {
-    await this.whenReady;
+    await when(() => this.initialized);
+    logger.info(`[CLUSTER]: refreshing status`, this.getMeta());
     const connectionStatus = await this.getConnectionStatus();
     this.online = connectionStatus > ClusterStatus.Offline;
     this.accessible = connectionStatus == ClusterStatus.AccessGranted;
@@ -191,9 +185,8 @@ export class Cluster implements ClusterModel {
     return this.preferences.prometheus?.prefix || ""
   }
 
-  protected k8sRequest(path: string, options: RequestPromiseOptions = {}) {
+  protected async k8sRequest(path: string, options: RequestPromiseOptions = {}) {
     const apiUrl = this.kubeProxyUrl + path;
-    logger.debug(`[CLUSTER]: getting request to: ${apiUrl}`);
     return request(apiUrl, {
       json: true,
       timeout: 10000,
@@ -211,7 +204,7 @@ export class Cluster implements ClusterModel {
       this.failureReason = null
       return ClusterStatus.AccessGranted;
     } catch (error) {
-      logger.error(`Failed to connect cluster "${this.contextName}": ${error.stack}`)
+      logger.error(`Failed to connect cluster "${this.contextName}": ${error}`)
       if (error.statusCode) {
         if (error.statusCode >= 400 && error.statusCode < 500) {
           this.failureReason = "Invalid credentials";
@@ -347,6 +340,15 @@ export class Cluster implements ClusterModel {
     })
   }
 
+  pushState = (clusterState = this.getState()) => {
+    logger.info(`[CLUSTER]: push-state`, clusterState);
+    broadcastIpc({
+      // webContentId: viewId, // todo: send to cluster-view only
+      channel: "cluster:state",
+      args: [clusterState],
+    })
+  }
+
   // get cluster system meta, e.g. use in "logger"
   getMeta() {
     return {

src/main/context-handler.ts

@@ -12,7 +12,7 @@ import { KubeAuthProxy } from "./kube-auth-proxy"
 export class ContextHandler {
   public proxyPort: number;
   public clusterUrl: UrlWithStringQuery;
-  protected proxyServer: KubeAuthProxy
+  protected kubeAuthProxy: KubeAuthProxy
   protected apiTarget: httpProxy.ServerOptions
   protected prometheusProvider: string
   protected prometheusPath: string
@@ -36,7 +36,7 @@ export class ContextHandler {
     return `${namespace}/services/${service}:${port}`
   }
 
-  public async getPrometheusProvider() {
+  async getPrometheusProvider() {
     if (!this.prometheusProvider) {
       const service = await this.getPrometheusService()
       logger.info(`using ${service.id} as prometheus provider`)
@@ -45,7 +45,7 @@ export class ContextHandler {
     return prometheusProviders.find(p => p.id === this.prometheusProvider)
   }
 
-  public async getPrometheusService(): Promise<PrometheusService> {
+  async getPrometheusService(): Promise<PrometheusService> {
     const providers = this.prometheusProvider ? prometheusProviders.filter(provider => provider.id == this.prometheusProvider) : prometheusProviders;
     const prometheusPromises: Promise<PrometheusService>[] = providers.map(async (provider: PrometheusProvider): Promise<PrometheusService> => {
       const apiClient = this.cluster.getProxyKubeconfig().makeApiClient(CoreV1Api)
@@ -61,19 +61,19 @@ export class ContextHandler {
     }
   }
 
-  public async getPrometheusPath(): Promise<string> {
+  async getPrometheusPath(): Promise<string> {
     if (!this.prometheusPath) {
       this.prometheusPath = await this.resolvePrometheusPath()
     }
     return this.prometheusPath;
   }
 
-  public async resolveAuthProxyUrl() {
+  async resolveAuthProxyUrl() {
     const proxyPort = await this.ensurePort();
     return `http://127.0.0.1:${proxyPort}`;
   }
 
-  public async getApiTarget(isWatchRequest = false): Promise<httpProxy.ServerOptions> {
+  async getApiTarget(isWatchRequest = false): Promise<httpProxy.ServerOptions> {
     if (this.apiTarget && !isWatchRequest) {
       return this.apiTarget
     }
@@ -104,26 +104,26 @@ export class ContextHandler {
     return this.proxyPort
   }
 
-  public async ensureServer() {
-    if (!this.proxyServer) {
+  async ensureServer() {
+    if (!this.kubeAuthProxy) {
       await this.ensurePort();
       const proxyEnv = Object.assign({}, process.env)
       if (this.cluster.preferences.httpsProxy) {
         proxyEnv.HTTPS_PROXY = this.cluster.preferences.httpsProxy
       }
-      this.proxyServer = new KubeAuthProxy(this.cluster, this.proxyPort, proxyEnv)
-      await this.proxyServer.run()
+      this.kubeAuthProxy = new KubeAuthProxy(this.cluster, this.proxyPort, proxyEnv)
+      await this.kubeAuthProxy.run()
     }
   }
 
-  public stopServer() {
-    if (this.proxyServer) {
-      this.proxyServer.exit()
-      this.proxyServer = null
+  stopServer() {
+    if (this.kubeAuthProxy) {
+      this.kubeAuthProxy.exit()
+      this.kubeAuthProxy = null
     }
   }
 
-  public proxyServerError(): string {
-    return this.proxyServer?.lastError || ""
+  get proxyLastError(): string {
+    return this.kubeAuthProxy?.lastError || ""
   }
 }

src/main/kube-auth-proxy.ts

@@ -1,6 +1,6 @@
 import { ChildProcess, spawn } from "child_process"
 import { waitUntilUsed } from "tcp-port-used";
-import { sendMessage } from "../common/ipc";
+import { broadcastIpc } from "../common/ipc";
 import type { Cluster } from "./cluster"
 import { bundledKubectl, Kubectl } from "./kubectl"
 import logger from "./logger"
@@ -85,8 +85,8 @@ export class KubeAuthProxy {
 
   protected async sendIpcLogMessage(res: KubeAuthProxyResponse) {
     const channel = `kube-auth:${this.cluster.id}`
-    logger.debug(`[KUBE-AUTH]: output for ${channel}`, { ...res, meta: this.cluster.getMeta() });
-    sendMessage({
+    logger.info(`[KUBE-AUTH]: out-channel "${channel}"`, { ...res, meta: this.cluster.getMeta() });
+    broadcastIpc({
       // webContentId: null, // todo: send a message only to single cluster's window
       channel: channel,
       args: [res],
@@ -95,7 +95,7 @@ export class KubeAuthProxy {
 
   public exit() {
     if (this.proxyProcess) {
-      logger.debug(`Stopping local proxy: ${this.cluster.contextName}`)
+      logger.debug("[KUBE-AUTH]: stopping local proxy", this.cluster.getMeta())
       this.proxyProcess.kill()
     }
   }

src/main/lens-proxy.ts

@@ -52,21 +52,17 @@ export class LensProxy {
 
   protected createProxy(): httpProxy {
     const proxy = httpProxy.createProxyServer();
-
     proxy.on("proxyRes", (proxyRes, req, res) => {
+      if (req.method !== "GET") {
+        return;
+      }
       if (proxyRes.statusCode === 502) {
         const cluster = this.clusterManager.getClusterForRequest(req)
-        if (cluster && cluster.contextHandler.proxyServerError()) {
-          res.writeHead(proxyRes.statusCode, {
-            "Content-Type": "text/plain"
-          })
-          res.end(cluster.contextHandler.proxyServerError())
-          return
+        const proxyError = cluster?.contextHandler.proxyLastError;
+        if (proxyError) {
+          return res.writeHead(502).end(proxyError);
         }
       }
-      if (req.method !== "GET") {
-        return
-      }
       const reqId = this.getRequestId(req);
       if (this.retryCounters.has(reqId)) {
         logger.debug(`Resetting proxy retry cache for url: ${reqId}`);
@@ -92,10 +88,7 @@ export class LensProxy {
           }
         }
       }
-      res.writeHead(500, {
-        'Content-Type': 'text/plain'
-      })
-      res.end('Oops, something went wrong.')
+      res.writeHead(500).end("Oops, something went wrong.")
     })
 
     return proxy;

src/main/window-manager.ts

@@ -1,4 +1,4 @@
-import { autorun, reaction } from "mobx";
+import { autorun, reaction, when } from "mobx";
 import { BrowserWindow, shell } from "electron"
 import windowStateKeeper from "electron-window-state"
 import type { ClusterId } from "../common/cluster-store";
@@ -9,7 +9,6 @@ import logger from "./logger";
 // fixme: remove switching view delay on first load
 
 export class WindowManager {
-  protected activeClusterId: ClusterId;
   protected activeView: BrowserWindow;
   protected views = new Map<ClusterId, BrowserWindow>();
   protected disposers: CallableFunction[] = [];
@@ -36,21 +35,7 @@ export class WindowManager {
     // Manage reactive state
     this.disposers.push(
       // auto-show active cluster window and subscribe for push-events
-      reaction(() => clusterStore.activeCluster, async activeCluster => {
-        if (this.activeClusterId) {
-          const prevCluster = clusterStore.getById(this.activeClusterId);
-          if (prevCluster) prevCluster.unbindEvents();
-          this.activeClusterId = null;
-        }
-        if (activeCluster) {
-          this.activeClusterId = activeCluster.id;
-          const viewId = await this.activateView(activeCluster.id);
-          if (viewId) {
-            await activeCluster.refreshStatus();
-            activeCluster.bindEvents(viewId);
-          }
-        }
-      }, {
+      reaction(() => clusterStore.activeClusterId, clusterId => this.activateView(clusterId), {
         fireImmediately: true,
       }),
 
@@ -108,7 +93,7 @@ export class WindowManager {
       if (activeView !== view) {
         this.activeView = view;
         if (!isLoadedBefore) {
-          await cluster.whenReady;
+          await when(() => cluster.initialized);
           await view.loadURL(cluster.webContentUrl);
           this.hideSplash();
         }