feat: paginate list() calls to the Kubernetes API

Ensures we control how hard we hit the Kubernetes API. Instead of listing everything in that namespace, do it in batches of 100.

As part of the pagination, remove unneeded metadata like "managedFields" from the listed workloads.

Author: ivanstanev

src/supervisor/kuberenetes-api-wrappers.ts

@@ -71,7 +71,7 @@ export async function retryKubernetesApiRequestIndefinitely<ResponseType>(
 }
 
 export function calculateSleepSeconds(
-  httpResponse: http.IncomingMessage,
+  httpResponse?: http.IncomingMessage,
 ): number {
   let sleepSeconds = DEFAULT_SLEEP_SEC;
   if (

src/supervisor/watchers/handlers/cron-job.ts

@@ -1,7 +1,28 @@
-import { V1beta1CronJob } from '@kubernetes/client-node';
+import { V1beta1CronJob, V1beta1CronJobList } from '@kubernetes/client-node';
 import { deleteWorkload } from './workload';
 import { WorkloadKind } from '../../types';
 import { FALSY_WORKLOAD_NAME_MARKER } from './types';
+import { IncomingMessage } from 'http';
+import { k8sApi } from '../../cluster';
+import { paginatedList } from './pagination';
+
+export async function paginatedCronJobList(namespace: string): Promise<{
+  response: IncomingMessage;
+  body: V1beta1CronJobList;
+}> {
+  const v1CronJobList = new V1beta1CronJobList();
+  v1CronJobList.apiVersion = 'batch/v1beta1';
+  v1CronJobList.kind = 'CronJobList';
+  v1CronJobList.items = new Array<V1beta1CronJob>();
+
+  return await paginatedList(
+    namespace,
+    v1CronJobList,
+    k8sApi.batchUnstableClient.listNamespacedCronJob.bind(
+      k8sApi.batchUnstableClient,
+    ),
+  );
+}
 
 export async function cronJobWatchHandler(
   cronJob: V1beta1CronJob,

src/supervisor/watchers/handlers/daemon-set.ts

@@ -1,7 +1,26 @@
-import { V1DaemonSet } from '@kubernetes/client-node';
+import { V1DaemonSet, V1DaemonSetList } from '@kubernetes/client-node';
 import { deleteWorkload } from './workload';
 import { WorkloadKind } from '../../types';
 import { FALSY_WORKLOAD_NAME_MARKER } from './types';
+import { IncomingMessage } from 'http';
+import { k8sApi } from '../../cluster';
+import { paginatedList } from './pagination';
+
+export async function paginatedDaemonSetList(namespace: string): Promise<{
+  response: IncomingMessage;
+  body: V1DaemonSetList;
+}> {
+  const v1DaemonSetList = new V1DaemonSetList();
+  v1DaemonSetList.apiVersion = 'apps/v1';
+  v1DaemonSetList.kind = 'DaemonSetList';
+  v1DaemonSetList.items = new Array<V1DaemonSet>();
+
+  return await paginatedList(
+    namespace,
+    v1DaemonSetList,
+    k8sApi.appsClient.listNamespacedDaemonSet.bind(k8sApi.appsClient),
+  );
+}
 
 export async function daemonSetWatchHandler(
   daemonSet: V1DaemonSet,

src/supervisor/watchers/handlers/deployment-config.ts

@@ -1,6 +1,50 @@
+import { IncomingMessage } from 'http';
 import { deleteWorkload } from './workload';
 import { WorkloadKind } from '../../types';
-import { FALSY_WORKLOAD_NAME_MARKER, V1DeploymentConfig } from './types';
+import {
+  FALSY_WORKLOAD_NAME_MARKER,
+  V1DeploymentConfig,
+  V1DeploymentConfigList,
+} from './types';
+import { paginatedList } from './pagination';
+import { k8sApi } from '../../cluster';
+
+export async function paginatedDeploymentConfigList(
+  namespace: string,
+): Promise<{
+  response: IncomingMessage;
+  body: V1DeploymentConfigList;
+}> {
+  const v1DeploymentConfigList = new V1DeploymentConfigList();
+  v1DeploymentConfigList.apiVersion = 'apps.openshift.io/v1';
+  v1DeploymentConfigList.kind = 'DeploymentConfigList';
+  v1DeploymentConfigList.items = new Array<V1DeploymentConfig>();
+
+  return await paginatedList(
+    namespace,
+    v1DeploymentConfigList,
+    async (
+      namespace: string,
+      pretty?: string,
+      _allowWatchBookmarks?: boolean,
+      _continue?: string,
+      fieldSelector?: string,
+      labelSelector?: string,
+      limit?: number,
+    ) =>
+      k8sApi.customObjectsClient.listNamespacedCustomObject(
+        'apps.openshift.io',
+        'v1',
+        namespace,
+        'deploymentconfigs',
+        pretty,
+        _continue,
+        fieldSelector,
+        labelSelector,
+        limit,
+      ) as any,
+  );
+}
 
 export async function deploymentConfigWatchHandler(
   deploymentConfig: V1DeploymentConfig,

src/supervisor/watchers/handlers/deployment.ts

@@ -1,7 +1,26 @@
-import { V1Deployment } from '@kubernetes/client-node';
+import { V1Deployment, V1DeploymentList } from '@kubernetes/client-node';
 import { deleteWorkload } from './workload';
 import { WorkloadKind } from '../../types';
 import { FALSY_WORKLOAD_NAME_MARKER } from './types';
+import { IncomingMessage } from 'http';
+import { k8sApi } from '../../cluster';
+import { paginatedList } from './pagination';
+
+export async function paginatedDeploymentList(namespace: string): Promise<{
+  response: IncomingMessage;
+  body: V1DeploymentList;
+}> {
+  const v1DeploymentList = new V1DeploymentList();
+  v1DeploymentList.apiVersion = 'apps/v1';
+  v1DeploymentList.kind = 'DeploymentList';
+  v1DeploymentList.items = new Array<V1Deployment>();
+
+  return await paginatedList(
+    namespace,
+    v1DeploymentList,
+    k8sApi.appsClient.listNamespacedDeployment.bind(k8sApi.appsClient),
+  );
+}
 
 export async function deploymentWatchHandler(
   deployment: V1Deployment,

src/supervisor/watchers/handlers/index.ts

@@ -9,15 +9,24 @@ import {
 
 import { logger } from '../../../common/logger';
 import { WorkloadKind } from '../../types';
-import { podWatchHandler, podDeletedHandler } from './pod';
-import { cronJobWatchHandler } from './cron-job';
-import { daemonSetWatchHandler } from './daemon-set';
-import { deploymentWatchHandler } from './deployment';
-import { jobWatchHandler } from './job';
-import { replicaSetWatchHandler } from './replica-set';
-import { replicationControllerWatchHandler } from './replication-controller';
-import { statefulSetWatchHandler } from './stateful-set';
-import { deploymentConfigWatchHandler } from './deployment-config';
+import { podWatchHandler, podDeletedHandler, paginatedPodList } from './pod';
+import { cronJobWatchHandler, paginatedCronJobList } from './cron-job';
+import { daemonSetWatchHandler, paginatedDaemonSetList } from './daemon-set';
+import { deploymentWatchHandler, paginatedDeploymentList } from './deployment';
+import { jobWatchHandler, paginatedJobList } from './job';
+import { paginatedReplicaSetList, replicaSetWatchHandler } from './replica-set';
+import {
+  paginatedReplicationControllerList,
+  replicationControllerWatchHandler,
+} from './replication-controller';
+import {
+  paginatedStatefulSetList,
+  statefulSetWatchHandler,
+} from './stateful-set';
+import {
+  deploymentConfigWatchHandler,
+  paginatedDeploymentConfigList,
+} from './deployment-config';
 import { k8sApi, kubeConfig } from '../../cluster';
 import * as kubernetesApiWrappers from '../../kuberenetes-api-wrappers';
 import { IWorkloadWatchMetadata, FALSY_WORKLOAD_NAME_MARKER } from './types';
@@ -49,64 +58,57 @@ const workloadWatchMetadata: Readonly<IWorkloadWatchMetadata> = {
       [UPDATE]: podWatchHandler,
       [DELETE]: podDeletedHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.coreClient.listNamespacedPod(namespace),
+    listFactory: (namespace) => () => paginatedPodList(namespace),
   },
   [WorkloadKind.ReplicationController]: {
     endpoint: '/api/v1/watch/namespaces/{namespace}/replicationcontrollers',
     handlers: {
       [DELETE]: replicationControllerWatchHandler,
     },
     listFactory: (namespace) => () =>
-      k8sApi.coreClient.listNamespacedReplicationController(namespace),
+      paginatedReplicationControllerList(namespace),
   },
   [WorkloadKind.CronJob]: {
     endpoint: '/apis/batch/v1beta1/watch/namespaces/{namespace}/cronjobs',
     handlers: {
       [DELETE]: cronJobWatchHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.batchUnstableClient.listNamespacedCronJob(namespace),
+    listFactory: (namespace) => () => paginatedCronJobList(namespace),
   },
   [WorkloadKind.Job]: {
     endpoint: '/apis/batch/v1/watch/namespaces/{namespace}/jobs',
     handlers: {
       [DELETE]: jobWatchHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.batchClient.listNamespacedJob(namespace),
+    listFactory: (namespace) => () => paginatedJobList(namespace),
   },
   [WorkloadKind.DaemonSet]: {
     endpoint: '/apis/apps/v1/watch/namespaces/{namespace}/daemonsets',
     handlers: {
       [DELETE]: daemonSetWatchHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.appsClient.listNamespacedDaemonSet(namespace),
+    listFactory: (namespace) => () => paginatedDaemonSetList(namespace),
   },
   [WorkloadKind.Deployment]: {
     endpoint: '/apis/apps/v1/watch/namespaces/{namespace}/deployments',
     handlers: {
       [DELETE]: deploymentWatchHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.appsClient.listNamespacedDeployment(namespace),
+    listFactory: (namespace) => () => paginatedDeploymentList(namespace),
   },
   [WorkloadKind.ReplicaSet]: {
     endpoint: '/apis/apps/v1/watch/namespaces/{namespace}/replicasets',
     handlers: {
       [DELETE]: replicaSetWatchHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.appsClient.listNamespacedReplicaSet(namespace),
+    listFactory: (namespace) => () => paginatedReplicaSetList(namespace),
   },
   [WorkloadKind.StatefulSet]: {
     endpoint: '/apis/apps/v1/watch/namespaces/{namespace}/statefulsets',
     handlers: {
       [DELETE]: statefulSetWatchHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.appsClient.listNamespacedStatefulSet(namespace),
+    listFactory: (namespace) => () => paginatedStatefulSetList(namespace),
   },
   [WorkloadKind.DeploymentConfig]: {
     /** https://docs.openshift.com/container-platform/4.7/rest_api/workloads_apis/deploymentconfig-apps-openshift-io-v1.html */
@@ -115,13 +117,7 @@ const workloadWatchMetadata: Readonly<IWorkloadWatchMetadata> = {
     handlers: {
       [DELETE]: deploymentConfigWatchHandler,
     },
-    listFactory: (namespace) => () =>
-      k8sApi.customObjectsClient.listNamespacedCustomObject(
-        'apps.openshift.io',
-        'v1',
-        namespace,
-        'deploymentconfigs',
-      ),
+    listFactory: (namespace) => () => paginatedDeploymentConfigList(namespace),
   },
 };
 

src/supervisor/watchers/handlers/job.ts

@@ -1,7 +1,26 @@
-import { V1Job } from '@kubernetes/client-node';
+import { V1Job, V1JobList } from '@kubernetes/client-node';
 import { deleteWorkload } from './workload';
 import { WorkloadKind } from '../../types';
 import { FALSY_WORKLOAD_NAME_MARKER } from './types';
+import { IncomingMessage } from 'http';
+import { k8sApi } from '../../cluster';
+import { paginatedList } from './pagination';
+
+export async function paginatedJobList(namespace: string): Promise<{
+  response: IncomingMessage;
+  body: V1JobList;
+}> {
+  const v1JobList = new V1JobList();
+  v1JobList.apiVersion = 'batch/v1';
+  v1JobList.kind = 'JobList';
+  v1JobList.items = new Array<V1Job>();
+
+  return await paginatedList(
+    namespace,
+    v1JobList,
+    k8sApi.batchClient.listNamespacedJob.bind(k8sApi.batchClient),
+  );
+}
 
 export async function jobWatchHandler(job: V1Job): Promise<void> {
   if (