hendrik/lens
1
0 Fork 0
mirror of https://github.com/lensapp/lens.git synced 2025-05-20 05:10:56 +00:00
Code Issues Projects Releases Wiki Activity

KubeObjectStore & KubeWatchApi fixes and optimizations (#2063)

Browse Source 
* Detach NamespaceStore from KubeWatchApi, proper KubeObjectStore.loadAll (rebase of #2033)

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

* Watch-api requests optimization (#2066)

* subscribe for watching resources in single request when has admin-like access rights

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

* responding to comments

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

* fix unit-tests

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

* fix: reloading stores when preloading enabled and waitUntilLoaded=false

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

* mark Cluster.canUseWatchApi() and Cluster.refreshAccessibility() as private

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

* fix unit test: make public Cluster.canUseWatchApi()

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

* responding to comments in #2066

Signed-off-by: Roman <ixrock@gmail.com>
This commit is contained in:
Roman authored and GitHub committed 2021-02-04 11:51:35 +02:00
1 parent f56969a64b
commit 5287e7e528
21 files changed
+155 -101
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/renderer/api/kube-watch-api.ts
+77 -52
View File
@@ -5,12 +5,11 @@ import type { Cluster } from "../../main/cluster";
import type { IKubeWatchEvent, IKubeWatchEventStreamEnd, IWatchRoutePayload } from "../../main/routes/watch-route";
import type { KubeObject } from "./kube-object";
import type { KubeObjectStore } from "../kube-object.store";
import type { NamespaceStore } from "../components/+namespaces/namespace.store";
import plimit from "p-limit";
import debounce from "lodash/debounce";
import { comparer, computed, observable, reaction } from "mobx";
import { autobind, EventEmitter } from "../utils";
import { autorun, comparer, computed, IReactionDisposer, observable, reaction } from "mobx";
import { autobind, EventEmitter, noop } from "../utils";
import { ensureObjectSelfLink, KubeApi, parseKubeApi } from "./kube-api";
import { KubeJsonApiData, KubeJsonApiError } from "./kube-json-api";
import { apiPrefix, isDebugging, isProduction } from "../../common/vars";
@@ -19,6 +18,7 @@ import { apiManager } from "./api-manager";
export { IKubeWatchEvent, IKubeWatchEventStreamEnd };
export interface IKubeWatchMessage<T extends KubeObject = any> {
namespace?: string;
data?: IKubeWatchEvent<KubeJsonApiData>
error?: IKubeWatchEvent<KubeJsonApiError>;
api?: KubeApi<T>;
@@ -28,7 +28,7 @@ export interface IKubeWatchMessage<T extends KubeObject = any> {
export interface IKubeWatchSubscribeStoreOptions {
preload?: boolean; // preload store items, default: true
waitUntilLoaded?: boolean; // subscribe only after loading all stores, default: true
cacheLoading?: boolean; // when enabled loading store will be skipped, default: false
loadOnce?: boolean; // check store.isLoaded to skip loading if done already, default: false
}
export interface IKubeWatchReconnectOptions {
@@ -43,50 +43,49 @@ export interface IKubeWatchLog {
@autobind()
export class KubeWatchApi {
private cluster: Cluster;
private namespaceStore: NamespaceStore;
private requestId = 0;
private isConnected = false;
private reader: ReadableStreamReader<string>;
private subscribers = observable.map<KubeApi, number>();
// events
public onMessage = new EventEmitter<[IKubeWatchMessage]>();
@observable.ref private cluster: Cluster;
@observable.ref private namespaces: string[] = [];
@observable subscribers = observable.map<KubeApi, number>();
@observable isConnected = false;
@computed get isReady(): boolean {
return Boolean(this.cluster && this.namespaces);
}
@computed get isActive(): boolean {
return this.apis.length > 0;
}
@computed get apis(): string[] {
const { cluster, namespaceStore } = this;
const activeApis = Array.from(this.subscribers.keys());
if (!this.isReady) {
return [];
}
return activeApis.map(api => {
if (!cluster.isAllowedResource(api.kind)) {
return Array.from(this.subscribers.keys()).map(api => {
if (!this.isAllowedApi(api)) {
return [];
}
if (api.isNamespaced) {
return namespaceStore.getContextNamespaces().map(namespace => api.getWatchUrl(namespace));
} else {
return api.getWatchUrl();
if (api.isNamespaced && !this.cluster.isGlobalWatchEnabled) {
return this.namespaces.map(namespace => api.getWatchUrl(namespace));
}
return api.getWatchUrl();
}).flat();
}
constructor() {
this.init();
}
private async init() {
const { getHostedCluster } = await import("../../common/cluster-store");
const { namespaceStore } = await import("../components/+namespaces/namespace.store");
await namespaceStore.whenReady;
this.cluster = getHostedCluster();
this.namespaceStore = namespaceStore;
async init({ getCluster, getNamespaces }: {
getCluster: () => Cluster,
getNamespaces: () => string[],
}): Promise<void> {
autorun(() => {
this.cluster = getCluster();
this.namespaces = getNamespaces();
});
this.bindAutoConnect();
}
@@ -108,7 +107,7 @@ export class KubeWatchApi {
}
isAllowedApi(api: KubeApi): boolean {
return !!this?.cluster.isAllowedResource(api.kind);
return Boolean(this?.cluster.isAllowedResource(api.kind));
}
subscribeApi(api: KubeApi | KubeApi[]): () => void {
@@ -129,45 +128,66 @@ export class KubeWatchApi {
};
}
subscribeStores(stores: KubeObjectStore[], options: IKubeWatchSubscribeStoreOptions = {}): () => void {
const { preload = true, waitUntilLoaded = true, cacheLoading = false } = options;
preloadStores(stores: KubeObjectStore[], { loadOnce = false } = {}) {
const limitRequests = plimit(1); // load stores one by one to allow quick skipping when fast clicking btw pages
const preloading: Promise<any>[] = [];
for (const store of stores) {
preloading.push(limitRequests(async () => {
if (store.isLoaded && loadOnce) return; // skip
return store.loadAll(this.namespaces);
}));
}
return {
loading: Promise.allSettled(preloading),
cancelLoading: () => limitRequests.clearQueue(),
};
}
subscribeStores(stores: KubeObjectStore[], options: IKubeWatchSubscribeStoreOptions = {}): () => void {
const { preload = true, waitUntilLoaded = true, loadOnce = false } = options;
const apis = new Set(stores.map(store => store.getSubscribeApis()).flat());
const unsubscribeList: (() => void)[] = [];
let isUnsubscribed = false;
const load = () => this.preloadStores(stores, { loadOnce });
let preloading = preload && load();
let cancelReloading: IReactionDisposer = noop;
const subscribe = () => {
if (isUnsubscribed) return;
apis.forEach(api => unsubscribeList.push(this.subscribeApi(api)));
};
if (preload) {
for (const store of stores) {
preloading.push(limitRequests(async () => {
if (cacheLoading && store.isLoaded) return; // skip
return store.loadAll();
}));
}
}
if (waitUntilLoaded) {
Promise.all(preloading).then(subscribe, error => {
this.log({
message: new Error("Loading stores has failed"),
meta: { stores, error, options },
if (preloading) {
if (waitUntilLoaded) {
preloading.loading.then(subscribe, error => {
this.log({
message: new Error("Loading stores has failed"),
meta: { stores, error, options },
});
});
} else {
subscribe();
}
// reload when context namespaces changes
cancelReloading = reaction(() => this.namespaces, () => {
preloading?.cancelLoading();
preloading = load();
}, {
equals: comparer.shallow,
});
} else {
subscribe();
}
// unsubscribe
return () => {
if (isUnsubscribed) return;
isUnsubscribed = true;
limitRequests.clearQueue();
cancelReloading();
preloading?.cancelLoading();
unsubscribeList.forEach(unsubscribe => unsubscribe());
};
}
@@ -254,6 +274,10 @@ export class KubeWatchApi {
const kubeEvent: IKubeWatchEvent = JSON.parse(json);
const message = this.getMessage(kubeEvent);
if (!this.namespaces.includes(message.namespace)) {
continue; // skip updates from non-watching resources context
}
this.onMessage.emit(message);
} catch (error) {
return json;
@@ -286,6 +310,7 @@ export class KubeWatchApi {
message.api = api;
message.store = apiManager.getStore(api);
message.namespace = namespace;
}
break;
}