clean-up after-merge hook, refactoring

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

src/main/routes/watch-route.ts

@@ -68,7 +68,7 @@ class ApiWatcher {
       const event: IKubeWatchEventStreamEnd = {
         type: "STREAM_END",
         url: this.apiUrl,
-        status: 410,
+        status: 410, // https://kubernetes.io/docs/reference/using-api/api-concepts/#410-gone-responses
       };
 
       this.sendEvent(event);

src/renderer/api/kube-api.ts

@@ -35,6 +35,7 @@ export interface IKubeApiOptions<T extends KubeObject> {
 export interface IKubeApiQueryParams {
   watch?: boolean | number;
   resourceVersion?: string;
+  resourceVersionMatch?: "Exact" | "NotOlderThan" // see also: https://kubernetes.io/docs/reference/using-api/api-concepts/#resourceversion-in-metadata
   timeoutSeconds?: number;
   limit?: number; // doesn't work with ?watch
   continue?: string; // might be used with ?limit from second request
@@ -275,6 +276,7 @@ export class KubeApi<T extends KubeObject = any> {
     if (KubeObject.isJsonApiDataList(data)) {
       const { apiVersion, items, metadata } = data;
 
+      // save "resourceVersion" metadata from list requests
       this.setResourceVersion(namespace, metadata.resourceVersion);
       this.setResourceVersion("", metadata.resourceVersion);
 

src/renderer/components/+namespaces/namespace.store.ts

@@ -1,6 +1,6 @@
 import { action, comparer, computed, IReactionDisposer, IReactionOptions, observable, reaction } from "mobx";
 import { autobind, createStorage } from "../../utils";
-import { KubeObjectStore, KubeObjectStoreLoadingParams } from "../../kube-object.store";
+import { KubeObjectStore, KubeStoreLoadItemsOptions } from "../../kube-object.store";
 import { Namespace, namespacesApi } from "../../api/endpoints/namespaces.api";
 import { createPageParam } from "../../navigation";
 import { apiManager } from "../../api/api-manager";
@@ -117,7 +117,7 @@ export class NamespaceStore extends KubeObjectStore<Namespace> {
     return super.getSubscribeApis();
   }
 
-  protected async loadItems(params: KubeObjectStoreLoadingParams) {
+  protected async loadItems(params: KubeStoreLoadItemsOptions) {
     const { allowedNamespaces } = this;
 
     let namespaces = await super.loadItems(params);

src/renderer/components/+user-management-roles-bindings/role-bindings.store.ts

@@ -1,13 +1,13 @@
 import difference from "lodash/difference";
 import uniqBy from "lodash/uniqBy";
 import { clusterRoleBindingApi, IRoleBindingSubject, RoleBinding, roleBindingApi } from "../../api/endpoints";
-import { KubeObjectStore, KubeObjectStoreLoadingParams } from "../../kube-object.store";
+import { KubeObjectStore, KubeStoreLoadItemsOptions } from "../../kube-object.store";
 import { autobind } from "../../utils";
 import { apiManager } from "../../api/api-manager";
 
 @autobind()
 export class RoleBindingsStore extends KubeObjectStore<RoleBinding> {
-  api = clusterRoleBindingApi;
+  api = roleBindingApi;
 
   getSubscribeApis() {
     return [clusterRoleBindingApi, roleBindingApi];
@@ -26,7 +26,7 @@ export class RoleBindingsStore extends KubeObjectStore<RoleBinding> {
     return clusterRoleBindingApi.get(params);
   }
 
-  protected async loadItems(params: KubeObjectStoreLoadingParams): Promise<RoleBinding[]> {
+  protected async loadItems(params: KubeStoreLoadItemsOptions): Promise<RoleBinding[]> {
     const items = await Promise.all([
       super.loadItems({ ...params, api: clusterRoleBindingApi }),
       super.loadItems({ ...params, api: roleBindingApi }),

src/renderer/components/+user-management-roles/roles.store.ts

@@ -1,11 +1,11 @@
 import { clusterRoleApi, Role, roleApi } from "../../api/endpoints";
 import { autobind } from "../../utils";
-import { KubeObjectStore, KubeObjectStoreLoadingParams } from "../../kube-object.store";
+import { KubeObjectStore, KubeStoreLoadItemsOptions } from "../../kube-object.store";
 import { apiManager } from "../../api/api-manager";
 
 @autobind()
 export class RolesStore extends KubeObjectStore<Role> {
-  api = clusterRoleApi;
+  api = roleApi;
 
   getSubscribeApis() {
     return [roleApi, clusterRoleApi];
@@ -24,7 +24,7 @@ export class RolesStore extends KubeObjectStore<Role> {
     return clusterRoleApi.get(params);
   }
 
-  protected async loadItems(params: KubeObjectStoreLoadingParams): Promise<Role[]> {
+  protected async loadItems(params: KubeStoreLoadItemsOptions): Promise<Role[]> {
     const items = await Promise.all([
       super.loadItems({ ...params, api: clusterRoleApi }),
       super.loadItems({ ...params, api: roleApi }),

src/renderer/kube-object.store.ts

@@ -1,5 +1,4 @@
 import type { ClusterContext } from "./components/context";
-
 import { action, computed, observable, reaction, when } from "mobx";
 import { autobind } from "./utils";
 import { KubeObject } from "./api/kube-object";
@@ -9,9 +8,23 @@ import { apiManager } from "./api/api-manager";
 import { IKubeApiQueryParams, KubeApi, parseKubeApi } from "./api/kube-api";
 import { KubeJsonApiData } from "./api/kube-json-api";
 
-export interface KubeObjectStoreLoadingParams {
+export interface KubeStoreLoadAllOptions {
-  namespaces: string[];
+  namespaces?: string[]; // load specific namespaces into store or all items (by default)
-  api?: KubeApi;
+  updateStore?: boolean; // merge loaded items with specific arguments or return as a result
+  autoCleanUp?: boolean; // run cleaning operation for non-updated namespaced items in store (default: true)
+}
+
+export interface KubeStoreLoadItemsOptions {
+  namespaces: string[]; // list of namespaces for loading into store with following merge-update
+  api?: KubeApi; // api for loading resources, used for overriding, see: roles-store.ts
+  merge?: boolean; // merge items into store, default: false
+}
+
+export interface KubeStoreMergeItemsOptions {
+  replaceAll?: boolean; // completely replace items in store, default: false
+  updateStore?: boolean; // merge items into store after loading, default: true
+  sort?: boolean; // sort items before update
+  filter?: boolean; // sort items before update
 }
 
 @autobind()
@@ -39,7 +52,9 @@ export abstract class KubeObjectStore<T extends KubeObject = any> extends ItemSt
     return this.items.filter(item => {
       const itemNamespace = item.getNs();
 
-      return !itemNamespace /* cluster-wide */ || namespaces.includes(itemNamespace);
+      if (!itemNamespace) return true; // cluster-wide resource
+
+      return namespaces.includes(itemNamespace);
     });
   }
 
@@ -98,24 +113,37 @@ export abstract class KubeObjectStore<T extends KubeObject = any> extends ItemSt
     }
   }
 
-  protected async loadItems({ namespaces, api }: KubeObjectStoreLoadingParams): Promise<T[]> {
+  protected async loadItems({ namespaces, api = this.api, merge = false }: KubeStoreLoadItemsOptions): Promise<T[]> {
-    if (this.context?.cluster.isAllowedResource(api.kind)) {
+    await this.contextReady;
-      if (!api.isNamespaced) {
+    const { allNamespaces, cluster } = this.context;
-        return api.list({}, this.query);
+    let items: T[] = [];
+
+    if (!cluster.isAllowedResource(api.kind)) {
+      return items;
     }
 
-      const isLoadingAll = this.context.allNamespaces.every(ns => namespaces.includes(ns));
+    try {
+      // optimize check for loading "all namespaces" with single k8s request
+      const allNamespacesAffected = allNamespaces.every(ns => namespaces.includes(ns));
 
-      if (isLoadingAll) {
+      // cluster list request, e.g. /api/v1/nodes
-        return api.list({}, this.query);
+      if (!api.isNamespaced || (cluster.isAdmin && allNamespacesAffected)) {
+        items = await api.list({}, this.query);
       } else {
-        return Promise // load resources per namespace
+        // otherwise load resources per requested namespaces
+        items = await Promise
           .all(namespaces.map(namespace => api.list({ namespace })))
           .then(items => items.flat());
       }
+    } catch (error) {
+      console.error("Loading items failed", { error, namespaces, api });
     }
 
-    return [];
+    if (merge && items.length > 0) {
+      this.mergeItems(items, { replaceAll: false, updateStore: true });
+    }
+
+    return items;
   }
 
   protected filterItemsOnLoad(items: T[]) {
@@ -123,25 +151,31 @@ export abstract class KubeObjectStore<T extends KubeObject = any> extends ItemSt
   }
 
   @action
-  async loadAll(options: { namespaces?: string[], merge?: boolean } = {}): Promise<void | T[]> {
+  async loadAll({ namespaces, updateStore = true, autoCleanUp = true }: KubeStoreLoadAllOptions = {}): Promise<void | T[]> {
     await this.contextReady;
     this.isLoading = true;
 
     try {
-      const {
+      const newItems = await this.loadItems({
-        namespaces = this.context.allNamespaces, // load all namespaces by default
+        namespaces: namespaces ?? this.context.allNamespaces, // load all by default
-        merge = true, // merge loaded items or return as result
+        api: this.api
-      } = options;
+      });
 
-      const items = await this.loadItems({ namespaces, api: this.api });
+      if (updateStore) {
+        this.mergeItems(newItems, {
+          replaceAll: false, // partial update
+          updateStore: true,
+        });
+      } else {
+        return newItems;
+      }
+
+      // clean up possibly stale items and reload removed namespaces
+      if (autoCleanUp) {
+        await this.cleanUpAfterLoad(newItems).refreshRemovedItems();
+      }
 
       this.isLoaded = true;
-
-      if (merge) {
-        this.mergeItems(items, { replace: false });
-      } else {
-        return items;
-      }
     } catch (error) {
       console.error("Loading store items failed", { error, store: this });
       this.resetOnError(error);
@@ -162,11 +196,17 @@ export abstract class KubeObjectStore<T extends KubeObject = any> extends ItemSt
   }
 
   @action
-  mergeItems(partialItems: T[], { replace = false, updateStore = true, sort = true, filter = true } = {}): T[] {
+  mergeItems(partialItems: T[], {
+    replaceAll = false,
+    updateStore = true,
+    sort = true,
+    filter = true,
+  }: KubeStoreMergeItemsOptions = {}): T[] {
     let items = partialItems;
 
+    try {
       // update existing items
-    if (!replace) {
+      if (!replaceAll) {
         const partialIds = partialItems.map(item => item.getId());
 
         items = [
@@ -178,10 +218,40 @@ export abstract class KubeObjectStore<T extends KubeObject = any> extends ItemSt
       if (filter) items = this.filterItemsOnLoad(items);
       if (sort) items = this.sortItems(items);
       if (updateStore) this.items.replace(items);
+    } catch (error) {
+      // todo: improve logging
+      console.error("[KUBE-STORE]: merging items failed", { error, store: this });
+
+      return [];
+    }
 
     return items;
   }
 
+  @action
+  private cleanUpAfterLoad(updatedItems: T[]) {
+    const getUniqNamespaces = (items: T[]) => Array.from(new Set(items.map(item => item.getNs()).filter(Boolean)));
+
+    const loadedNamespaces = getUniqNamespaces(this.items);
+    const updatedNamespaces = getUniqNamespaces(updatedItems);
+    const staleNamespaces = loadedNamespaces.filter(ns => !updatedNamespaces.includes(ns));
+
+    if (staleNamespaces.length > 0) {
+      const freshItems = this.items.toJS().filter(item => {
+        if (!item.getNs()) return true; // cluster resource
+
+        return !staleNamespaces.includes(item.getNs());
+      });
+
+      this.items.replace(freshItems);
+    }
+
+    return {
+      removedNamespaces: staleNamespaces,
+      refreshRemovedItems: () => this.loadItems({ namespaces: staleNamespaces, merge: true }),
+    };
+  }
+
   protected resetOnError(error: any) {
     if (error) this.reset();
   }