clean up / fixes

Signed-off-by: Roman <ixrock@gmail.com>
This commit is contained in:
Roman committed 2020-07-11 16:19:38 +03:00
1 parent e72e28ddab
commit 5eca8bc01c
19 files changed
+168 -199

No files matched your search

+2 -2
View File
@@ -4,7 +4,7 @@ import path from "path"
import http from "http"
import { copyFile, ensureDir } from "fs-extra"
import filenamify from "filenamify"
import { apiPrefix, appProto } from "../common/vars";
import { apiKubePrefix, appProto } from "../common/vars";
import { ClusterId, ClusterModel, clusterStore } from "../common/cluster-store"
import { handleMessages } from "../common/ipc-helpers";
import { ClusterIpcMessage } from "../common/ipc-messages";
@@ -100,7 +100,7 @@ export class ClusterManager {
cluster = this.getCluster(clusterId)
if (cluster) {
// we need to swap path prefix so that request is proxied to kube api
req.url = req.url.replace(`/${clusterId}`, apiPrefix.KUBE_BASE)
req.url = req.url.replace(`/${clusterId}`, apiKubePrefix)
}
}
} else {
+5 -4
View File
@@ -37,8 +37,8 @@ export class Cluster implements ClusterModel {
protected kubeconfigManager: KubeconfigManager;
@observable initialized = false;
@observable workspace: string;
@observable contextName: string;
@observable workspace: string;
@observable kubeConfigPath: string;
@observable port: number;
@observable url: string; // cluster-api url
@@ -70,14 +70,15 @@ export class Cluster implements ClusterModel {
@action
async init() {
try {
// fixme: all broken
// fixme
this.contextHandler = new ContextHandler(this);
this.port = await this.contextHandler.resolveProxyPort(); // resolve port before KubeconfigManager
this.webContentUrl = `http://${this.id}.localhost:${this.port}`;
this.port = await this.contextHandler.ensurePort(); // resolve port before KubeconfigManager
this.kubeAuthProxyUrl = `http://127.0.0.1:${this.port}`;
this.kubeconfigManager = new KubeconfigManager(this);
// this.url = this.kubeconfigManager.getCurrentClusterServer();
// this.apiUrl = url.parse(this.url);
this.webContentUrl = `http://${this.id}.localhost:${this.port}`;
logger.info(`[CLUSTER]: init success`, {
id: this.id,
+18 -17
View File
@@ -1,18 +1,19 @@
import type { PrometheusProvider, PrometheusService } from "./prometheus/provider-registry"
import type { ClusterPreferences } from "../common/cluster-store";
import type { ServerOptions } from "http-proxy"
import type { Cluster } from "./cluster"
import type httpProxy from "http-proxy"
import { CoreV1Api } from "@kubernetes/client-node"
import { observable } from "mobx";
import { prometheusProviders } from "../common/prometheus-providers"
import logger from "./logger"
import { getFreePort } from "./port"
import { KubeAuthProxy } from "./kube-auth-proxy"
export class ContextHandler {
public proxyPort: number
@observable proxyPort: number;
protected proxyServer: KubeAuthProxy
protected apiTarget: ServerOptions
protected apiTarget: httpProxy.ServerOptions
protected prometheusProvider: string
protected prometheusPath: string
@@ -44,7 +45,7 @@ export class ContextHandler {
}
public async getPrometheusService(): Promise<PrometheusService> {
const providers = this.prometheusProvider ? prometheusProviders.filter((p, _) => p.id == this.prometheusProvider) : prometheusProviders
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.proxyKubeconfig().makeApiClient(CoreV1Api)
return await provider.getPrometheusService(apiClient)
@@ -66,7 +67,7 @@ export class ContextHandler {
return this.prometheusPath;
}
public async getApiTarget(isWatchRequest = false): Promise<ServerOptions> {
public async getApiTarget(isWatchRequest = false): Promise<httpProxy.ServerOptions> {
if (this.apiTarget && !isWatchRequest) {
return this.apiTarget
}
@@ -78,24 +79,24 @@ export class ContextHandler {
return apiTarget
}
// fixme
protected async newApiTarget(timeout: number): Promise<ServerOptions> {
public async getApiTargetUrl(): Promise<string> {
const port = await this.ensurePort();
const { path } = this.cluster.apiUrl;
return `http://127.0.0.1:${port}${path}`;
}
protected async newApiTarget(timeout: number): Promise<httpProxy.ServerOptions> {
return {
changeOrigin: true,
target: await this.getApiTargetUrl(),
timeout: timeout,
headers: {
"Host": this.cluster.apiUrl.hostname
},
target: {
port: await this.resolveProxyPort(),
protocol: "http://",
host: "localhost",
path: this.cluster.apiUrl.path,
},
"Host": this.cluster.apiUrl.hostname,
}
}
}
async resolveProxyPort(): Promise<number> {
async ensurePort(): Promise<number> {
if (!this.proxyPort) {
this.proxyPort = await getFreePort();
}
@@ -104,7 +105,7 @@ export class ContextHandler {
public async ensureServer() {
if (!this.proxyServer) {
await this.resolveProxyPort();
await this.ensurePort();
const proxyEnv = Object.assign({}, process.env)
if (this.cluster.preferences.httpsProxy) {
proxyEnv.HTTPS_PROXY = this.cluster.preferences.httpsProxy
+2 -2
View File
@@ -1,6 +1,6 @@
import { ChildProcess, spawn } from "child_process"
import { waitUntilUsed } from "tcp-port-used";
import { sendMessageToRenderer } from "../common/ipc-helpers";
import { sendMessage } from "../common/ipc-helpers";
import type { Cluster } from "./cluster"
import { bundledKubectl, Kubectl } from "./kubectl"
import logger from "./logger"
@@ -84,7 +84,7 @@ export class KubeAuthProxy {
const channel = `kube-auth:${this.cluster.id}`
const message = { data, stream };
logger.debug(channel, message);
sendMessageToRenderer(channel, message);
sendMessage(channel, message);
}
public exit() {
+8 -4
View File
@@ -10,7 +10,11 @@ export class KubeconfigManager {
protected tempFile: string
constructor(protected cluster: Cluster) {
this.tempFile = this.createTemporaryKubeconfig();
this.init();
}
protected async init() {
this.tempFile = await this.createTemporaryKubeconfig();
}
getPath() {
@@ -21,7 +25,7 @@ export class KubeconfigManager {
* Creates new "temporary" kubeconfig that point to the kubectl-proxy.
* This way any user of the config does not need to know anything about the auth etc. details.
*/
protected createTemporaryKubeconfig(): string {
protected async createTemporaryKubeconfig(): Promise<string> {
fs.ensureDir(this.configDir);
const path = `${this.configDir}/${randomFileName("kubeconfig")}`;
const { contextName, contextHandler, kubeConfigPath } = this.cluster;
@@ -29,7 +33,7 @@ export class KubeconfigManager {
kubeConfig.clusters = [
{
name: contextName,
server: `http://127.0.0.1:${contextHandler.proxyPort}`, // fixme: extract
server: await contextHandler.getApiTargetUrl(),
skipTLSVerify: true,
}
];
@@ -42,7 +46,7 @@ export class KubeconfigManager {
user: "proxy",
name: contextName,
cluster: contextName,
// namespace: kubeConfig.getContextObject(contextName).namespace,
namespace: kubeConfig.getContextObject(contextName).namespace,
}
];
logger.info(`Creating temp config for context "${contextName}" at "${path}"`);
+25 -23
View File
@@ -1,13 +1,13 @@
import net from "net";
import http from "http";
import httpProxy from "http-proxy";
import { Socket } from "net";
import * as url from "url";
import url from "url";
import * as WebSocket from "ws"
import { ContextHandler } from "./context-handler";
import * as nodeShell from "./node-shell-session"
import { ClusterManager } from "./cluster-manager"
import { Router } from "./router"
import { apiPrefix } from "../common/vars";
import { ClusterManager } from "./cluster-manager"
import { ContextHandler } from "./context-handler";
import { apiKubePrefix } from "../common/vars";
import logger from "./logger"
export class LensProxy {
@@ -27,7 +27,7 @@ export class LensProxy {
}
listen(): this {
const proxyServer = this.buildProxyServer();
const proxyServer = this.buildCustomProxy();
const { proxyPort } = this.clusterManager;
this.proxyServer = proxyServer.listen(proxyPort);
logger.info(`LensProxy server has started http://localhost:${proxyPort}`);
@@ -40,21 +40,21 @@ export class LensProxy {
this.closed = true
}
protected buildProxyServer() {
protected buildCustomProxy(): http.Server {
const proxy = this.createProxy();
const proxyServer = http.createServer((req: http.IncomingMessage, res: http.ServerResponse) => {
const customProxy = http.createServer((req: http.IncomingMessage, res: http.ServerResponse) => {
this.handleRequest(proxy, req, res);
});
proxyServer.on("upgrade", (req: http.IncomingMessage, socket: Socket, head: Buffer) => {
customProxy.on("upgrade", (req: http.IncomingMessage, socket: net.Socket, head: Buffer) => {
this.handleWsUpgrade(req, socket, head)
});
proxyServer.on("error", (err) => {
customProxy.on("error", (err) => {
logger.error("proxy error", err)
});
return proxyServer;
return customProxy;
}
protected createProxy() {
protected createProxy(): httpProxy {
const proxy = httpProxy.createProxyServer();
proxy.on("proxyRes", (proxyRes, req, res) => {
@@ -71,16 +71,18 @@ export class LensProxy {
if (req.method !== "GET") {
return
}
const key = `${req.headers.host}${req.url}`
if (this.retryCounters.has(key)) {
logger.debug("Resetting proxy retry cache for url: " + key)
this.retryCounters.delete(key)
const reqUrl = `${req.headers.host}${req.url}`
if (this.retryCounters.has(reqUrl)) {
logger.debug("Resetting proxy retry cache for url: " + reqUrl)
this.retryCounters.delete(reqUrl)
}
})
proxy.on("error", (error, req, res, target) => {
if (this.closed) return;
if (this.closed) {
return;
}
if (target) {
logger.debug("Failed proxy to target: " + JSON.stringify(target))
logger.debug("Failed proxy to target: " + JSON.stringify(target, null, 2));
if (req.method === "GET" && (!res.statusCode || res.statusCode >= 500)) {
const retryCounterKey = `${req.headers.host}${req.url}`
const retryCount = this.retryCounters.get(retryCounterKey) || 0
@@ -112,11 +114,11 @@ export class LensProxy {
}));
}
// fixme: remove api prefix?
protected async getProxyTarget(req: http.IncomingMessage, contextHandler: ContextHandler): Promise<httpProxy.ServerOptions> {
const prefix = apiPrefix.KUBE_BASE;
if (req.url.startsWith(prefix)) {
if (req.url.startsWith(apiKubePrefix)) {
delete req.headers.authorization
req.url = req.url.replace(prefix, "")
req.url = req.url.replace(apiKubePrefix, "")
const isWatchRequest = req.url.includes("watch=")
return await contextHandler.getApiTarget(isWatchRequest)
}
@@ -137,11 +139,11 @@ export class LensProxy {
if (proxyTarget) {
proxy.web(req, res, proxyTarget)
} else {
this.router.route(cluster, req, res); // todo: handle "not-found" if isBoom==true?
this.router.route(cluster, req, res);
}
}
protected async handleWsUpgrade(req: http.IncomingMessage, socket: Socket, head: Buffer) {
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);
+1 -1
View File
@@ -1,7 +1,7 @@
import logger from "./logger"
import { createServer, AddressInfo } from "net"
// todo: replace with https://github.com/http-party/node-portfinder ?
// todo: use https://github.com/http-party/node-portfinder ?
const getNextAvailablePort = () => {
logger.debug("getNextAvailablePort() start")
+52 -52
View File
@@ -9,37 +9,28 @@ import { resourceApplierApi } from "./resource-applier-api"
import { apiPrefix, appName, outDir } from "../common/vars";
import { configRoute, kubeconfigRoute, metricsRoute, portForwardRoute, watchRoute } from "./routes";
const mimeTypes: Record<string, string> = {
"html": "text/html",
"txt": "text/plain",
"css": "text/css",
"gif": "image/gif",
"jpg": "image/jpeg",
"png": "image/png",
"svg": "image/svg+xml",
"js": "application/javascript",
"woff2": "font/woff2",
"ttf": "font/ttf"
};
interface RouteParams {
[key: string]: string | undefined;
export interface RouterRequestOpts<P extends Record<string, string> = any> {
req: http.IncomingMessage;
res: http.ServerResponse;
cluster: Cluster;
url: URL;
params: P; // https://hapi.dev/module/call/api
}
export type LensApiRequest = {
export interface LensApiRequest<D = any, P = any> {
path: string;
payload: D;
params: P;
cluster: Cluster;
payload: any;
raw: {
req: http.IncomingMessage;
};
params: RouteParams;
response: http.ServerResponse;
query: URLSearchParams;
path: string;
raw: {
req: http.IncomingMessage;
}
}
export class Router {
protected router: any
protected router: any;
public constructor() {
this.router = new Call.Router();
@@ -47,20 +38,23 @@ export class Router {
}
public async route(cluster: Cluster, req: http.IncomingMessage, res: http.ServerResponse): Promise<boolean> {
const reqUrl = new URL(req.url, "http://localhost");
const path = reqUrl.pathname
const url = new URL(req.url, "http://localhost");
const path = url.pathname
const method = req.method.toLowerCase()
const matchingRoute = this.router.route(method, path);
const routeExists = !matchingRoute.isBoom;
if (routeExists) {
const request = await this.getRequest({ req, res, cluster, url: reqUrl, params: matchingRoute.params })
const routeFound = !matchingRoute.isBoom;
if (routeFound) {
const request = await this.getRequest({
req, res, cluster, url,
params: matchingRoute.params
});
await matchingRoute.route(request)
return true
}
return false;
}
protected async getRequest(opts: { req: http.IncomingMessage; res: http.ServerResponse; cluster: Cluster; url: URL; params: RouteParams }) {
protected async getRequest(opts: RouterRequestOpts) {
const { req, res, url, cluster, params } = opts
const { payload } = await Subtext.parse(req, null, { parse: true, output: 'data' });
const request: LensApiRequest = {
@@ -78,6 +72,18 @@ export class Router {
}
protected getMimeType(filename: string) {
const mimeTypes: Record<string, string> = {
html: "text/html",
txt: "text/plain",
css: "text/css",
gif: "image/gif",
jpg: "image/jpeg",
png: "image/png",
svg: "image/svg+xml",
js: "application/javascript",
woff2: "font/woff2",
ttf: "font/ttf"
};
return mimeTypes[path.extname(filename).slice(1)] || "text/plain"
}
@@ -94,12 +100,6 @@ export class Router {
}
protected addRoutes() {
const {
BASE: apiBase,
KUBE_HELM: apiHelm,
KUBE_RESOURCE_APPLIER: apiResource,
} = apiPrefix;
// Static assets
this.router.add({ method: 'get', path: '/{path*}' }, (request: LensApiRequest) => {
const { response, params } = request
@@ -107,33 +107,33 @@ export class Router {
this.handleStaticFile(file, response)
})
this.router.add({ method: "get", path: `${apiBase}/config` }, configRoute.routeConfig.bind(configRoute))
this.router.add({ method: "get", path: `${apiBase}/kubeconfig/service-account/{namespace}/{account}` }, kubeconfigRoute.routeServiceAccountRoute.bind(kubeconfigRoute))
this.router.add({ method: "get", path: `${apiPrefix}/config` }, configRoute.routeConfig.bind(configRoute))
this.router.add({ method: "get", path: `${apiPrefix}/kubeconfig/service-account/{namespace}/{account}` }, kubeconfigRoute.routeServiceAccountRoute.bind(kubeconfigRoute))
// Watch API
this.router.add({ method: "get", path: `${apiBase}/watch` }, watchRoute.routeWatch.bind(watchRoute))
this.router.add({ method: "get", path: `${apiPrefix}/watch` }, watchRoute.routeWatch.bind(watchRoute))
// Metrics API
this.router.add({ method: "post", path: `${apiBase}/metrics` }, metricsRoute.routeMetrics.bind(metricsRoute))
this.router.add({ method: "post", path: `${apiPrefix}/metrics` }, metricsRoute.routeMetrics.bind(metricsRoute))
// Port-forward API
this.router.add({ method: "post", path: `${apiBase}/services/{namespace}/{service}/port-forward/{port}` }, portForwardRoute.routeServicePortForward.bind(portForwardRoute))
this.router.add({ method: "post", path: `${apiPrefix}/services/{namespace}/{service}/port-forward/{port}` }, portForwardRoute.routeServicePortForward.bind(portForwardRoute))
// Helm API
this.router.add({ method: "get", path: `${apiHelm}/v2/charts` }, helmApi.listCharts.bind(helmApi))
this.router.add({ method: "get", path: `${apiHelm}/v2/charts/{repo}/{chart}` }, helmApi.getChart.bind(helmApi))
this.router.add({ method: "get", path: `${apiHelm}/v2/charts/{repo}/{chart}/values` }, helmApi.getChartValues.bind(helmApi))
this.router.add({ method: "get", path: `${apiPrefix}/v2/charts` }, helmApi.listCharts.bind(helmApi))
this.router.add({ method: "get", path: `${apiPrefix}/v2/charts/{repo}/{chart}` }, helmApi.getChart.bind(helmApi))
this.router.add({ method: "get", path: `${apiPrefix}/v2/charts/{repo}/{chart}/values` }, helmApi.getChartValues.bind(helmApi))
this.router.add({ method: "post", path: `${apiHelm}/v2/releases` }, helmApi.installChart.bind(helmApi))
this.router.add({ method: `put`, path: `${apiHelm}/v2/releases/{namespace}/{release}` }, helmApi.updateRelease.bind(helmApi))
this.router.add({ method: `put`, path: `${apiHelm}/v2/releases/{namespace}/{release}/rollback` }, helmApi.rollbackRelease.bind(helmApi))
this.router.add({ method: "get", path: `${apiHelm}/v2/releases/{namespace?}` }, helmApi.listReleases.bind(helmApi))
this.router.add({ method: "get", path: `${apiHelm}/v2/releases/{namespace}/{release}` }, helmApi.getRelease.bind(helmApi))
this.router.add({ method: "get", path: `${apiHelm}/v2/releases/{namespace}/{release}/values` }, helmApi.getReleaseValues.bind(helmApi))
this.router.add({ method: "get", path: `${apiHelm}/v2/releases/{namespace}/{release}/history` }, helmApi.getReleaseHistory.bind(helmApi))
this.router.add({ method: "delete", path: `${apiHelm}/v2/releases/{namespace}/{release}` }, helmApi.deleteRelease.bind(helmApi))
this.router.add({ method: "post", path: `${apiPrefix}/v2/releases` }, helmApi.installChart.bind(helmApi))
this.router.add({ method: `put`, path: `${apiPrefix}/v2/releases/{namespace}/{release}` }, helmApi.updateRelease.bind(helmApi))
this.router.add({ method: `put`, path: `${apiPrefix}/v2/releases/{namespace}/{release}/rollback` }, helmApi.rollbackRelease.bind(helmApi))
this.router.add({ method: "get", path: `${apiPrefix}/v2/releases/{namespace?}` }, helmApi.listReleases.bind(helmApi))
this.router.add({ method: "get", path: `${apiPrefix}/v2/releases/{namespace}/{release}` }, helmApi.getRelease.bind(helmApi))
this.router.add({ method: "get", path: `${apiPrefix}/v2/releases/{namespace}/{release}/values` }, helmApi.getReleaseValues.bind(helmApi))
this.router.add({ method: "get", path: `${apiPrefix}/v2/releases/{namespace}/{release}/history` }, helmApi.getReleaseHistory.bind(helmApi))
this.router.add({ method: "delete", path: `${apiPrefix}/v2/releases/{namespace}/{release}` }, helmApi.deleteRelease.bind(helmApi))
// Resource Applier API
this.router.add({ method: "post", path: `${apiResource}/stack` }, resourceApplierApi.applyResource.bind(resourceApplierApi))
this.router.add({ method: "post", path: `${apiPrefix}/stack` }, resourceApplierApi.applyResource.bind(resourceApplierApi))
}
}
+20 -34
View File
@@ -2,55 +2,41 @@ import { LensApiRequest } from "../router"
import { LensApi } from "../lens-api"
import requestPromise from "request-promise-native"
import { PrometheusClusterQuery, PrometheusIngressQuery, PrometheusNodeQuery, PrometheusPodQuery, PrometheusProvider, PrometheusPvcQuery, PrometheusQueryOpts } from "../prometheus/provider-registry"
import { apiPrefix } from "../../common/vars";
export type IMetricsQuery = string | string[] | {
[metricName: string]: string;
}
class MetricsRoute extends LensApi {
public async routeMetrics(request: LensApiRequest) {
const { response, cluster } = request
const serverUrl = `http://127.0.0.1:${cluster.port}${apiPrefix.KUBE_BASE}` // fixme: extract
const query: IMetricsQuery = request.payload;
const headers: Record<string, string> = {
"Host": `${cluster.id}.localhost:${cluster.port}`,
"Content-type": "application/json",
}
const queryParams: IMetricsQuery = {}
request.query.forEach((value: string, key: string) => {
queryParams[key] = value
})
public async routeMetrics(request: LensApiRequest<IMetricsQuery>) {
const { response, cluster, payload } = request
const { contextHandler } = cluster;
const serverUrl = await contextHandler.getApiTargetUrl();
let metricsUrl: string
let prometheusProvider: PrometheusProvider
try {
const prometheusPath = await cluster.contextHandler.getPrometheusPath()
const prometheusPath = await contextHandler.getPrometheusPath()
metricsUrl = `${serverUrl}/api/v1/namespaces/${prometheusPath}/proxy${cluster.getPrometheusApiPrefix()}/api/v1/query_range`
prometheusProvider = await cluster.contextHandler.getPrometheusProvider()
prometheusProvider = await contextHandler.getPrometheusProvider()
} catch {
this.respondJson(response, {})
return
}
// prometheus metrics loader
const attempts: { [query: string]: number } = {};
const attempts: Record<string, number> = {};
const maxAttempts = 5;
const loadMetrics = (orgQuery: string): Promise<any> => {
const query = orgQuery.trim()
const attempt = attempts[query] = (attempts[query] || 0) + 1;
const loadMetrics = (promQuery: string): Promise<any> => {
const queryString = request.query.toString() + `&query=` + promQuery;
const attempt = attempts[queryString] = (attempts[queryString] || 0) + 1;
return requestPromise(metricsUrl, {
resolveWithFullResponse: false,
headers: headers,
json: true,
qs: {
query: query,
...queryParams
}
qs: queryString,
useQuerystring: true,
resolveWithFullResponse: false,
}).catch(async (error) => {
if (attempt < maxAttempts && (error.statusCode && error.statusCode != 404)) {
await new Promise(resolve => setTimeout(resolve, attempt * 1000)); // add delay before repeating request
return loadMetrics(query);
return loadMetrics(queryString);
}
return {
status: error.toString(),
@@ -63,14 +49,14 @@ class MetricsRoute extends LensApi {
// return data in same structure as query
let data: any;
if (typeof query === "string") {
data = await loadMetrics(query)
} else if (Array.isArray(query)) {
data = await Promise.all(query.map(loadMetrics));
if (typeof payload === "string") {
data = await loadMetrics(payload)
} else if (Array.isArray(payload)) {
data = await Promise.all(payload.map(loadMetrics));
} else {
data = {};
const result = await Promise.all(
Object.entries(query).map((queryEntry: any) => {
Object.entries(payload).map((queryEntry: any) => {
const queryName: string = queryEntry[0]
const queryOpts: PrometheusQueryOpts = queryEntry[1]
const queries = prometheusProvider.getQueries(queryOpts)
@@ -78,7 +64,7 @@ class MetricsRoute extends LensApi {
return loadMetrics(q)
})
);
Object.keys(query).forEach((metricName, index) => {
Object.keys(payload).forEach((metricName, index) => {
data[metricName] = result[index];
});
}