|
|
@@ -0,0 +1,420 @@
|
|
|
+/**
|
|
|
+ * Kubernetes client with list-watch and in-memory cache.
|
|
|
+ *
|
|
|
+ * Watches a set of Custom Resources and keeps an in-memory cache updated via
|
|
|
+ * Kubernetes informers (list + watch). Cache reads are synchronous and
|
|
|
+ * allocation-free; all I/O happens in the background after init().
|
|
|
+ *
|
|
|
+ * KubeConfig is built from K8sClientConfig supplied by the caller (Server):
|
|
|
+ * - apiUrl + token present → explicit out-of-cluster config
|
|
|
+ * - apiUrl + token absent → loadFromDefault (in-cluster SA or ~/.kube/config)
|
|
|
+ *
|
|
|
+ * TLS:
|
|
|
+ * - skipTlsVerify: true disables certificate verification (dev only).
|
|
|
+ * - In production, mount the cluster CA and let KubeConfig handle it.
|
|
|
+ */
|
|
|
+import {
|
|
|
+ ApiException,
|
|
|
+ AppsV1Api,
|
|
|
+ BatchV1Api,
|
|
|
+ CoreV1Api,
|
|
|
+ CustomObjectsApi,
|
|
|
+ type Informer,
|
|
|
+ KubeConfig,
|
|
|
+ type KubernetesObject,
|
|
|
+ makeInformer,
|
|
|
+ NetworkingV1Api,
|
|
|
+ setHeaderOptions,
|
|
|
+} from '@kubernetes/client-node';
|
|
|
+
|
|
|
+import { k8sLogger as log } from '$lib/server/logger';
|
|
|
+
|
|
|
+// --- Types ---
|
|
|
+
|
|
|
+/** Identifies a single Kubernetes resource type to watch. */
|
|
|
+type WatchedResource = {
|
|
|
+ group: string;
|
|
|
+ version: string;
|
|
|
+ resource: string;
|
|
|
+ namespace: string;
|
|
|
+ labelSelector?: string;
|
|
|
+ fieldSelector?: string;
|
|
|
+};
|
|
|
+
|
|
|
+type K8sClientConfig = {
|
|
|
+ /** Explicit API server URL (out-of-cluster / dev). When absent, loadFromDefault() is used. */
|
|
|
+ apiUrl?: string;
|
|
|
+ /** Bearer token for explicit config. Required when apiUrl is set. */
|
|
|
+ token?: string;
|
|
|
+ /** Skip TLS certificate verification. Dev only. */
|
|
|
+ skipTlsVerify?: boolean;
|
|
|
+};
|
|
|
+
|
|
|
+// --- Helpers ---
|
|
|
+
|
|
|
+const cacheKey = (r: WatchedResource): string =>
|
|
|
+ `${r.group}/${r.version}/namespaces/${r.namespace}/${r.resource}`;
|
|
|
+
|
|
|
+const selectorQueryString = (
|
|
|
+ r: Pick<WatchedResource, 'labelSelector' | 'fieldSelector'>,
|
|
|
+): string => {
|
|
|
+ const params = new URLSearchParams();
|
|
|
+ if (r.labelSelector) params.set('labelSelector', r.labelSelector);
|
|
|
+ if (r.fieldSelector) params.set('fieldSelector', r.fieldSelector);
|
|
|
+ const query = params.toString();
|
|
|
+ return query ? `?${query}` : '';
|
|
|
+};
|
|
|
+
|
|
|
+const isCoreGroup = (group: string): boolean => group === '' || group === 'core';
|
|
|
+
|
|
|
+const isSupportedBuiltinResource = (r: WatchedResource): boolean => {
|
|
|
+ if (r.version !== 'v1') return false;
|
|
|
+
|
|
|
+ if (isCoreGroup(r.group)) {
|
|
|
+ return r.resource === 'pods' || r.resource === 'services' || r.resource === 'secrets';
|
|
|
+ }
|
|
|
+
|
|
|
+ return r.group === 'apps' && r.resource === 'deployments';
|
|
|
+};
|
|
|
+
|
|
|
+const watchPath = (r: WatchedResource): string =>
|
|
|
+ `${isCoreGroup(r.group) ? '/api' : '/apis'}/${isCoreGroup(r.group) ? r.version : `${r.group}/${r.version}`}/namespaces/${r.namespace}/${r.resource}${selectorQueryString(r)}`;
|
|
|
+
|
|
|
+const buildKubeConfig = (cfg: K8sClientConfig): KubeConfig => {
|
|
|
+ const kc = new KubeConfig();
|
|
|
+
|
|
|
+ if (cfg.apiUrl && cfg.token) {
|
|
|
+ log.debug('Loading kubeconfig from explicit config');
|
|
|
+ kc.loadFromOptions({
|
|
|
+ clusters: [
|
|
|
+ { name: 'current', server: cfg.apiUrl, skipTLSVerify: cfg.skipTlsVerify ?? false },
|
|
|
+ ],
|
|
|
+ users: [{ name: 'current', token: cfg.token }],
|
|
|
+ contexts: [{ name: 'current', cluster: 'current', user: 'current' }],
|
|
|
+ currentContext: 'current',
|
|
|
+ });
|
|
|
+ } else {
|
|
|
+ log.debug('Loading kubeconfig from cluster / default');
|
|
|
+ kc.loadFromDefault();
|
|
|
+ }
|
|
|
+
|
|
|
+ return kc;
|
|
|
+};
|
|
|
+
|
|
|
+// --- Helpers ---
|
|
|
+
|
|
|
+const isNotFound = (err: unknown): boolean => err instanceof ApiException && err.code === 404;
|
|
|
+const isAlreadyExists = (err: unknown): boolean => err instanceof ApiException && err.code === 409;
|
|
|
+
|
|
|
+// --- K8sClient ---
|
|
|
+
|
|
|
+const RECONNECT_DELAY_MS = 5_000;
|
|
|
+
|
|
|
+class K8sClient {
|
|
|
+ readonly #cfg: K8sClientConfig;
|
|
|
+ readonly #cache = new Map<string, Map<string, KubernetesObject>>();
|
|
|
+ readonly #kubeConfig: KubeConfig;
|
|
|
+ #informers: Informer<KubernetesObject>[] = [];
|
|
|
+
|
|
|
+ constructor(cfg: K8sClientConfig) {
|
|
|
+ this.#cfg = cfg;
|
|
|
+ this.#kubeConfig = buildKubeConfig(cfg);
|
|
|
+ }
|
|
|
+
|
|
|
+ async init(): Promise<void> {}
|
|
|
+
|
|
|
+ async shutdown(): Promise<void> {
|
|
|
+ log.info('Stopping all informers');
|
|
|
+ await Promise.all(this.#informers.map((i) => i.stop()));
|
|
|
+ this.#informers = [];
|
|
|
+ log.info('All informers stopped');
|
|
|
+ }
|
|
|
+
|
|
|
+ async watchResources<T extends KubernetesObject>(
|
|
|
+ res: WatchedResource,
|
|
|
+ onEvent?: (type: 'add' | 'update' | 'delete', obj: T) => void,
|
|
|
+ ): Promise<T[]> {
|
|
|
+ const key = cacheKey(res);
|
|
|
+ this.#cache.set(key, new Map());
|
|
|
+
|
|
|
+ const listFn = async () => {
|
|
|
+ if (isSupportedBuiltinResource(res)) {
|
|
|
+ if (res.group === 'apps' && res.resource === 'deployments') {
|
|
|
+ const appsApi = this.#kubeConfig.makeApiClient(AppsV1Api);
|
|
|
+ return appsApi.listNamespacedDeployment({
|
|
|
+ namespace: res.namespace,
|
|
|
+ labelSelector: res.labelSelector,
|
|
|
+ fieldSelector: res.fieldSelector,
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ const coreApi = this.#kubeConfig.makeApiClient(CoreV1Api);
|
|
|
+ if (res.resource === 'pods') {
|
|
|
+ return coreApi.listNamespacedPod({
|
|
|
+ namespace: res.namespace,
|
|
|
+ labelSelector: res.labelSelector,
|
|
|
+ fieldSelector: res.fieldSelector,
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ if (res.resource === 'services') {
|
|
|
+ return coreApi.listNamespacedService({
|
|
|
+ namespace: res.namespace,
|
|
|
+ labelSelector: res.labelSelector,
|
|
|
+ fieldSelector: res.fieldSelector,
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ if (res.resource === 'secrets') {
|
|
|
+ return coreApi.listNamespacedSecret({
|
|
|
+ namespace: res.namespace,
|
|
|
+ labelSelector: res.labelSelector,
|
|
|
+ fieldSelector: res.fieldSelector,
|
|
|
+ });
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ const customApi = this.#kubeConfig.makeApiClient(CustomObjectsApi);
|
|
|
+ return customApi.listNamespacedCustomObject({
|
|
|
+ group: res.group,
|
|
|
+ version: res.version,
|
|
|
+ namespace: res.namespace,
|
|
|
+ plural: res.resource,
|
|
|
+ labelSelector: res.labelSelector,
|
|
|
+ fieldSelector: res.fieldSelector,
|
|
|
+ });
|
|
|
+ };
|
|
|
+
|
|
|
+ const informer = makeInformer<T>(this.#kubeConfig, watchPath(res), listFn);
|
|
|
+
|
|
|
+ const upsert = (obj: T) => {
|
|
|
+ const name = obj.metadata?.name;
|
|
|
+ if (name) this.#cache.get(key)!.set(name, obj as unknown as T);
|
|
|
+ };
|
|
|
+
|
|
|
+ informer.on('add', (obj: T) => {
|
|
|
+ upsert(obj);
|
|
|
+ onEvent?.('add', obj);
|
|
|
+ });
|
|
|
+ informer.on('update', (obj: T) => {
|
|
|
+ upsert(obj);
|
|
|
+ onEvent?.('update', obj);
|
|
|
+ });
|
|
|
+ informer.on('delete', (obj: T) => {
|
|
|
+ const name = obj.metadata?.name;
|
|
|
+ if (name) this.#cache.get(key)!.delete(name);
|
|
|
+ onEvent?.('delete', obj);
|
|
|
+ });
|
|
|
+
|
|
|
+ informer.on('error', (err: Error) => {
|
|
|
+ log.error('Watch error, reconnecting', {
|
|
|
+ resource: `${res.group}/${res.resource}`,
|
|
|
+ namespace: res.namespace,
|
|
|
+ err,
|
|
|
+ });
|
|
|
+ setTimeout(() => informer.start(), RECONNECT_DELAY_MS);
|
|
|
+ });
|
|
|
+
|
|
|
+ this.#informers.push(informer);
|
|
|
+ log.info('Starting informer', { resource: res.resource, namespace: res.namespace });
|
|
|
+ await informer.start();
|
|
|
+ log.info('Informer ready', {
|
|
|
+ resource: res.resource,
|
|
|
+ namespace: res.namespace,
|
|
|
+ count: this.#cache.get(key)!.size,
|
|
|
+ });
|
|
|
+
|
|
|
+ return this.listResources<T>(res.group, res.version, res.resource, res.namespace);
|
|
|
+ }
|
|
|
+
|
|
|
+ async watchDeployments<T extends KubernetesObject>(
|
|
|
+ namespace: string,
|
|
|
+ selectors: Pick<WatchedResource, 'labelSelector' | 'fieldSelector'> = {},
|
|
|
+ ): Promise<T[]> {
|
|
|
+ return this.watchResources<T>({
|
|
|
+ group: 'apps',
|
|
|
+ version: 'v1',
|
|
|
+ resource: 'deployments',
|
|
|
+ namespace,
|
|
|
+ ...selectors,
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ async watchPods<T extends KubernetesObject>(
|
|
|
+ namespace: string,
|
|
|
+ selectors: Pick<WatchedResource, 'labelSelector' | 'fieldSelector'> = {},
|
|
|
+ ): Promise<T[]> {
|
|
|
+ return this.watchResources<T>({
|
|
|
+ group: '',
|
|
|
+ version: 'v1',
|
|
|
+ resource: 'pods',
|
|
|
+ namespace,
|
|
|
+ ...selectors,
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ async watchServices<T extends KubernetesObject>(
|
|
|
+ namespace: string,
|
|
|
+ selectors: Pick<WatchedResource, 'labelSelector' | 'fieldSelector'> = {},
|
|
|
+ ): Promise<T[]> {
|
|
|
+ return this.watchResources<T>({
|
|
|
+ group: '',
|
|
|
+ version: 'v1',
|
|
|
+ resource: 'services',
|
|
|
+ namespace,
|
|
|
+ ...selectors,
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Returns all cached resources for the given resource type. Synchronous. */
|
|
|
+ listResources<T>(group: string, version: string, resource: string, namespace: string): T[] {
|
|
|
+ const key = cacheKey({ group, version, resource, namespace });
|
|
|
+ const cache = this.#cache.get(key);
|
|
|
+ if (!cache) return [];
|
|
|
+ return Array.from(cache.values()) as unknown as T[];
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Returns a single cached resource by name. Synchronous. Returns undefined if not found. */
|
|
|
+ getResource<T>(
|
|
|
+ group: string,
|
|
|
+ version: string,
|
|
|
+ resource: string,
|
|
|
+ name: string,
|
|
|
+ namespace: string,
|
|
|
+ ): T | undefined {
|
|
|
+ const key = cacheKey({ group, version, resource, namespace });
|
|
|
+ return this.#cache.get(key)?.get(name) as unknown as T | undefined;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Returns all cached resource statuses for the given resource type. Synchronous. */
|
|
|
+ listResourceStatuses<TStatus>(
|
|
|
+ group: string,
|
|
|
+ version: string,
|
|
|
+ resource: string,
|
|
|
+ namespace: string,
|
|
|
+ ): TStatus[] {
|
|
|
+ const objects = this.listResources<{ status?: TStatus }>(group, version, resource, namespace);
|
|
|
+ return objects.flatMap((obj) => (obj.status === undefined ? [] : [obj.status]));
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Returns a single cached resource status by name. Synchronous. Returns undefined if not found. */
|
|
|
+ getResourceStatus<TStatus>(
|
|
|
+ group: string,
|
|
|
+ version: string,
|
|
|
+ resource: string,
|
|
|
+ name: string,
|
|
|
+ namespace: string,
|
|
|
+ ): TStatus | undefined {
|
|
|
+ const object = this.getResource<{ status?: TStatus }>(
|
|
|
+ group,
|
|
|
+ version,
|
|
|
+ resource,
|
|
|
+ name,
|
|
|
+ namespace,
|
|
|
+ );
|
|
|
+ return object?.status;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Create or patch a custom resource.
|
|
|
+ *
|
|
|
+ * Attempts to create the resource first (POST). If it already exists (409), falls back to
|
|
|
+ * Server-Side Apply (PATCH with `application/apply-patch+yaml`, `fieldManager: 'locostack-admin-web'`,
|
|
|
+ * `force: true`) which takes ownership of all fields and updates in-place.
|
|
|
+ */
|
|
|
+ async applyCustomResource(
|
|
|
+ group: string,
|
|
|
+ version: string,
|
|
|
+ plural: string,
|
|
|
+ namespace: string,
|
|
|
+ obj: KubernetesObject,
|
|
|
+ ): Promise<void> {
|
|
|
+ const name = obj.metadata?.name;
|
|
|
+ if (!name) throw new Error('metadata.name is required for applyCustomResource');
|
|
|
+
|
|
|
+ const customApi = this.#kubeConfig.makeApiClient(CustomObjectsApi);
|
|
|
+
|
|
|
+ // Try to create the resource; if it already exists, fall through to SSA patch.
|
|
|
+ try {
|
|
|
+ await customApi.createNamespacedCustomObject({
|
|
|
+ group,
|
|
|
+ version,
|
|
|
+ namespace,
|
|
|
+ plural,
|
|
|
+ body: obj,
|
|
|
+ });
|
|
|
+ log.debug('Created custom resource', { group, version, plural, namespace, name });
|
|
|
+ return;
|
|
|
+ } catch (err: unknown) {
|
|
|
+ if (!isAlreadyExists(err)) throw err;
|
|
|
+ }
|
|
|
+
|
|
|
+ await customApi.patchNamespacedCustomObject(
|
|
|
+ {
|
|
|
+ group,
|
|
|
+ version,
|
|
|
+ namespace,
|
|
|
+ plural,
|
|
|
+ name,
|
|
|
+ body: obj,
|
|
|
+ fieldManager: 'locostack-admin-web',
|
|
|
+ force: true,
|
|
|
+ },
|
|
|
+ setHeaderOptions('Content-Type', 'application/apply-patch+yaml'),
|
|
|
+ );
|
|
|
+ log.debug('Patched custom resource', { group, version, plural, namespace, name });
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Delete a custom resource by name.
|
|
|
+ * Does nothing (logs a warning) when the resource does not exist.
|
|
|
+ */
|
|
|
+ async deleteCustomResource(
|
|
|
+ group: string,
|
|
|
+ version: string,
|
|
|
+ plural: string,
|
|
|
+ namespace: string,
|
|
|
+ name: string,
|
|
|
+ ): Promise<void> {
|
|
|
+ const customApi = this.#kubeConfig.makeApiClient(CustomObjectsApi);
|
|
|
+ try {
|
|
|
+ await customApi.deleteNamespacedCustomObject({ group, version, namespace, plural, name });
|
|
|
+ log.debug('Deleted custom resource', { group, version, plural, namespace, name });
|
|
|
+ } catch (err: unknown) {
|
|
|
+ if (isNotFound(err)) {
|
|
|
+ log.warn('Custom resource not found on delete (ignored)', {
|
|
|
+ group,
|
|
|
+ version,
|
|
|
+ plural,
|
|
|
+ namespace,
|
|
|
+ name,
|
|
|
+ });
|
|
|
+ } else {
|
|
|
+ throw err;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ coreApi(): CoreV1Api {
|
|
|
+ return this.#kubeConfig.makeApiClient(CoreV1Api);
|
|
|
+ }
|
|
|
+
|
|
|
+ appsApi(): AppsV1Api {
|
|
|
+ return this.#kubeConfig.makeApiClient(AppsV1Api);
|
|
|
+ }
|
|
|
+
|
|
|
+ batchApi(): BatchV1Api {
|
|
|
+ return this.#kubeConfig.makeApiClient(BatchV1Api);
|
|
|
+ }
|
|
|
+
|
|
|
+ networkingApi(): NetworkingV1Api {
|
|
|
+ return this.#kubeConfig.makeApiClient(NetworkingV1Api);
|
|
|
+ }
|
|
|
+
|
|
|
+ customObjectsApi(): CustomObjectsApi {
|
|
|
+ return this.#kubeConfig.makeApiClient(CustomObjectsApi);
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+export type { K8sClientConfig, WatchedResource };
|
|
|
+export { K8sClient };
|