mirror of
https://github.com/lensapp/lens.git
synced 2025-05-20 05:10:56 +00:00
Cleanup Cluster, ClusterManger, et al. (#2698)
- Removed `getFreePort` as its use is always a race condition. Change all uses of it to retrive the port after listening - Added `getPortFrom` as a helper function to read a port from a stream - Remove `Cluster.ownerRef` as it is outdated and no longer needed for 5.0 - Remove `Cluster.enabled`, no longer needed because of above - Removed `Cluster.init`, moved its contents into `Cluster.constructor` as nothing in that function is asyncronous. Currently only being run on `main` as a stop gap until `renderer` gets its own version of `Cluster` - Refactored `LensProxy` so as to prevent `pty.node` (a NodeJS extension) being included in `webpack.extension.ts`'s run - Removed the passing around of the proxy port as that can now be accessed from an instance of `LensProxy` Signed-off-by: Sebastian Malton <sebastian@malton.name>
This commit is contained in:
Sebastian Malton
GitHub
parent
7494121a86
commit
9a4cd53182
@ -92,7 +92,6 @@ describe("empty config", () => {
|
||||
expect(storedCluster.id).toBe("foo");
|
||||
expect(storedCluster.preferences.terminalCWD).toBe("/tmp");
|
||||
expect(storedCluster.preferences.icon).toBe("data:image/jpeg;base64, iVBORw0KGgoAAAANSUhEUgAAA1wAAAKoCAYAAABjkf5");
|
||||
expect(storedCluster.enabled).toBe(true);
|
||||
});
|
||||
|
||||
it("removes cluster from store", async () => {
|
||||
@ -215,13 +214,6 @@ describe("config with existing clusters", () => {
|
||||
expect(storedClusters[1].preferences.terminalCWD).toBe("/foo2");
|
||||
expect(storedClusters[2].id).toBe("cluster3");
|
||||
});
|
||||
|
||||
it("marks owned cluster disabled by default", () => {
|
||||
const storedClusters = ClusterStore.getInstance().clustersList;
|
||||
|
||||
expect(storedClusters[0].enabled).toBe(true);
|
||||
expect(storedClusters[2].enabled).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("config with invalid cluster kubeconfig", () => {
|
||||
@ -288,10 +280,7 @@ users:
|
||||
it("does not enable clusters with invalid kubeconfig", () => {
|
||||
const storedClusters = ClusterStore.getInstance().clustersList;
|
||||
|
||||
expect(storedClusters.length).toBe(2);
|
||||
expect(storedClusters[0].enabled).toBeFalsy;
|
||||
expect(storedClusters[1].id).toBe("cluster2");
|
||||
expect(storedClusters[1].enabled).toBeTruthy;
|
||||
expect(storedClusters.length).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@ -64,11 +64,6 @@ export interface ClusterModel {
|
||||
/** Metadata */
|
||||
metadata?: ClusterMetadata;
|
||||
|
||||
/**
|
||||
* If extension sets ownerRef it has to explicitly mark a cluster as enabled during onActive (or when cluster is saved)
|
||||
*/
|
||||
ownerRef?: string;
|
||||
|
||||
/** List of accessible namespaces */
|
||||
accessibleNamespaces?: string[];
|
||||
|
||||
@ -172,9 +167,6 @@ export class ClusterStore extends BaseStore<ClusterStoreModel> {
|
||||
protected pushStateToViewsAutomatically() {
|
||||
if (ipcMain) {
|
||||
this.disposer.push(
|
||||
reaction(() => this.enabledClustersList, () => {
|
||||
this.pushState();
|
||||
}),
|
||||
reaction(() => this.connectedClustersList, () => {
|
||||
this.pushState();
|
||||
}),
|
||||
@ -210,10 +202,6 @@ export class ClusterStore extends BaseStore<ClusterStoreModel> {
|
||||
return Array.from(this.clusters.values());
|
||||
}
|
||||
|
||||
@computed get enabledClustersList(): Cluster[] {
|
||||
return this.clustersList.filter((c) => c.enabled);
|
||||
}
|
||||
|
||||
@computed get active(): Cluster | null {
|
||||
return this.getById(this.activeCluster);
|
||||
}
|
||||
@ -232,13 +220,9 @@ export class ClusterStore extends BaseStore<ClusterStoreModel> {
|
||||
|
||||
@action
|
||||
setActive(clusterId: ClusterId) {
|
||||
const cluster = this.clusters.get(clusterId);
|
||||
|
||||
if (!cluster?.enabled) {
|
||||
clusterId = null;
|
||||
}
|
||||
|
||||
this.activeCluster = clusterId;
|
||||
this.activeCluster = this.clusters.has(clusterId)
|
||||
? clusterId
|
||||
: null;
|
||||
}
|
||||
|
||||
deactivate(id: ClusterId) {
|
||||
@ -274,10 +258,6 @@ export class ClusterStore extends BaseStore<ClusterStoreModel> {
|
||||
? clusterOrModel
|
||||
: new Cluster(clusterOrModel);
|
||||
|
||||
if (!cluster.isManaged) {
|
||||
cluster.enabled = true;
|
||||
}
|
||||
|
||||
this.clusters.set(cluster.id, cluster);
|
||||
|
||||
return cluster;
|
||||
@ -314,18 +294,18 @@ export class ClusterStore extends BaseStore<ClusterStoreModel> {
|
||||
|
||||
// update new clusters
|
||||
for (const clusterModel of clusters) {
|
||||
let cluster = currentClusters.get(clusterModel.id);
|
||||
try {
|
||||
let cluster = currentClusters.get(clusterModel.id);
|
||||
|
||||
if (cluster) {
|
||||
cluster.updateModel(clusterModel);
|
||||
} else {
|
||||
cluster = new Cluster(clusterModel);
|
||||
|
||||
if (!cluster.isManaged && cluster.apiUrl) {
|
||||
cluster.enabled = true;
|
||||
if (cluster) {
|
||||
cluster.updateModel(clusterModel);
|
||||
} else {
|
||||
cluster = new Cluster(clusterModel);
|
||||
}
|
||||
newClusters.set(clusterModel.id, cluster);
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
newClusters.set(clusterModel.id, cluster);
|
||||
}
|
||||
|
||||
// update removed clusters
|
||||
@ -335,7 +315,7 @@ export class ClusterStore extends BaseStore<ClusterStoreModel> {
|
||||
}
|
||||
});
|
||||
|
||||
this.activeCluster = newClusters.get(activeCluster)?.enabled ? activeCluster : null;
|
||||
this.setActive(activeCluster);
|
||||
this.clusters.replace(newClusters);
|
||||
this.removedClusters.replace(removedClusters);
|
||||
}
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
// Lens-extensions api developer's kit
|
||||
export * from "../lens-main-extension";
|
||||
export * from "../lens-renderer-extension";
|
||||
export { LensMainExtension } from "../lens-main-extension";
|
||||
export { LensRendererExtension } from "../lens-renderer-extension";
|
||||
|
||||
// APIs
|
||||
import * as App from "./app";
|
||||
|
||||
@ -32,7 +32,6 @@ import { Console } from "console";
|
||||
import mockFs from "mock-fs";
|
||||
import { Cluster } from "../cluster";
|
||||
import { ContextHandler } from "../context-handler";
|
||||
import { getFreePort } from "../port";
|
||||
import { V1ResourceAttributes } from "@kubernetes/client-node";
|
||||
import { apiResources } from "../../common/rbac";
|
||||
import request from "request-promise-native";
|
||||
@ -99,18 +98,7 @@ describe("create clusters", () => {
|
||||
expect(() => c.disconnect()).not.toThrowError();
|
||||
});
|
||||
|
||||
it("init should not throw if everything is in order", async () => {
|
||||
await c.init(await getFreePort());
|
||||
expect(logger.info).toBeCalledWith(expect.stringContaining("init success"), {
|
||||
id: "foo",
|
||||
apiUrl: "https://192.168.64.3:8443",
|
||||
context: "minikube",
|
||||
});
|
||||
});
|
||||
|
||||
it("activating cluster should try to connect to cluster and do a refresh", async () => {
|
||||
const port = await getFreePort();
|
||||
|
||||
jest.spyOn(ContextHandler.prototype, "ensureServer");
|
||||
|
||||
const mockListNSs = jest.fn();
|
||||
@ -142,9 +130,7 @@ describe("create clusters", () => {
|
||||
}
|
||||
}));
|
||||
|
||||
mockedRequest.mockImplementationOnce(((uri: any) => {
|
||||
expect(uri).toBe(`http://localhost:${port}/api-kube/version`);
|
||||
|
||||
mockedRequest.mockImplementationOnce((() => {
|
||||
return Promise.resolve({ gitVersion: "1.2.3" });
|
||||
}) as any);
|
||||
|
||||
@ -162,7 +148,6 @@ describe("create clusters", () => {
|
||||
kubeConfigPath: "minikube-config.yml"
|
||||
});
|
||||
|
||||
await c.init(port);
|
||||
await c.activate();
|
||||
|
||||
expect(ContextHandler.prototype.ensureServer).toBeCalled();
|
||||
|
||||
@ -29,7 +29,6 @@ jest.mock("tcp-port-used");
|
||||
|
||||
import { Cluster } from "../cluster";
|
||||
import { KubeAuthProxy } from "../kube-auth-proxy";
|
||||
import { getFreePort } from "../port";
|
||||
import { broadcastMessage } from "../../common/ipc";
|
||||
import { ChildProcess, spawn } from "child_process";
|
||||
import { bundledKubectlPath, Kubectl } from "../kubectl";
|
||||
@ -54,8 +53,7 @@ describe("kube auth proxy tests", () => {
|
||||
});
|
||||
|
||||
it("calling exit multiple times shouldn't throw", async () => {
|
||||
const port = await getFreePort();
|
||||
const kap = new KubeAuthProxy(new Cluster({ id: "foobar", kubeConfigPath: "fake-path.yml" }), port, {});
|
||||
const kap = new KubeAuthProxy(new Cluster({ id: "foobar", kubeConfigPath: "fake-path.yml" }), {});
|
||||
|
||||
kap.exit();
|
||||
kap.exit();
|
||||
@ -63,13 +61,11 @@ describe("kube auth proxy tests", () => {
|
||||
});
|
||||
|
||||
describe("spawn tests", () => {
|
||||
let port: number;
|
||||
let mockedCP: MockProxy<ChildProcess>;
|
||||
let listeners: Record<string, (...args: any[]) => void>;
|
||||
let proxy: KubeAuthProxy;
|
||||
|
||||
beforeEach(async () => {
|
||||
port = await getFreePort();
|
||||
mockedCP = mock<ChildProcess>();
|
||||
listeners = {};
|
||||
|
||||
@ -101,7 +97,7 @@ describe("kube auth proxy tests", () => {
|
||||
const cluster = new Cluster({ id: "foobar", kubeConfigPath: "fake-path.yml" });
|
||||
|
||||
jest.spyOn(cluster, "apiUrl", "get").mockReturnValue("https://fake.k8s.internal");
|
||||
proxy = new KubeAuthProxy(cluster, port, {});
|
||||
proxy = new KubeAuthProxy(cluster, {});
|
||||
});
|
||||
|
||||
it("should call spawn and broadcast errors", async () => {
|
||||
|
||||
@ -27,7 +27,6 @@ import { KubeconfigManager } from "../kubeconfig-manager";
|
||||
import mockFs from "mock-fs";
|
||||
import { Cluster } from "../cluster";
|
||||
import { ContextHandler } from "../context-handler";
|
||||
import { getFreePort } from "../port";
|
||||
import fse from "fs-extra";
|
||||
import { loadYaml } from "@kubernetes/client-node";
|
||||
import { Console } from "console";
|
||||
@ -75,8 +74,7 @@ describe("kubeconfig manager tests", () => {
|
||||
kubeConfigPath: "minikube-config.yml",
|
||||
});
|
||||
const contextHandler = new ContextHandler(cluster);
|
||||
const port = await getFreePort();
|
||||
const kubeConfManager = new KubeconfigManager(cluster, contextHandler, port);
|
||||
const kubeConfManager = new KubeconfigManager(cluster, contextHandler);
|
||||
|
||||
expect(logger.error).not.toBeCalled();
|
||||
expect(await kubeConfManager.getPath()).toBe(`tmp${path.sep}kubeconfig-foo`);
|
||||
@ -86,7 +84,7 @@ describe("kubeconfig manager tests", () => {
|
||||
const yml = loadYaml<any>(file.toString());
|
||||
|
||||
expect(yml["current-context"]).toBe("minikube");
|
||||
expect(yml["clusters"][0]["cluster"]["server"]).toBe(`http://127.0.0.1:${port}/foo`);
|
||||
expect(yml["clusters"][0]["cluster"]["server"].endsWith("/foo")).toBe(true);
|
||||
expect(yml["users"][0]["name"]).toBe("proxy");
|
||||
});
|
||||
|
||||
@ -97,8 +95,7 @@ describe("kubeconfig manager tests", () => {
|
||||
kubeConfigPath: "minikube-config.yml",
|
||||
});
|
||||
const contextHandler = new ContextHandler(cluster);
|
||||
const port = await getFreePort();
|
||||
const kubeConfManager = new KubeconfigManager(cluster, contextHandler, port);
|
||||
const kubeConfManager = new KubeconfigManager(cluster, contextHandler);
|
||||
const configPath = await kubeConfManager.getPath();
|
||||
|
||||
expect(await fse.pathExists(configPath)).toBe(true);
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
import request, { RequestPromiseOptions } from "request-promise-native";
|
||||
import { RequestPromiseOptions } from "request-promise-native";
|
||||
import { Cluster } from "../cluster";
|
||||
import { k8sRequest } from "../k8s-request";
|
||||
|
||||
export type ClusterDetectionResult = {
|
||||
value: string | number | boolean
|
||||
@ -7,11 +8,9 @@ export type ClusterDetectionResult = {
|
||||
};
|
||||
|
||||
export class BaseClusterDetector {
|
||||
cluster: Cluster;
|
||||
key: string;
|
||||
|
||||
constructor(cluster: Cluster) {
|
||||
this.cluster = cluster;
|
||||
constructor(public cluster: Cluster) {
|
||||
}
|
||||
|
||||
detect(): Promise<ClusterDetectionResult> {
|
||||
@ -19,16 +18,6 @@ export class BaseClusterDetector {
|
||||
}
|
||||
|
||||
protected async k8sRequest<T = any>(path: string, options: RequestPromiseOptions = {}): Promise<T> {
|
||||
const apiUrl = this.cluster.kubeProxyUrl + path;
|
||||
|
||||
return request(apiUrl, {
|
||||
json: true,
|
||||
timeout: 30000,
|
||||
...options,
|
||||
headers: {
|
||||
Host: `${this.cluster.id}.${new URL(this.cluster.kubeProxyUrl).host}`, // required in ClusterManager.getClusterForRequest()
|
||||
...(options.headers || {}),
|
||||
},
|
||||
});
|
||||
return k8sRequest(this.cluster, path, options);
|
||||
}
|
||||
}
|
||||
|
||||
@ -10,25 +10,12 @@ import { Singleton } from "../common/utils";
|
||||
import { catalogEntityRegistry } from "../common/catalog";
|
||||
import { KubernetesCluster } from "../common/catalog-entities/kubernetes-cluster";
|
||||
|
||||
const clusterOwnerRef = "ClusterManager";
|
||||
|
||||
export class ClusterManager extends Singleton {
|
||||
constructor(public readonly port: number) {
|
||||
constructor() {
|
||||
super();
|
||||
|
||||
// auto-init clusters
|
||||
reaction(() => ClusterStore.getInstance().enabledClustersList, (clusters) => {
|
||||
clusters.forEach((cluster) => {
|
||||
if (!cluster.initialized && !cluster.initializing) {
|
||||
logger.info(`[CLUSTER-MANAGER]: init cluster`, cluster.getMeta());
|
||||
cluster.init(port);
|
||||
}
|
||||
});
|
||||
|
||||
}, { fireImmediately: true });
|
||||
|
||||
reaction(() => toJS(ClusterStore.getInstance().enabledClustersList, { recurseEverything: true }), () => {
|
||||
this.updateCatalog(ClusterStore.getInstance().enabledClustersList);
|
||||
reaction(() => toJS(ClusterStore.getInstance().clustersList, { recurseEverything: true }), () => {
|
||||
this.updateCatalog(ClusterStore.getInstance().clustersList);
|
||||
}, { fireImmediately: true });
|
||||
|
||||
reaction(() => catalogEntityRegistry.getItemsForApiKind<KubernetesCluster>("entity.k8slens.dev/v1alpha1", "KubernetesCluster"), (entities) => {
|
||||
@ -84,8 +71,6 @@ export class ClusterManager extends Singleton {
|
||||
if (!cluster) {
|
||||
ClusterStore.getInstance().addCluster({
|
||||
id: entity.metadata.uid,
|
||||
enabled: true,
|
||||
ownerRef: clusterOwnerRef,
|
||||
preferences: {
|
||||
clusterName: entity.metadata.name
|
||||
},
|
||||
@ -93,8 +78,6 @@ export class ClusterManager extends Singleton {
|
||||
contextName: entity.spec.kubeconfigContext
|
||||
});
|
||||
} else {
|
||||
cluster.enabled = true;
|
||||
cluster.ownerRef ||= clusterOwnerRef;
|
||||
cluster.kubeConfigPath = entity.spec.kubeconfigPath;
|
||||
cluster.contextName = entity.spec.kubeconfigContext;
|
||||
|
||||
@ -108,7 +91,7 @@ export class ClusterManager extends Singleton {
|
||||
|
||||
protected onNetworkOffline() {
|
||||
logger.info("[CLUSTER-MANAGER]: network is offline");
|
||||
ClusterStore.getInstance().enabledClustersList.forEach((cluster) => {
|
||||
ClusterStore.getInstance().clustersList.forEach((cluster) => {
|
||||
if (!cluster.disconnected) {
|
||||
cluster.online = false;
|
||||
cluster.accessible = false;
|
||||
@ -119,7 +102,7 @@ export class ClusterManager extends Singleton {
|
||||
|
||||
protected onNetworkOnline() {
|
||||
logger.info("[CLUSTER-MANAGER]: network is online");
|
||||
ClusterStore.getInstance().enabledClustersList.forEach((cluster) => {
|
||||
ClusterStore.getInstance().clustersList.forEach((cluster) => {
|
||||
if (!cluster.disconnected) {
|
||||
cluster.refreshConnectionStatus().catch((e) => e);
|
||||
}
|
||||
|
||||
@ -1,15 +1,12 @@
|
||||
import { ipcMain } from "electron";
|
||||
import type { ClusterId, ClusterMetadata, ClusterModel, ClusterPreferences, ClusterPrometheusPreferences, UpdateClusterModel } from "../common/cluster-store";
|
||||
import type { IMetricsReqParams } from "../renderer/api/endpoints/metrics.api";
|
||||
import { action, comparer, computed, observable, reaction, toJS, when } from "mobx";
|
||||
import { apiKubePrefix } from "../common/vars";
|
||||
import { broadcastMessage, InvalidKubeconfigChannel, ClusterListNamespaceForbiddenChannel } from "../common/ipc";
|
||||
import { broadcastMessage, ClusterListNamespaceForbiddenChannel } from "../common/ipc";
|
||||
import { ContextHandler } from "./context-handler";
|
||||
import { AuthorizationV1Api, CoreV1Api, HttpError, KubeConfig, V1ResourceAttributes } from "@kubernetes/client-node";
|
||||
import { Kubectl } from "./kubectl";
|
||||
import { KubeconfigManager } from "./kubeconfig-manager";
|
||||
import { loadConfig, validateKubeConfig } from "../common/kube-helpers";
|
||||
import request, { RequestPromiseOptions } from "request-promise-native";
|
||||
import { apiResources, KubeApiResource } from "../common/rbac";
|
||||
import logger from "./logger";
|
||||
import { VersionDetector } from "./cluster-detectors/version-detector";
|
||||
@ -36,8 +33,6 @@ export type ClusterRefreshOptions = {
|
||||
};
|
||||
|
||||
export interface ClusterState {
|
||||
initialized: boolean;
|
||||
enabled: boolean;
|
||||
apiUrl: string;
|
||||
online: boolean;
|
||||
disconnected: boolean;
|
||||
@ -70,33 +65,13 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
* @internal
|
||||
*/
|
||||
public contextHandler: ContextHandler;
|
||||
/**
|
||||
* Owner reference
|
||||
*
|
||||
* If extension sets this it needs to also mark cluster as enabled on activate (or when added to a store)
|
||||
*/
|
||||
public ownerRef: string;
|
||||
protected kubeconfigManager: KubeconfigManager;
|
||||
protected eventDisposers: Function[] = [];
|
||||
protected activated = false;
|
||||
private resourceAccessStatuses: Map<KubeApiResource, boolean> = new Map();
|
||||
|
||||
whenInitialized = when(() => this.initialized);
|
||||
whenReady = when(() => this.ready);
|
||||
|
||||
/**
|
||||
* Is cluster object initializing on-going
|
||||
*
|
||||
* @observable
|
||||
*/
|
||||
@observable initializing = false;
|
||||
|
||||
/**
|
||||
* Is cluster object initialized
|
||||
*
|
||||
* @observable
|
||||
*/
|
||||
@observable initialized = false;
|
||||
/**
|
||||
* Kubeconfig context name
|
||||
*
|
||||
@ -119,19 +94,6 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
* @observable
|
||||
*/
|
||||
@observable apiUrl: string; // cluster server url
|
||||
/**
|
||||
* Internal authentication proxy URL
|
||||
*
|
||||
* @observable
|
||||
* @internal
|
||||
*/
|
||||
@observable kubeProxyUrl: string; // lens-proxy to kube-api url
|
||||
/**
|
||||
* Is cluster instance enabled (disabled clusters are currently hidden)
|
||||
*
|
||||
* @observable
|
||||
*/
|
||||
@observable enabled = false; // only enabled clusters are visible to users
|
||||
/**
|
||||
* Is cluster online
|
||||
*
|
||||
@ -260,27 +222,26 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
this.id = model.id;
|
||||
this.updateModel(model);
|
||||
|
||||
try {
|
||||
const kubeconfig = this.getKubeconfig();
|
||||
const error = validateKubeConfig(kubeconfig, this.contextName, { validateCluster: true, validateUser: false, validateExec: false});
|
||||
const kubeconfig = this.getKubeconfig();
|
||||
const error = validateKubeConfig(kubeconfig, this.contextName, { validateCluster: true, validateUser: false, validateExec: false});
|
||||
|
||||
if (error) {
|
||||
throw error;
|
||||
}
|
||||
|
||||
this.apiUrl = kubeconfig.getCluster(kubeconfig.getContextObject(this.contextName).cluster).server;
|
||||
} catch(err) {
|
||||
logger.error(err);
|
||||
logger.error(`[CLUSTER] Failed to load kubeconfig for the cluster '${this.name || this.contextName}' (context: ${this.contextName}, kubeconfig: ${this.kubeConfigPath}).`);
|
||||
broadcastMessage(InvalidKubeconfigChannel, model.id);
|
||||
if (error) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Is cluster managed by an extension
|
||||
*/
|
||||
get isManaged(): boolean {
|
||||
return !!this.ownerRef;
|
||||
this.apiUrl = kubeconfig.getCluster(kubeconfig.getContextObject(this.contextName).cluster).server;
|
||||
|
||||
if (ipcMain) {
|
||||
// for the time being, until renderer gets its own cluster type
|
||||
this.contextHandler = new ContextHandler(this);
|
||||
this.kubeconfigManager = new KubeconfigManager(this, this.contextHandler);
|
||||
|
||||
logger.debug(`[CLUSTER]: Cluster init success`, {
|
||||
id: this.id,
|
||||
context: this.contextName,
|
||||
apiUrl: this.apiUrl
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@ -309,44 +270,11 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
this.metadata = model.metadata;
|
||||
}
|
||||
|
||||
if (model.ownerRef) {
|
||||
this.ownerRef = model.ownerRef;
|
||||
}
|
||||
|
||||
if (model.accessibleNamespaces) {
|
||||
this.accessibleNamespaces = model.accessibleNamespaces;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize a cluster (can be done only in main process)
|
||||
*
|
||||
* @param port port where internal auth proxy is listening
|
||||
* @internal
|
||||
*/
|
||||
@action
|
||||
async init(port: number) {
|
||||
try {
|
||||
this.initializing = true;
|
||||
this.contextHandler = new ContextHandler(this);
|
||||
this.kubeconfigManager = new KubeconfigManager(this, this.contextHandler, port);
|
||||
this.kubeProxyUrl = `http://localhost:${port}${apiKubePrefix}`;
|
||||
this.initialized = true;
|
||||
logger.info(`[CLUSTER]: "${this.contextName}" init success`, {
|
||||
id: this.id,
|
||||
context: this.contextName,
|
||||
apiUrl: this.apiUrl
|
||||
});
|
||||
} catch (err) {
|
||||
logger.error(`[CLUSTER]: init failed: ${err}`, {
|
||||
id: this.id,
|
||||
error: err,
|
||||
});
|
||||
} finally {
|
||||
this.initializing = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @internal
|
||||
*/
|
||||
@ -386,7 +314,6 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
return this.pushState();
|
||||
}
|
||||
logger.info(`[CLUSTER]: activate`, this.getMeta());
|
||||
await this.whenInitialized;
|
||||
|
||||
if (!this.eventDisposers.length) {
|
||||
this.bindEvents();
|
||||
@ -450,7 +377,6 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
@action
|
||||
async refresh(opts: ClusterRefreshOptions = {}) {
|
||||
logger.info(`[CLUSTER]: refresh`, this.getMeta());
|
||||
await this.whenInitialized;
|
||||
await this.refreshConnectionStatus();
|
||||
|
||||
if (this.accessible) {
|
||||
@ -527,34 +453,6 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
return this.kubeconfigManager.getPath();
|
||||
}
|
||||
|
||||
protected async k8sRequest<T = any>(path: string, options: RequestPromiseOptions = {}): Promise<T> {
|
||||
options.headers ??= {};
|
||||
options.json ??= true;
|
||||
options.timeout ??= 30000;
|
||||
options.headers.Host = `${this.id}.${new URL(this.kubeProxyUrl).host}`; // required in ClusterManager.getClusterForRequest()
|
||||
|
||||
return request(this.kubeProxyUrl + path, options);
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param prometheusPath path to prometheus service
|
||||
* @param queryParams query parameters
|
||||
* @internal
|
||||
*/
|
||||
getMetrics(prometheusPath: string, queryParams: IMetricsReqParams & { query: string }) {
|
||||
const prometheusPrefix = this.preferences.prometheus?.prefix || "";
|
||||
const metricsPath = `/api/v1/namespaces/${prometheusPath}/proxy${prometheusPrefix}/api/v1/query_range`;
|
||||
|
||||
return this.k8sRequest(metricsPath, {
|
||||
timeout: 0,
|
||||
resolveWithFullResponse: false,
|
||||
json: true,
|
||||
method: "POST",
|
||||
form: queryParams,
|
||||
});
|
||||
}
|
||||
|
||||
protected async getConnectionStatus(): Promise<ClusterStatus> {
|
||||
try {
|
||||
const versionDetector = new VersionDetector(this);
|
||||
@ -647,7 +545,6 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
workspace: this.workspace,
|
||||
preferences: this.preferences,
|
||||
metadata: this.metadata,
|
||||
ownerRef: this.ownerRef,
|
||||
accessibleNamespaces: this.accessibleNamespaces,
|
||||
};
|
||||
|
||||
@ -661,8 +558,6 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
*/
|
||||
getState(): ClusterState {
|
||||
const state: ClusterState = {
|
||||
initialized: this.initialized,
|
||||
enabled: this.enabled,
|
||||
apiUrl: this.apiUrl,
|
||||
online: this.online,
|
||||
ready: this.ready,
|
||||
@ -702,7 +597,6 @@ export class Cluster implements ClusterModel, ClusterState {
|
||||
return {
|
||||
id: this.id,
|
||||
name: this.contextName,
|
||||
initialized: this.initialized,
|
||||
ready: this.ready,
|
||||
online: this.online,
|
||||
accessible: this.accessible,
|
||||
|
||||
@ -6,7 +6,6 @@ import url, { UrlWithStringQuery } from "url";
|
||||
import { CoreV1Api } from "@kubernetes/client-node";
|
||||
import { prometheusProviders } from "../common/prometheus-providers";
|
||||
import logger from "./logger";
|
||||
import { getFreePort } from "./port";
|
||||
import { KubeAuthProxy } from "./kube-auth-proxy";
|
||||
|
||||
export class ContextHandler {
|
||||
@ -77,10 +76,10 @@ export class ContextHandler {
|
||||
}
|
||||
|
||||
async resolveAuthProxyUrl() {
|
||||
const proxyPort = await this.ensurePort();
|
||||
await this.ensureServer();
|
||||
const path = this.clusterUrl.path !== "/" ? this.clusterUrl.path : "";
|
||||
|
||||
return `http://127.0.0.1:${proxyPort}${path}`;
|
||||
return `http://127.0.0.1:${this.kubeAuthProxy.port}${path}`;
|
||||
}
|
||||
|
||||
async getApiTarget(isWatchRequest = false): Promise<httpProxy.ServerOptions> {
|
||||
@ -110,23 +109,14 @@ export class ContextHandler {
|
||||
};
|
||||
}
|
||||
|
||||
async ensurePort(): Promise<number> {
|
||||
if (!this.proxyPort) {
|
||||
this.proxyPort = await getFreePort();
|
||||
}
|
||||
|
||||
return this.proxyPort;
|
||||
}
|
||||
|
||||
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.kubeAuthProxy = new KubeAuthProxy(this.cluster, this.proxyPort, proxyEnv);
|
||||
this.kubeAuthProxy = new KubeAuthProxy(this.cluster, proxyEnv);
|
||||
await this.kubeAuthProxy.run();
|
||||
}
|
||||
}
|
||||
|
||||
@ -7,11 +7,10 @@ import * as LensExtensions from "../extensions/core-api";
|
||||
import { app, autoUpdater, ipcMain, dialog, powerMonitor } from "electron";
|
||||
import { appName, isMac, productName } from "../common/vars";
|
||||
import path from "path";
|
||||
import { LensProxy } from "./lens-proxy";
|
||||
import { LensProxy } from "./proxy/lens-proxy";
|
||||
import { WindowManager } from "./window-manager";
|
||||
import { ClusterManager } from "./cluster-manager";
|
||||
import { shellSync } from "./shell-sync";
|
||||
import { getFreePort } from "./port";
|
||||
import { mangleProxyEnv } from "./proxy-env";
|
||||
import { registerFileProtocol } from "../common/register-protocol";
|
||||
import logger from "./logger";
|
||||
@ -34,6 +33,7 @@ import { catalogEntityRegistry } from "../common/catalog";
|
||||
import { HotbarStore } from "../common/hotbar-store";
|
||||
import { HelmRepoManager } from "./helm/helm-repo-manager";
|
||||
import { KubeconfigSyncManager } from "./catalog-sources";
|
||||
import { handleWsUpgrade } from "./proxy/ws-upgrade";
|
||||
|
||||
const workingDir = path.join(app.getPath("appData"), appName);
|
||||
|
||||
@ -118,45 +118,33 @@ app.on("ready", async () => {
|
||||
filesystemStore.load(),
|
||||
]);
|
||||
|
||||
try {
|
||||
logger.info("🔑 Getting free port for LensProxy server");
|
||||
const proxyPort = await getFreePort();
|
||||
const lensProxy = LensProxy.createInstance(handleWsUpgrade);
|
||||
|
||||
// create cluster manager
|
||||
ClusterManager.createInstance(proxyPort);
|
||||
} catch (error) {
|
||||
logger.error(error);
|
||||
dialog.showErrorBox("Lens Error", "Could not find a free port for the cluster proxy");
|
||||
app.exit();
|
||||
}
|
||||
|
||||
const clusterManager = ClusterManager.getInstance();
|
||||
|
||||
// create kubeconfig sync manager
|
||||
ClusterManager.createInstance();
|
||||
KubeconfigSyncManager.createInstance().startSync();
|
||||
|
||||
// run proxy
|
||||
try {
|
||||
logger.info("🔌 Starting LensProxy");
|
||||
// eslint-disable-next-line unused-imports/no-unused-vars-ts
|
||||
LensProxy.createInstance(clusterManager.port).listen();
|
||||
await lensProxy.listen();
|
||||
} catch (error) {
|
||||
logger.error(`Could not start proxy (127.0.0:${clusterManager.port}): ${error?.message}`);
|
||||
dialog.showErrorBox("Lens Error", `Could not start proxy (127.0.0:${clusterManager.port}): ${error?.message || "unknown error"}`);
|
||||
dialog.showErrorBox("Lens Error", `Could not start proxy: ${error?.message || "unknown error"}`);
|
||||
app.exit();
|
||||
}
|
||||
|
||||
// test proxy connection
|
||||
try {
|
||||
logger.info("🔎 Testing LensProxy connection ...");
|
||||
const versionFromProxy = await getAppVersionFromProxyServer(clusterManager.port);
|
||||
const versionFromProxy = await getAppVersionFromProxyServer(lensProxy.port);
|
||||
|
||||
if (getAppVersion() !== versionFromProxy) {
|
||||
logger.error(`Proxy server responded with invalid response`);
|
||||
logger.error("Proxy server responded with invalid response");
|
||||
app.exit();
|
||||
} else {
|
||||
logger.info("⚡ LensProxy connection OK");
|
||||
}
|
||||
logger.info("⚡ LensProxy connection OK");
|
||||
} catch (error) {
|
||||
logger.error("Checking proxy server connection failed", error);
|
||||
logger.error(`🛑 LensProxy: failed connection test: ${error}`);
|
||||
app.exit();
|
||||
}
|
||||
|
||||
const extensionDiscovery = ExtensionDiscovery.createInstance();
|
||||
@ -169,7 +157,7 @@ app.on("ready", async () => {
|
||||
const startHidden = process.argv.includes("--hidden") || (isMac && app.getLoginItemSettings().wasOpenedAsHidden);
|
||||
|
||||
logger.info("🖥️ Starting WindowManager");
|
||||
const windowManager = WindowManager.createInstance(clusterManager.port);
|
||||
const windowManager = WindowManager.createInstance();
|
||||
|
||||
installDeveloperTools();
|
||||
|
||||
|
||||
29
src/main/k8s-request.ts
Normal file
29
src/main/k8s-request.ts
Normal file
@ -0,0 +1,29 @@
|
||||
import request, { RequestPromiseOptions } from "request-promise-native";
|
||||
import { apiKubePrefix } from "../common/vars";
|
||||
import { IMetricsReqParams } from "../renderer/api/endpoints/metrics.api";
|
||||
import { LensProxy } from "./proxy/lens-proxy";
|
||||
import { Cluster } from "./cluster";
|
||||
|
||||
export async function k8sRequest<T = any>(cluster: Cluster, path: string, options: RequestPromiseOptions = {}): Promise<T> {
|
||||
const kubeProxyUrl = `http://localhost:${LensProxy.getInstance().port}${apiKubePrefix}`;
|
||||
|
||||
options.headers ??= {};
|
||||
options.json ??= true;
|
||||
options.timeout ??= 30000;
|
||||
options.headers.Host = `${cluster.id}.${new URL(kubeProxyUrl).host}`; // required in ClusterManager.getClusterForRequest()
|
||||
|
||||
return request(kubeProxyUrl + path, options);
|
||||
}
|
||||
|
||||
export async function getMetrics(cluster: Cluster, prometheusPath: string, queryParams: IMetricsReqParams & { query: string }): Promise<any> {
|
||||
const prometheusPrefix = cluster.preferences.prometheus?.prefix || "";
|
||||
const metricsPath = `/api/v1/namespaces/${prometheusPath}/proxy${prometheusPrefix}/api/v1/query_range`;
|
||||
|
||||
return k8sRequest(cluster, metricsPath, {
|
||||
timeout: 0,
|
||||
resolveWithFullResponse: false,
|
||||
json: true,
|
||||
method: "POST",
|
||||
form: queryParams,
|
||||
});
|
||||
}
|
||||
@ -5,24 +5,30 @@ import type { Cluster } from "./cluster";
|
||||
import { Kubectl } from "./kubectl";
|
||||
import logger from "./logger";
|
||||
import * as url from "url";
|
||||
import { getPortFrom } from "./utils/get-port";
|
||||
|
||||
export interface KubeAuthProxyLog {
|
||||
data: string;
|
||||
error?: boolean; // stream=stderr
|
||||
}
|
||||
|
||||
const startingServeRegex = /^starting to serve on (?<address>.+)/i;
|
||||
|
||||
export class KubeAuthProxy {
|
||||
public lastError: string;
|
||||
|
||||
public get port(): number {
|
||||
return this._port;
|
||||
}
|
||||
|
||||
protected _port: number;
|
||||
protected cluster: Cluster;
|
||||
protected env: NodeJS.ProcessEnv = null;
|
||||
protected proxyProcess: ChildProcess;
|
||||
protected port: number;
|
||||
protected kubectl: Kubectl;
|
||||
|
||||
constructor(cluster: Cluster, port: number, env: NodeJS.ProcessEnv) {
|
||||
constructor(cluster: Cluster, env: NodeJS.ProcessEnv) {
|
||||
this.env = env;
|
||||
this.port = port;
|
||||
this.cluster = cluster;
|
||||
this.kubectl = Kubectl.bundled();
|
||||
}
|
||||
@ -39,7 +45,7 @@ export class KubeAuthProxy {
|
||||
const proxyBin = await this.kubectl.getPath();
|
||||
const args = [
|
||||
"proxy",
|
||||
"-p", `${this.port}`,
|
||||
"-p", "0",
|
||||
"--kubeconfig", `${this.cluster.kubeConfigPath}`,
|
||||
"--context", `${this.cluster.contextName}`,
|
||||
"--accept-hosts", this.acceptHosts,
|
||||
@ -50,6 +56,7 @@ export class KubeAuthProxy {
|
||||
args.push("-v", "9");
|
||||
}
|
||||
logger.debug(`spawning kubectl proxy with args: ${args}`);
|
||||
|
||||
this.proxyProcess = spawn(proxyBin, args, { env: this.env, });
|
||||
this.proxyProcess.on("error", (error) => {
|
||||
this.sendIpcLogMessage({ data: error.message, error: true });
|
||||
@ -61,20 +68,20 @@ export class KubeAuthProxy {
|
||||
this.exit();
|
||||
});
|
||||
|
||||
this.proxyProcess.stdout.on("data", (data) => {
|
||||
let logItem = data.toString();
|
||||
|
||||
if (logItem.startsWith("Starting to serve on")) {
|
||||
logItem = "Authentication proxy started\n";
|
||||
}
|
||||
this.sendIpcLogMessage({ data: logItem });
|
||||
});
|
||||
|
||||
this.proxyProcess.stderr.on("data", (data) => {
|
||||
this.lastError = this.parseError(data.toString());
|
||||
this.sendIpcLogMessage({ data: data.toString(), error: true });
|
||||
});
|
||||
|
||||
this._port = await getPortFrom(this.proxyProcess.stdout, {
|
||||
lineRegex: startingServeRegex,
|
||||
onFind: () => this.sendIpcLogMessage({ data: "Authentication proxy started\n" }),
|
||||
});
|
||||
|
||||
this.proxyProcess.stdout.on("data", (data: any) => {
|
||||
this.sendIpcLogMessage({ data: data.toString() });
|
||||
});
|
||||
|
||||
return waitUntilUsed(this.port, 500, 10000);
|
||||
}
|
||||
|
||||
@ -96,7 +103,7 @@ export class KubeAuthProxy {
|
||||
return errorMsg;
|
||||
}
|
||||
|
||||
protected async sendIpcLogMessage(res: KubeAuthProxyLog) {
|
||||
protected sendIpcLogMessage(res: KubeAuthProxyLog) {
|
||||
const channel = `kube-auth:${this.cluster.id}`;
|
||||
|
||||
logger.info(`[KUBE-AUTH]: out-channel "${channel}"`, { ...res, meta: this.cluster.getMeta() });
|
||||
|
||||
@ -6,12 +6,13 @@ import path from "path";
|
||||
import fs from "fs-extra";
|
||||
import { dumpConfigYaml, loadConfig } from "../common/kube-helpers";
|
||||
import logger from "./logger";
|
||||
import { LensProxy } from "./proxy/lens-proxy";
|
||||
|
||||
export class KubeconfigManager {
|
||||
protected configDir = app.getPath("temp");
|
||||
protected tempFile: string = null;
|
||||
|
||||
constructor(protected cluster: Cluster, protected contextHandler: ContextHandler, protected port: number) { }
|
||||
constructor(protected cluster: Cluster, protected contextHandler: ContextHandler) { }
|
||||
|
||||
async getPath(): Promise<string> {
|
||||
if (this.tempFile === undefined) {
|
||||
@ -46,15 +47,15 @@ export class KubeconfigManager {
|
||||
|
||||
protected async init() {
|
||||
try {
|
||||
await this.contextHandler.ensurePort();
|
||||
await this.contextHandler.ensureServer();
|
||||
this.tempFile = await this.createProxyKubeconfig();
|
||||
} catch (err) {
|
||||
logger.error(`Failed to created temp config for auth-proxy`, { err });
|
||||
}
|
||||
}
|
||||
|
||||
protected resolveProxyUrl() {
|
||||
return `http://127.0.0.1:${this.port}/${this.cluster.id}`;
|
||||
get resolveProxyUrl() {
|
||||
return `http://127.0.0.1:${LensProxy.getInstance().port}/${this.cluster.id}`;
|
||||
}
|
||||
|
||||
/**
|
||||
@ -71,7 +72,7 @@ export class KubeconfigManager {
|
||||
clusters: [
|
||||
{
|
||||
name: contextName,
|
||||
server: this.resolveProxyUrl(),
|
||||
server: this.resolveProxyUrl,
|
||||
skipTLSVerify: undefined,
|
||||
}
|
||||
],
|
||||
|
||||
@ -1,25 +0,0 @@
|
||||
import net, { AddressInfo } from "net";
|
||||
import logger from "./logger";
|
||||
|
||||
// todo: check https://github.com/http-party/node-portfinder ?
|
||||
|
||||
export async function getFreePort(): Promise<number> {
|
||||
logger.debug("Lookup new free port..");
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const server = net.createServer();
|
||||
|
||||
server.unref();
|
||||
server.on("listening", () => {
|
||||
const port = (server.address() as AddressInfo).port;
|
||||
|
||||
server.close(() => resolve(port));
|
||||
logger.debug(`New port found: ${port}`);
|
||||
});
|
||||
server.on("error", error => {
|
||||
logger.error(`Can't resolve new port: "${error}"`);
|
||||
reject(error);
|
||||
});
|
||||
server.listen({ host: "127.0.0.1", port: 0 });
|
||||
});
|
||||
}
|
||||
@ -1,33 +0,0 @@
|
||||
import { EventEmitter } from "events";
|
||||
import { getFreePort } from "./port";
|
||||
|
||||
let newPort = 0;
|
||||
|
||||
jest.mock("net", () => {
|
||||
return {
|
||||
createServer() {
|
||||
return new class MockServer extends EventEmitter {
|
||||
listen = jest.fn(() => {
|
||||
this.emit("listening");
|
||||
|
||||
return this;
|
||||
});
|
||||
address = () => {
|
||||
newPort = Math.round(Math.random() * 10000);
|
||||
|
||||
return {
|
||||
port: newPort
|
||||
};
|
||||
};
|
||||
unref = jest.fn();
|
||||
close = jest.fn(cb => cb());
|
||||
};
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
describe("getFreePort", () => {
|
||||
it("finds the next free port", async () => {
|
||||
return expect(getFreePort()).resolves.toEqual(newPort);
|
||||
});
|
||||
});
|
||||
2
src/main/proxy/index.ts
Normal file
2
src/main/proxy/index.ts
Normal file
@ -0,0 +1,2 @@
|
||||
// Don't export the contents here
|
||||
// It will break the extension webpack
|
||||
@ -3,45 +3,30 @@ import http from "http";
|
||||
import spdy from "spdy";
|
||||
import httpProxy from "http-proxy";
|
||||
import url from "url";
|
||||
import * as WebSocket from "ws";
|
||||
import { apiPrefix, apiKubePrefix } from "../common/vars";
|
||||
import { Router } from "./router";
|
||||
import { ContextHandler } from "./context-handler";
|
||||
import logger from "./logger";
|
||||
import { NodeShellSession, LocalShellSession } from "./shell-session";
|
||||
import { Singleton } from "../common/utils";
|
||||
import { ClusterManager } from "./cluster-manager";
|
||||
import { apiPrefix, apiKubePrefix } from "../../common/vars";
|
||||
import { Router } from "../router";
|
||||
import { ContextHandler } from "../context-handler";
|
||||
import logger from "../logger";
|
||||
import { Singleton } from "../../common/utils";
|
||||
import { ClusterManager } from "../cluster-manager";
|
||||
|
||||
type WSUpgradeHandler = (req: http.IncomingMessage, socket: net.Socket, head: Buffer) => void;
|
||||
|
||||
export class LensProxy extends Singleton {
|
||||
protected origin: string;
|
||||
protected proxyServer: http.Server;
|
||||
protected router: Router;
|
||||
protected router = new Router();
|
||||
protected closed = false;
|
||||
protected retryCounters = new Map<string, number>();
|
||||
|
||||
constructor(protected port: number) {
|
||||
public port: number;
|
||||
|
||||
constructor(handleWsUpgrade: WSUpgradeHandler) {
|
||||
super();
|
||||
|
||||
this.origin = `http://localhost:${port}`;
|
||||
this.router = new Router();
|
||||
}
|
||||
|
||||
listen(port = this.port): this {
|
||||
this.proxyServer = this.buildCustomProxy().listen(port, "127.0.0.1");
|
||||
logger.info(`[LENS-PROXY]: Proxy server has started at ${this.origin}`);
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
close() {
|
||||
logger.info("Closing proxy server");
|
||||
this.proxyServer.close();
|
||||
this.closed = true;
|
||||
}
|
||||
|
||||
protected buildCustomProxy(): http.Server {
|
||||
const proxy = this.createProxy();
|
||||
const spdyProxy = spdy.createServer({
|
||||
|
||||
this.proxyServer = spdy.createServer({
|
||||
spdy: {
|
||||
plain: true,
|
||||
protocols: ["http/1.1", "spdy/3.1"]
|
||||
@ -50,18 +35,51 @@ export class LensProxy extends Singleton {
|
||||
this.handleRequest(proxy, req, res);
|
||||
});
|
||||
|
||||
spdyProxy.on("upgrade", (req: http.IncomingMessage, socket: net.Socket, head: Buffer) => {
|
||||
if (req.url.startsWith(`${apiPrefix}?`)) {
|
||||
this.handleWsUpgrade(req, socket, head);
|
||||
} else {
|
||||
this.handleProxyUpgrade(proxy, req, socket, head);
|
||||
}
|
||||
});
|
||||
spdyProxy.on("error", (err) => {
|
||||
logger.error("proxy error", err);
|
||||
});
|
||||
this.proxyServer
|
||||
.on("upgrade", (req: http.IncomingMessage, socket: net.Socket, head: Buffer) => {
|
||||
if (req.url.startsWith(`${apiPrefix}?`)) {
|
||||
handleWsUpgrade(req, socket, head);
|
||||
} else {
|
||||
this.handleProxyUpgrade(proxy, req, socket, head);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
return spdyProxy;
|
||||
/**
|
||||
* Starts the lens proxy.
|
||||
* @resolves After the server is listening
|
||||
* @rejects if there is an error before that happens
|
||||
*/
|
||||
listen(): Promise<void> {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
this.proxyServer.listen(0, "127.0.0.1");
|
||||
|
||||
this.proxyServer
|
||||
.once("listening", () => {
|
||||
this.proxyServer.removeAllListeners("error"); // don't reject the promise
|
||||
|
||||
const { address, port } = this.proxyServer.address() as net.AddressInfo;
|
||||
|
||||
logger.info(`[LENS-PROXY]: Proxy server has started at ${address}:${port}`);
|
||||
|
||||
this.proxyServer.on("error", (error) => {
|
||||
logger.info(`[LENS-PROXY]: Subsequent error: ${error}`);
|
||||
});
|
||||
|
||||
this.port = port;
|
||||
resolve();
|
||||
})
|
||||
.once("error", (error) => {
|
||||
logger.info(`[LENS-PROXY]: Proxy server failed to start: ${error}`);
|
||||
reject(error);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
close() {
|
||||
logger.info("Closing proxy server");
|
||||
this.proxyServer.close();
|
||||
this.closed = true;
|
||||
}
|
||||
|
||||
protected async handleProxyUpgrade(proxy: httpProxy, req: http.IncomingMessage, socket: net.Socket, head: Buffer) {
|
||||
@ -166,21 +184,6 @@ export class LensProxy extends Singleton {
|
||||
return proxy;
|
||||
}
|
||||
|
||||
protected createWsListener(): WebSocket.Server {
|
||||
const ws = new WebSocket.Server({ noServer: true });
|
||||
|
||||
return ws.on("connection", ((socket: WebSocket, req: http.IncomingMessage) => {
|
||||
const cluster = ClusterManager.getInstance().getClusterForRequest(req);
|
||||
const nodeParam = url.parse(req.url, true).query["node"]?.toString();
|
||||
const shell = nodeParam
|
||||
? new NodeShellSession(socket, cluster, nodeParam)
|
||||
: new LocalShellSession(socket, cluster);
|
||||
|
||||
shell.open()
|
||||
.catch(error => logger.error(`[SHELL-SESSION]: failed to open: ${error}`, { error }));
|
||||
}));
|
||||
}
|
||||
|
||||
protected async getProxyTarget(req: http.IncomingMessage, contextHandler: ContextHandler): Promise<httpProxy.ServerOptions> {
|
||||
if (req.url.startsWith(apiKubePrefix)) {
|
||||
delete req.headers.authorization;
|
||||
@ -211,12 +214,4 @@ export class LensProxy extends Singleton {
|
||||
}
|
||||
this.router.route(cluster, req, res);
|
||||
}
|
||||
|
||||
protected async handleWsUpgrade(req: http.IncomingMessage, socket: net.Socket, head: Buffer) {
|
||||
const wsServer = this.createWsListener();
|
||||
|
||||
wsServer.handleUpgrade(req, socket, head, (con) => {
|
||||
wsServer.emit("connection", con, req);
|
||||
});
|
||||
}
|
||||
}
|
||||
36
src/main/proxy/ws-upgrade.ts
Normal file
36
src/main/proxy/ws-upgrade.ts
Normal file
@ -0,0 +1,36 @@
|
||||
/**
|
||||
* This file is here so that the "../shell-session" import can be injected into
|
||||
* LensProxy at creation time. So that the `pty.node` extension isn't loaded
|
||||
* into Lens Extension webpack bundle.
|
||||
*/
|
||||
|
||||
import * as WebSocket from "ws";
|
||||
import http from "http";
|
||||
import net from "net";
|
||||
import url from "url";
|
||||
import { NodeShellSession, LocalShellSession } from "../shell-session";
|
||||
import { ClusterManager } from "../cluster-manager";
|
||||
import logger from "../logger";
|
||||
|
||||
function createWsListener(): WebSocket.Server {
|
||||
const ws = new WebSocket.Server({ noServer: true });
|
||||
|
||||
return ws.on("connection", ((socket: WebSocket, req: http.IncomingMessage) => {
|
||||
const cluster = ClusterManager.getInstance().getClusterForRequest(req);
|
||||
const nodeParam = url.parse(req.url, true).query["node"]?.toString();
|
||||
const shell = nodeParam
|
||||
? new NodeShellSession(socket, cluster, nodeParam)
|
||||
: new LocalShellSession(socket, cluster);
|
||||
|
||||
shell.open()
|
||||
.catch(error => logger.error(`[SHELL-SESSION]: failed to open: ${error}`, { error }));
|
||||
}));
|
||||
}
|
||||
|
||||
export async function handleWsUpgrade(req: http.IncomingMessage, socket: net.Socket, head: Buffer) {
|
||||
const wsServer = createWsListener();
|
||||
|
||||
wsServer.handleUpgrade(req, socket, head, (con) => {
|
||||
wsServer.emit("connection", con, req);
|
||||
});
|
||||
}
|
||||
@ -4,6 +4,7 @@ import { LensApi } from "../lens-api";
|
||||
import { Cluster, ClusterMetadataKey } from "../cluster";
|
||||
import { ClusterPrometheusMetadata } from "../../common/cluster-store";
|
||||
import logger from "../logger";
|
||||
import { getMetrics } from "../k8s-request";
|
||||
|
||||
export type IMetricsQuery = string | string[] | {
|
||||
[metricName: string]: string;
|
||||
@ -22,7 +23,7 @@ async function loadMetrics(promQueries: string[], cluster: Cluster, prometheusPa
|
||||
async function loadMetricHelper(): Promise<any> {
|
||||
for (const [attempt, lastAttempt] of ATTEMPTS.entries()) { // retry
|
||||
try {
|
||||
return await cluster.getMetrics(prometheusPath, { query, ...queryParams });
|
||||
return await getMetrics(cluster, prometheusPath, { query, ...queryParams });
|
||||
} catch (error) {
|
||||
if (lastAttempt || (error?.statusCode >= 400 && error?.statusCode < 500)) {
|
||||
logger.error("[Metrics]: metrics not available", { error });
|
||||
|
||||
@ -2,48 +2,58 @@ import { LensApiRequest } from "../router";
|
||||
import { LensApi } from "../lens-api";
|
||||
import { spawn, ChildProcessWithoutNullStreams } from "child_process";
|
||||
import { Kubectl } from "../kubectl";
|
||||
import { getFreePort } from "../port";
|
||||
import { shell } from "electron";
|
||||
import * as tcpPortUsed from "tcp-port-used";
|
||||
import logger from "../logger";
|
||||
import { getPortFrom } from "../utils/get-port";
|
||||
|
||||
interface PortForwardArgs {
|
||||
clusterId: string;
|
||||
kind: string;
|
||||
namespace: string;
|
||||
name: string;
|
||||
port: string;
|
||||
}
|
||||
|
||||
const internalPortRegex = /^forwarding from (?<address>.+) ->/i;
|
||||
|
||||
class PortForward {
|
||||
public static portForwards: PortForward[] = [];
|
||||
|
||||
static getPortforward(forward: {clusterId: string; kind: string; name: string; namespace: string; port: string}) {
|
||||
return PortForward.portForwards.find((pf) => {
|
||||
return (
|
||||
pf.clusterId == forward.clusterId &&
|
||||
pf.kind == forward.kind &&
|
||||
pf.name == forward.name &&
|
||||
pf.namespace == forward.namespace &&
|
||||
pf.port == forward.port
|
||||
);
|
||||
});
|
||||
static getPortforward(forward: PortForwardArgs) {
|
||||
return PortForward.portForwards.find((pf) => (
|
||||
pf.clusterId == forward.clusterId &&
|
||||
pf.kind == forward.kind &&
|
||||
pf.name == forward.name &&
|
||||
pf.namespace == forward.namespace &&
|
||||
pf.port == forward.port
|
||||
));
|
||||
}
|
||||
|
||||
public clusterId: string;
|
||||
public process: ChildProcessWithoutNullStreams;
|
||||
public kubeConfig: string;
|
||||
public clusterId: string;
|
||||
public kind: string;
|
||||
public namespace: string;
|
||||
public name: string;
|
||||
public port: string;
|
||||
public localPort: number;
|
||||
public internalPort?: number;
|
||||
|
||||
constructor(obj: any) {
|
||||
Object.assign(this, obj);
|
||||
constructor(public kubeConfig: string, args: PortForwardArgs) {
|
||||
this.clusterId = args.clusterId;
|
||||
this.kind = args.kind;
|
||||
this.namespace = args.namespace;
|
||||
this.name = args.name;
|
||||
this.port = args.port;
|
||||
}
|
||||
|
||||
public async start() {
|
||||
this.localPort = await getFreePort();
|
||||
const kubectlBin = await Kubectl.bundled().getPath();
|
||||
const args = [
|
||||
"--kubeconfig", this.kubeConfig,
|
||||
"port-forward",
|
||||
"-n", this.namespace,
|
||||
`${this.kind}/${this.name}`,
|
||||
`${this.localPort}:${this.port}`
|
||||
`:${this.port}`
|
||||
];
|
||||
|
||||
this.process = spawn(kubectlBin, args, {
|
||||
@ -58,8 +68,12 @@ class PortForward {
|
||||
}
|
||||
});
|
||||
|
||||
this.internalPort = await getPortFrom(this.process.stdout, {
|
||||
lineRegex: internalPortRegex,
|
||||
});
|
||||
|
||||
try {
|
||||
await tcpPortUsed.waitUntilUsed(this.localPort, 500, 15000);
|
||||
await tcpPortUsed.waitUntilUsed(this.internalPort, 500, 15000);
|
||||
|
||||
return true;
|
||||
} catch (error) {
|
||||
@ -70,7 +84,14 @@ class PortForward {
|
||||
}
|
||||
|
||||
public open() {
|
||||
shell.openExternal(`http://localhost:${this.localPort}`);
|
||||
shell.openExternal(`http://localhost:${this.internalPort}`)
|
||||
.catch(error => logger.error(`[PORT-FORWARD]: failed to open external shell: ${error}`, {
|
||||
clusterId: this.clusterId,
|
||||
port: this.port,
|
||||
kind: this.kind,
|
||||
namespace: this.namespace,
|
||||
name: this.name,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
@ -86,13 +107,12 @@ class PortForwardRoute extends LensApi {
|
||||
|
||||
if (!portForward) {
|
||||
logger.info(`Creating a new port-forward ${namespace}/${resourceType}/${resourceName}:${port}`);
|
||||
portForward = new PortForward({
|
||||
portForward = new PortForward(await cluster.getProxyKubeconfigPath(), {
|
||||
clusterId: cluster.id,
|
||||
kind: resourceType,
|
||||
namespace,
|
||||
name: resourceName,
|
||||
port,
|
||||
kubeConfig: await cluster.getProxyKubeconfigPath()
|
||||
});
|
||||
const started = await portForward.start();
|
||||
|
||||
|
||||
51
src/main/utils/get-port.ts
Normal file
51
src/main/utils/get-port.ts
Normal file
@ -0,0 +1,51 @@
|
||||
import { Readable } from "stream";
|
||||
import URLParse from "url-parse";
|
||||
|
||||
interface GetPortArgs {
|
||||
/**
|
||||
* Should be case insensitive
|
||||
* Must have a named matching group called `address`
|
||||
*/
|
||||
lineRegex: RegExp;
|
||||
/**
|
||||
* Called when the port is found
|
||||
*/
|
||||
onFind?: () => void;
|
||||
/**
|
||||
* Timeout for how long to wait for the port.
|
||||
* Default: 5s
|
||||
*/
|
||||
timeout?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse lines from `stream` (assumes data comes in lines) to find the port
|
||||
* which the source of the stream is watching on.
|
||||
* @param stream A readable stream to match lines against
|
||||
* @param args The args concerning the stream
|
||||
* @returns A Promise for port number
|
||||
*/
|
||||
export function getPortFrom(stream: Readable, args: GetPortArgs): Promise<number> {
|
||||
return new Promise<number>((resolve, reject) => {
|
||||
const handler = (data: any) => {
|
||||
const logItem: string = data.toString();
|
||||
const match = logItem.match(args.lineRegex);
|
||||
|
||||
if (match) {
|
||||
// use unknown protocol so that there is no default port
|
||||
const addr = new URLParse(`s://${match.groups.address.trim()}`);
|
||||
|
||||
args.onFind?.();
|
||||
stream.off("data", handler);
|
||||
clearTimeout(timeoutID);
|
||||
resolve(+addr.port);
|
||||
}
|
||||
};
|
||||
const timeoutID = setTimeout(() => {
|
||||
stream.off("data", handler);
|
||||
reject(new Error("failed to retrieve port from stream"));
|
||||
}, args.timeout ?? 5000);
|
||||
|
||||
stream.on("data", handler);
|
||||
});
|
||||
}
|
||||
@ -11,6 +11,7 @@ import { ClusterFrameInfo, clusterFrameMap } from "../common/cluster-frames";
|
||||
import { IpcRendererNavigationEvents } from "../renderer/navigation/events";
|
||||
import logger from "./logger";
|
||||
import { productName } from "../common/vars";
|
||||
import { LensProxy } from "./proxy/lens-proxy";
|
||||
|
||||
export class WindowManager extends Singleton {
|
||||
protected mainWindow: BrowserWindow;
|
||||
@ -20,7 +21,7 @@ export class WindowManager extends Singleton {
|
||||
|
||||
@observable activeClusterId: ClusterId;
|
||||
|
||||
constructor(protected proxyPort: number) {
|
||||
constructor() {
|
||||
super();
|
||||
this.bindEvents();
|
||||
this.initMenu();
|
||||
@ -28,7 +29,7 @@ export class WindowManager extends Singleton {
|
||||
}
|
||||
|
||||
get mainUrl() {
|
||||
return `http://localhost:${this.proxyPort}`;
|
||||
return `http://localhost:${LensProxy.getInstance().port}`;
|
||||
}
|
||||
|
||||
async initMainWindow(showSplash = true) {
|
||||
|
||||
@ -9,7 +9,7 @@ export default migration({
|
||||
run(store) {
|
||||
const hotbars: Hotbar[] = [];
|
||||
|
||||
ClusterStore.getInstance().enabledClustersList.forEach((cluster: any) => {
|
||||
ClusterStore.getInstance().clustersList.forEach((cluster: any) => {
|
||||
const name = cluster.workspace;
|
||||
|
||||
if (!name) return;
|
||||
|
||||
@ -13,13 +13,7 @@ import { CatalogEntity } from "../../api/catalog-entity";
|
||||
|
||||
|
||||
function getClusterForEntity(entity: CatalogEntity) {
|
||||
const cluster = ClusterStore.getInstance().getById(entity.metadata.uid);
|
||||
|
||||
if (!cluster?.enabled) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return cluster;
|
||||
return ClusterStore.getInstance().getById(entity.metadata.uid);
|
||||
}
|
||||
|
||||
entitySettingRegistry.add([
|
||||
|
||||
@ -27,10 +27,8 @@ export class RemoveClusterButton extends React.Component<Props> {
|
||||
}
|
||||
|
||||
render() {
|
||||
const { cluster } = this.props;
|
||||
|
||||
return (
|
||||
<Button accent onClick={this.confirmRemoveCluster} className="button-area" disabled={cluster.isManaged}>
|
||||
<Button accent onClick={this.confirmRemoveCluster} className="button-area">
|
||||
Remove Cluster
|
||||
</Button>
|
||||
);
|
||||
|
||||
Loading…
Reference in New Issue
Block a user