chore: extract kube-api from core

Signed-off-by: Jari Kolehmainen <jari.kolehmainen@gmail.com>

packages/utility-features/kube-api/src/kube-api.ts

@@ -0,0 +1,887 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+
+// Base class for building all kubernetes apis
+
+import { merge } from "lodash";
+import { stringify } from "querystring";
+import { createKubeApiURL, parseKubeApi } from "./kube-api-parse";
+import type {
+  KubeObjectConstructor,
+  KubeJsonApiDataFor,
+  KubeObjectMetadata,
+  KubeJsonApiData,
+  KubeObject,
+  KubeObjectScope,
+  Scale,
+} from "@k8slens/kube-object";
+import {
+  isJsonApiData,
+  isJsonApiDataList,
+  isPartialJsonApiData,
+  KubeStatus,
+  isKubeStatusData,
+} from "@k8slens/kube-object";
+import byline from "byline";
+import type { IKubeWatchEvent } from "./kube-watch-event";
+import type { KubeJsonApi } from "./kube-json-api";
+import type { Disposer } from "@k8slens/utilities";
+import { isDefined, noop, WrappedAbortController } from "@k8slens/utilities";
+import type { RequestInit, Response } from "@k8slens/node-fetch";
+import type { Patch } from "rfc6902";
+import assert from "assert";
+import type { PartialDeep } from "type-fest";
+import type { Logger } from "@k8slens/logger";
+import { matches } from "lodash/fp";
+import { makeObservable, observable } from "mobx";
+import type { ScaleCreateOptions } from "@k8slens/kube-object/src/types/scale";
+
+/**
+ * The options used for creating a `KubeApi`
+ */
+export interface KubeApiOptions<T extends KubeObject, Data extends KubeJsonApiDataFor<T> = KubeJsonApiDataFor<T>>
+  extends DerivedKubeApiOptions {
+  /**
+   * base api-path for listing all resources, e.g. "/api/v1/pods"
+   *
+   * Must be provided either here or under `objectConstructor.apiBase`
+   * @deprecated should be specified by `objectConstructor`
+   */
+  apiBase?: string;
+
+  /**
+   * The constructor for the kube objects returned from the API
+   */
+  objectConstructor: KubeObjectConstructor<T, Data>;
+
+  /**
+   * Must be provided either here or under `objectConstructor.namespaced`
+   * @deprecated should be specified by `objectConstructor`
+   */
+  isNamespaced?: boolean;
+
+  /**
+   * Must be provided either here or under `objectConstructor.kind`
+   * @deprecated should be specified by `objectConstructor`
+   */
+  kind?: string;
+}
+
+export interface DerivedKubeApiOptions {
+  /**
+   * If the API uses a different API endpoint (e.g. apiBase) depending on the cluster version,
+   * fallback API bases can be listed individually.
+   *
+   * The first (existing) API base is used in the requests, if apiBase is not found.
+   *
+   * This option only has effect if {@link DerivedKubeApiOptions.checkPreferredVersion} is `true`.
+   */
+  fallbackApiBases?: string[];
+
+  /**
+   * This option is useful for protecting against newer versions on the same apiBase from being
+   * used. So that if a certain type only supports `v1`, or `v2` of some kind and then the `v3`
+   * version becomes the `preferredVersion` on the server but still has `v2` then the `v2` version
+   * will be used instead.
+   *
+   * This can help to prevent crashes in the future if the shape of a kind sufficently changes.
+   *
+   * The order is important. It should be sorted and the first entry should be the most preferable.
+   *
+   * This option only has effect if {@link DerivedKubeApiOptions.checkPreferredVersion} is `true`
+   */
+  allowedUsableVersions?: Partial<Record<string, [string, ...string[]]>>;
+
+  /**
+   * If `true` then will check all declared apiBases against the kube api server
+   * for the first accepted one.
+   *
+   * @default false
+   */
+  checkPreferredVersion?: boolean;
+
+  /**
+   * The api instance to use for making requests
+   *
+   * @default apiKube
+   */
+  request?: KubeJsonApi;
+}
+
+export interface KubeApiQueryParams {
+  watch?: boolean | number;
+  resourceVersion?: string;
+  timeoutSeconds?: number;
+  limit?: number; // doesn't work with ?watch
+  continue?: string; // might be used with ?limit from second request
+  labelSelector?: string | string[]; // restrict list of objects by their labels, e.g. labelSelector: ["label=value"]
+  fieldSelector?: string | string[]; // restrict list of objects by their fields, e.g. fieldSelector: "field=name"
+}
+
+export interface KubeApiListOptions {
+  namespace?: string;
+  reqInit?: RequestInit;
+}
+
+export interface IKubePreferredVersion {
+  preferredVersion?: {
+    version: string;
+  };
+}
+
+export interface KubeApiResource {
+  categories?: string[];
+  group?: string;
+  kind: string;
+  name: string;
+  namespaced: boolean;
+  shortNames?: string[];
+  singularName: string;
+  storageVersionHash?: string;
+  verbs: string[];
+  version?: string;
+}
+
+export interface KubeApiResourceList {
+  apiVersion?: string;
+  groupVersion?: string;
+  kind?: string;
+  resources: KubeApiResource[];
+}
+
+export interface KubeApiResourceVersion {
+  groupVersion: string;
+  version: string;
+}
+
+export interface KubeApiResourceVersionList {
+  apiVersion: string;
+  kind: string;
+  name: string;
+  preferredVersion: KubeApiResourceVersion;
+  versions: KubeApiResourceVersion[];
+}
+
+const not =
+  <T>(fn: (val: T) => boolean) =>
+  (val: T) =>
+    !fn(val);
+
+const getOrderedVersions = (
+  list: KubeApiResourceVersionList,
+  allowedUsableVersions: string[] | undefined,
+): KubeApiResourceVersion[] =>
+  [list.preferredVersion, ...list.versions.filter(not(matches(list.preferredVersion)))].filter(
+    ({ version }) => !allowedUsableVersions || allowedUsableVersions.includes(version),
+  );
+
+export type PropagationPolicy = undefined | "Orphan" | "Foreground" | "Background";
+
+export type KubeApiWatchCallback<T extends KubeJsonApiData = KubeJsonApiData> = (
+  data: IKubeWatchEvent<T> | null,
+  error: KubeStatus | Response | null | Record<string, unknown>,
+) => void;
+
+export interface KubeApiWatchOptions<Object extends KubeObject, Data extends KubeJsonApiDataFor<Object>> {
+  /**
+   * If the resource is namespaced then the default is `"default"`
+   */
+  namespace?: string;
+
+  /**
+   * This will be called when either an error occurs or some data is received
+   */
+  callback?: KubeApiWatchCallback<Data>;
+
+  /**
+   * This is a way of aborting the request
+   */
+  abortController?: AbortController;
+
+  /**
+   * The ID used for tracking within logs
+   */
+  watchId?: string;
+
+  /**
+   * @default false
+   */
+  retry?: boolean;
+
+  /**
+   * timeout in seconds
+   */
+  timeout?: number;
+}
+
+export type KubeApiPatchType = "merge" | "json" | "strategic";
+
+export const patchTypeHeaders: Record<KubeApiPatchType, string> = {
+  merge: "application/merge-patch+json",
+  json: "application/json-patch+json",
+  strategic: "application/strategic-merge-patch+json",
+};
+
+export interface ResourceDescriptor {
+  /**
+   * The name of the kubernetes resource
+   */
+  name: string;
+
+  /**
+   * The namespace that the resource lives in (if the resource is namespaced)
+   *
+   * Note: if not provided and the resource kind is namespaced, then this defaults to `"default"`
+   */
+  namespace?: string;
+}
+
+export interface SubResourceDescriptor {
+  subResource: string;
+}
+
+export type SpecificResourceDescriptor<Scope extends KubeObjectScope> = {
+  /**
+   * The name of the kubernetes resource
+   */
+  name: string;
+} & (Scope extends KubeObjectScope.Cluster
+  ? {}
+  : Scope extends KubeObjectScope.Namespace
+  ? {
+      /**
+       * The namespace that the resource lives in
+       */
+      namespace: string;
+    }
+  : {
+      namespace?: string;
+    });
+
+export type NamespacedResourceDescriptor = SpecificResourceDescriptor<KubeObjectScope.Namespace>;
+export type ClusterScopedResourceDescriptor = SpecificResourceDescriptor<KubeObjectScope.Cluster>;
+
+export interface DeleteResourceDescriptor extends ResourceDescriptor {
+  /**
+   * This determinines how child resources should be handled by kubernetes
+   *
+   * @default "Background"
+   */
+  propagationPolicy?: PropagationPolicy;
+}
+
+export interface KubeApiDependencies {
+  readonly logger: Logger;
+  readonly maybeKubeApi: KubeJsonApi | undefined;
+}
+
+export class KubeApi<
+  Object extends KubeObject = KubeObject,
+  Data extends KubeJsonApiDataFor<Object> = KubeJsonApiDataFor<Object>,
+> {
+  readonly kind: string;
+
+  readonly apiVersion: string;
+
+  @observable apiBase: string;
+
+  apiPrefix: string;
+
+  apiGroup: string;
+
+  apiVersionPreferred: string | undefined;
+
+  readonly apiResource: string;
+
+  readonly isNamespaced: boolean;
+
+  public readonly objectConstructor: KubeObjectConstructor<Object, Data>;
+
+  protected readonly request: KubeJsonApi;
+
+  protected readonly resourceVersions = new Map<string, string>();
+
+  protected readonly watchDisposer: Disposer | undefined;
+
+  private watchId = 1;
+
+  protected readonly doCheckPreferredVersion: boolean;
+
+  protected readonly fullApiPathname: string;
+
+  protected readonly fallbackApiBases: string[] | undefined;
+
+  protected readonly allowedUsableVersions: Partial<Record<string, string[]>> | undefined;
+
+  constructor(protected readonly dependencies: KubeApiDependencies, opts: KubeApiOptions<Object, Data>) {
+    const {
+      objectConstructor,
+      request = this.dependencies.maybeKubeApi,
+      kind = objectConstructor.kind,
+      isNamespaced,
+      apiBase: fullApiPathname = objectConstructor.apiBase,
+      checkPreferredVersion: doCheckPreferredVersion = false,
+      fallbackApiBases,
+      allowedUsableVersions,
+    } = opts;
+
+    assert(
+      fullApiPathname,
+      "apiBase MUST be provided either via KubeApiOptions.apiBase or KubeApiOptions.objectConstructor.apiBase",
+    );
+    assert(request, "request MUST be provided if not in a cluster page frame context");
+
+    const parsedApi = parseKubeApi(fullApiPathname);
+
+    assert(parsedApi, "apiBase MUST be a valid kube api pathname");
+
+    const { apiBase, apiPrefix, apiGroup, apiVersion, resource } = parsedApi;
+
+    assert(kind, "kind MUST be provided either via KubeApiOptions.kind or KubeApiOptions.objectConstructor.kind");
+    assert(apiPrefix, "apiBase MUST be parsable as a kubeApi selfLink style string");
+
+    this.doCheckPreferredVersion = doCheckPreferredVersion;
+    this.fallbackApiBases = fallbackApiBases;
+    this.allowedUsableVersions = allowedUsableVersions;
+    this.fullApiPathname = fullApiPathname;
+    this.kind = kind;
+    this.isNamespaced = isNamespaced ?? objectConstructor.namespaced ?? false;
+    this.apiBase = apiBase;
+    this.apiPrefix = apiPrefix;
+    this.apiGroup = apiGroup;
+    this.apiVersion = apiVersion;
+    this.apiResource = resource;
+    this.request = request;
+    this.objectConstructor = objectConstructor;
+    makeObservable(this);
+  }
+
+  get apiVersionWithGroup() {
+    return [this.apiGroup, this.apiVersionPreferred ?? this.apiVersion].filter(Boolean).join("/");
+  }
+
+  /**
+   * Returns the latest API prefix/group that contains the required resource.
+   * First tries fullApiPathname, then urls in order from fallbackApiBases.
+   */
+  private async getLatestApiPrefixGroup() {
+    // Note that this.fullApiPathname is the "full" url, whereas this.apiBase is parsed
+    const rawApiBases = [this.fullApiPathname, this.objectConstructor.apiBase, ...(this.fallbackApiBases ?? [])].filter(
+      isDefined,
+    );
+    const apiBases = new Set(rawApiBases);
+
+    for (const apiUrl of apiBases) {
+      const parsedApi = parseKubeApi(apiUrl);
+
+      if (!parsedApi) {
+        continue;
+      }
+
+      try {
+        const { apiPrefix, apiGroup, resource } = parsedApi;
+        const list = (await this.request.get(`${apiPrefix}/${apiGroup}`)) as KubeApiResourceVersionList;
+        const resourceVersions = getOrderedVersions(list, this.allowedUsableVersions?.[apiGroup]);
+
+        for (const resourceVersion of resourceVersions) {
+          const { resources } = (await this.request.get(
+            `${apiPrefix}/${resourceVersion.groupVersion}`,
+          )) as KubeApiResourceList;
+
+          if (resources.some(({ name }) => name === resource)) {
+            return {
+              apiPrefix,
+              apiGroup,
+              apiVersionPreferred: resourceVersion.version,
+            };
+          }
+        }
+      } catch {
+        // ignore exception to try next url
+      }
+    }
+
+    throw new Error(`Can't find working API for the Kubernetes resource ${this.apiResource}`);
+  }
+
+  protected async checkPreferredVersion() {
+    if (!this.doCheckPreferredVersion && (this.fallbackApiBases || this.allowedUsableVersions)) {
+      throw new Error(
+        "checkPreferredVersion must be enabled if either fallbackApiBases or allowedUsableVersions are set in KubeApi",
+      );
+    }
+
+    if (this.doCheckPreferredVersion && this.apiVersionPreferred === undefined) {
+      const { apiPrefix, apiGroup, apiVersionPreferred } = await this.getLatestApiPrefixGroup();
+
+      this.apiPrefix = apiPrefix;
+      this.apiGroup = apiGroup;
+      this.apiVersionPreferred = apiVersionPreferred;
+      this.apiBase = this.computeApiBase();
+    }
+  }
+
+  // eslint-disable-next-line @typescript-eslint/default-param-last
+  setResourceVersion(namespace = "", newVersion: string) {
+    this.resourceVersions.set(namespace, newVersion);
+  }
+
+  getResourceVersion(namespace = "") {
+    return this.resourceVersions.get(namespace);
+  }
+
+  async refreshResourceVersion(params?: KubeApiListOptions) {
+    return this.list(params, { limit: 1 });
+  }
+
+  private computeApiBase(): string {
+    return createKubeApiURL({
+      apiPrefix: this.apiPrefix,
+      apiVersion: this.apiVersionWithGroup,
+      resource: this.apiResource,
+    });
+  }
+
+  /**
+   * This method differs from {@link formatUrlForNotListing} because this treats `""` as "all namespaces"
+   * NOTE: This is also useful for watching
+   * @param namespace The namespace to list in or `""` for all namespaces
+   */
+  formatUrlForListing(namespace: string | undefined, query?: Partial<KubeApiQueryParams>) {
+    const resourcePath = createKubeApiURL({
+      apiPrefix: this.apiPrefix,
+      apiVersion: this.apiVersionWithGroup,
+      resource: this.apiResource,
+      namespace: this.isNamespaced ? namespace ?? "default" : undefined,
+    });
+
+    return resourcePath + (query ? `?${stringify(this.normalizeQuery(query))}` : "");
+  }
+
+  /**
+   * Format a URL pathname and query for acting upon a specific resource.
+   */
+  formatUrlForNotListing({ name, namespace }: Partial<ResourceDescriptor> = {}) {
+    return createKubeApiURL({
+      apiPrefix: this.apiPrefix,
+      apiVersion: this.apiVersionWithGroup,
+      resource: this.apiResource,
+      namespace: this.isNamespaced ? namespace || "default" : undefined,
+      name,
+    });
+  }
+
+  /**
+   * @deprecated use {@link formatUrlForNotListing} or {@link formatUrlForListing} instead
+   */
+  getUrl(resource?: Partial<ResourceDescriptor>, query?: Partial<KubeApiQueryParams>) {
+    if (query) {
+      return this.formatUrlForListing(resource?.namespace, query);
+    }
+
+    return this.formatUrlForNotListing(resource);
+  }
+
+  protected normalizeQuery(query: Partial<KubeApiQueryParams> = {}) {
+    if (query.labelSelector) {
+      query.labelSelector = [query.labelSelector].flat().join(",");
+    }
+
+    if (query.fieldSelector) {
+      query.fieldSelector = [query.fieldSelector].flat().join(",");
+    }
+
+    return query;
+  }
+
+  protected parseResponse(data: unknown, namespace?: string): Object | Object[] | null {
+    if (!data) {
+      return null;
+    }
+
+    const KubeObjectConstructor = this.objectConstructor;
+
+    // process items list response, check before single item since there is overlap
+    if (isJsonApiDataList(data, isPartialJsonApiData)) {
+      const { apiVersion, items, metadata } = data;
+
+      this.setResourceVersion(namespace, metadata.resourceVersion);
+      this.setResourceVersion("", metadata.resourceVersion);
+
+      return items
+        .map((item) => {
+          if (item.metadata) {
+            this.ensureMetadataSelfLink(item.metadata);
+          } else {
+            return undefined;
+          }
+
+          const object = new KubeObjectConstructor({
+            ...(item as Data),
+            kind: this.kind,
+            apiVersion,
+          });
+
+          return object;
+        })
+        .filter(isDefined);
+    }
+
+    // process a single item
+    if (isJsonApiData(data)) {
+      this.ensureMetadataSelfLink(data.metadata);
+
+      return new KubeObjectConstructor(data as Data);
+    }
+
+    if (!Array.isArray(data)) {
+      return null;
+    }
+
+    // custom apis might return array for list response, e.g. users, groups, etc.
+    return data
+      .filter(isJsonApiData)
+      .map((data) => (this.ensureMetadataSelfLink(data.metadata), data))
+      .map((data) => new KubeObjectConstructor(data as Data));
+  }
+
+  private ensureMetadataSelfLink<T extends { selfLink?: string; namespace?: string; name: string }>(
+    metadata: T,
+  ): asserts metadata is T & { selfLink: string } {
+    metadata.selfLink ||= createKubeApiURL({
+      apiPrefix: this.apiPrefix,
+      apiVersion: this.apiVersionWithGroup,
+      resource: this.apiResource,
+      namespace: metadata.namespace,
+      name: metadata.name,
+    });
+  }
+
+  async list(
+    { namespace = "", reqInit }: KubeApiListOptions = {},
+    query?: KubeApiQueryParams,
+  ): Promise<Object[] | null> {
+    await this.checkPreferredVersion();
+
+    const url = this.formatUrlForListing(namespace);
+    const res = await this.request.get(url, { query }, reqInit);
+    const parsed = this.parseResponse(res, namespace);
+
+    if (Array.isArray(parsed)) {
+      return parsed;
+    }
+
+    if (!parsed) {
+      return null;
+    }
+
+    throw new Error(`GET multiple request to ${url} returned not an array: ${JSON.stringify(parsed)}`);
+  }
+
+  async get(desc: ResourceDescriptor, query?: KubeApiQueryParams): Promise<Object | null> {
+    await this.checkPreferredVersion();
+
+    const url = this.formatUrlForNotListing(desc);
+    const res = await this.request.get(url, { query });
+    const parsed = this.parseResponse(res);
+
+    if (Array.isArray(parsed)) {
+      throw new Error(`GET single request to ${url} returned an array: ${JSON.stringify(parsed)}`);
+    }
+
+    return parsed;
+  }
+
+  async create(
+    { name, namespace }: Partial<ResourceDescriptor>,
+    partialData?: PartialDeep<Object>,
+  ): Promise<Object | null> {
+    await this.checkPreferredVersion();
+
+    const apiUrl = this.formatUrlForNotListing({ namespace });
+    const data = merge(partialData, {
+      kind: this.kind,
+      apiVersion: this.apiVersionWithGroup,
+      metadata: {
+        name,
+        namespace,
+      },
+    });
+    const res = await this.request.post(apiUrl, { data });
+    const parsed = this.parseResponse(res);
+
+    if (Array.isArray(parsed)) {
+      throw new Error(`POST request to ${apiUrl} returned an array: ${JSON.stringify(parsed)}`);
+    }
+
+    return parsed;
+  }
+
+  async update({ name, namespace }: ResourceDescriptor, data: PartialDeep<Object>): Promise<Object | null> {
+    await this.checkPreferredVersion();
+    const apiUrl = this.formatUrlForNotListing({ namespace, name });
+
+    const res = await this.request.put(apiUrl, {
+      data: merge(data, {
+        metadata: {
+          name,
+          namespace,
+        },
+      }),
+    });
+    const parsed = this.parseResponse(res);
+
+    if (Array.isArray(parsed)) {
+      throw new Error(`PUT request to ${apiUrl} returned an array: ${JSON.stringify(parsed)}`);
+    }
+
+    return parsed;
+  }
+
+  /**
+   * An internal method for requesting the `/scale` sub-resource if it exists
+   */
+  protected async getResourceScale(desc: ResourceDescriptor): Promise<Scale> {
+    await this.checkPreferredVersion();
+    const apiUrl = this.formatUrlForNotListing(desc);
+
+    const res = await this.request.get(`${apiUrl}/scale`);
+
+    return res as Scale;
+  }
+
+  /**
+   * An internal method for requesting the `/scale` sub-resource if it exists
+   */
+  protected async scaleResource(desc: ResourceDescriptor, data: ScaleCreateOptions): Promise<Scale> {
+    await this.checkPreferredVersion();
+    const apiUrl = this.formatUrlForNotListing(desc);
+
+    const res = await this.request.patch(
+      `${apiUrl}/scale`,
+      {
+        data: { ...data },
+      },
+      {
+        headers: {
+          "content-type": patchTypeHeaders.merge,
+        },
+      },
+    );
+
+    return res as Scale;
+  }
+
+  async patch(desc: ResourceDescriptor, data: PartialDeep<Object>): Promise<Object | null>;
+  async patch(
+    desc: ResourceDescriptor,
+    data: PartialDeep<Object>,
+    strategy: "strategic" | "merge",
+  ): Promise<Object | null>;
+  async patch(desc: ResourceDescriptor, data: Patch, strategy: "json"): Promise<Object | null>;
+  async patch(
+    desc: ResourceDescriptor,
+    data: PartialDeep<Object> | Patch,
+    strategy: KubeApiPatchType,
+  ): Promise<Object | null>;
+  async patch(
+    desc: ResourceDescriptor,
+    data: PartialDeep<Object> | Patch,
+    strategy: KubeApiPatchType = "strategic",
+  ): Promise<Object | null> {
+    await this.checkPreferredVersion();
+    const apiUrl = this.formatUrlForNotListing(desc);
+
+    const res = await this.request.patch(
+      apiUrl,
+      { data },
+      {
+        headers: {
+          "content-type": patchTypeHeaders[strategy],
+        },
+      },
+    );
+    const parsed = this.parseResponse(res);
+
+    if (Array.isArray(parsed)) {
+      throw new Error(`PATCH request to ${apiUrl} returned an array: ${JSON.stringify(parsed)}`);
+    }
+
+    return parsed;
+  }
+
+  /**
+   * Some k8s resources might implement special "delete" (e.g. pod.api)
+   * See also: https://kubernetes.io/docs/concepts/scheduling-eviction/api-eviction/
+   * By default should work same as delete()
+   */
+  async evict(desc: DeleteResourceDescriptor): Promise<KubeStatus | KubeObject | unknown> {
+    return this.delete(desc);
+  }
+
+  async delete({ propagationPolicy = "Background", ...desc }: DeleteResourceDescriptor) {
+    await this.checkPreferredVersion();
+    const apiUrl = this.formatUrlForNotListing(desc);
+
+    return this.request.del(apiUrl, {
+      query: {
+        propagationPolicy,
+      },
+    });
+  }
+
+  getWatchUrl(namespace?: string, query: KubeApiQueryParams = {}) {
+    return this.formatUrlForListing(namespace, {
+      watch: 1,
+      resourceVersion: this.getResourceVersion(namespace),
+      ...query,
+    });
+  }
+
+  watch(opts?: KubeApiWatchOptions<Object, Data>): Disposer {
+    let errorReceived = false;
+    let timedRetry: NodeJS.Timeout;
+    const {
+      namespace,
+      callback = noop as KubeApiWatchCallback<Data>,
+      retry = false,
+      timeout = 600,
+      watchId = `${this.kind.toLowerCase()}-${this.watchId++}`,
+    } = opts ?? {};
+
+    // Create AbortController for this request
+    const abortController = new WrappedAbortController(opts?.abortController);
+    const watchUrl = this.getWatchUrl(namespace);
+
+    abortController.signal.addEventListener("abort", () => {
+      this.dependencies.logger.info(`[KUBE-API] watch (${watchId}) aborted ${watchUrl}`);
+      clearTimeout(timedRetry);
+    });
+
+    const requestParams = timeout ? { query: { timeoutSeconds: timeout } } : {};
+    const responsePromise = this.request.getResponse(watchUrl, requestParams, {
+      signal: abortController.signal,
+    });
+
+    this.dependencies.logger.info(
+      `[KUBE-API] watch (${watchId}) ${retry === true ? "retried" : "started"} ${watchUrl}`,
+    );
+
+    responsePromise
+      .then((response) => {
+        // True if the current watch request was retried
+        let requestRetried = false;
+
+        if (!response.ok) {
+          this.dependencies.logger.warn(`[KUBE-API] watch (${watchId}) error response ${watchUrl}`, {
+            status: response.status,
+          });
+
+          return callback(null, response);
+        }
+
+        // Add mechanism to retry in case timeoutSeconds is set but the watch wasn't timed out.
+        // This can happen if e.g. network is offline and AWS NLB is used.
+        if (timeout) {
+          setTimeout(() => {
+            // We only retry if we haven't retried, haven't aborted and haven't received k8s error
+            if (requestRetried || abortController.signal.aborted || errorReceived) {
+              return;
+            }
+
+            // Close current request
+            abortController.abort();
+
+            this.dependencies.logger.info(`[KUBE-API] Watch timeout set, but not retried, retrying now`);
+
+            requestRetried = true;
+
+            // Clearing out any possible timeout, although we don't expect this to be set
+            clearTimeout(timedRetry);
+            this.watch({ ...opts, namespace, callback, watchId, retry: true });
+            // We wait longer than the timeout, as we expect the request to be retried with timeoutSeconds
+          }, timeout * 1000 * 1.1);
+        }
+
+        if (!response.body || !response.body.readable) {
+          if (!response.body) {
+            this.dependencies.logger.warn(`[KUBE-API]: watch (${watchId}) did not return a body`);
+          }
+          requestRetried = true;
+
+          clearTimeout(timedRetry);
+          timedRetry = setTimeout(() => {
+            // we did not get any kubernetes errors so let's retry
+            this.watch({ ...opts, namespace, callback, watchId, retry: true });
+          }, 1000);
+
+          return;
+        }
+
+        for (const eventName of ["end", "close", "error"]) {
+          // eslint-disable-next-line @typescript-eslint/no-loop-func
+          response.body.on(eventName, () => {
+            // We only retry if we haven't retried, haven't aborted and haven't received k8s error
+            // kubernetes errors (=errorReceived set) should be handled in a callback
+            if (requestRetried || abortController.signal.aborted || errorReceived) {
+              return;
+            }
+
+            this.dependencies.logger.info(`[KUBE-API] watch (${watchId}) ${eventName} ${watchUrl}`);
+
+            requestRetried = true;
+
+            clearTimeout(timedRetry);
+            timedRetry = setTimeout(() => {
+              // we did not get any kubernetes errors so let's retry
+              this.watch({ ...opts, namespace, callback, watchId, retry: true });
+            }, 1000);
+          });
+        }
+
+        byline(response.body).on("data", (line: string) => {
+          try {
+            const event = JSON.parse(line) as IKubeWatchEvent<Data>;
+
+            if (event.type === "ERROR" && isKubeStatusData(event.object)) {
+              errorReceived = true;
+
+              return callback(null, new KubeStatus(event.object));
+            }
+
+            this.modifyWatchEvent(event);
+            callback(event, null);
+          } catch (ignore) {
+            // ignore parse errors
+          }
+        });
+      })
+      .catch((error: unknown) => {
+        if (!abortController.signal.aborted) {
+          this.dependencies.logger.error(`[KUBE-API] watch (${watchId}) threw ${watchUrl}`, error);
+        }
+        callback(null, error as Record<string, unknown>);
+      });
+
+    return () => {
+      abortController.abort();
+    };
+  }
+
+  protected modifyWatchEvent(event: IKubeWatchEvent<KubeJsonApiData<KubeObjectMetadata>>) {
+    if (event.type === "ERROR") {
+      return;
+    }
+
+    this.ensureMetadataSelfLink(event.object.metadata);
+
+    const { namespace, resourceVersion } = event.object.metadata;
+
+    assert(resourceVersion, "watch events failed to return resourceVersion from kube api");
+
+    this.setResourceVersion(namespace, resourceVersion);
+    this.setResourceVersion("", resourceVersion);
+  }
+}