feat: remove workloads from scan queue when they are deleted in K8s

Up until now the queue of workloads to scan would keep growing indefinitely as we only appended new workloads to it.
However, once workloads are deleted from Kubernetes, they no longer need to be in the queue.

This didn't cause any issues with trying to process deleted workloads, because the queue handler would detect that the workload is gone and continue processing other items. However, it created alerts that the queue grows indefinitely and caused the snyk-monitor to process some work that was never going to succeed.

Author: ivanstanev

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

@@ -20,6 +20,7 @@ import {
 import { logger } from '../../../common/logger';
 import { retryKubernetesApiRequest } from '../../kuberenetes-api-wrappers';
 import { trimWorkload } from '../../workload-sanitization';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedCronJobList(
   namespace: string,
@@ -117,6 +118,7 @@ export async function cronJobWatchHandler(
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

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

@@ -11,6 +11,7 @@ import {
   kubernetesObjectToWorkloadAlreadyScanned,
 } from '../../../state';
 import { trimWorkload } from '../../workload-sanitization';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedDaemonSetList(
   namespace: string,
@@ -71,6 +72,7 @@ export async function daemonSetWatchHandler(
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

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

@@ -15,6 +15,7 @@ import {
 } from '../../../state';
 import { retryKubernetesApiRequest } from '../../kuberenetes-api-wrappers';
 import { logger } from '../../../common/logger';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedDeploymentConfigList(
   namespace: string,
@@ -112,6 +113,7 @@ export async function deploymentConfigWatchHandler(
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

src/supervisor/watchers/handlers/deployment.ts

@@ -11,6 +11,7 @@ import {
   kubernetesObjectToWorkloadAlreadyScanned,
 } from '../../../state';
 import { trimWorkload } from '../../workload-sanitization';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedDeploymentList(
   namespace: string,
@@ -71,6 +72,7 @@ export async function deploymentWatchHandler(
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

src/supervisor/watchers/handlers/job.ts

@@ -11,6 +11,7 @@ import {
   kubernetesObjectToWorkloadAlreadyScanned,
 } from '../../../state';
 import { trimWorkload } from '../../workload-sanitization';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedJobList(namespace: string): Promise<{
   response: IncomingMessage;
@@ -65,6 +66,7 @@ export async function jobWatchHandler(job: V1Job): Promise<void> {
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

src/supervisor/watchers/handlers/pod.ts

@@ -1,12 +1,9 @@
 import { V1Pod, V1PodList } from '@kubernetes/client-node';
 import { IncomingMessage } from 'http';
-import * as async from 'async';
 
 import { logger } from '../../../common/logger';
-import { config } from '../../../common/config';
-import { processWorkload } from '../../../scanner';
 import { sendWorkloadMetadata } from '../../../transmitter';
-import { IWorkload, Telemetry } from '../../../transmitter/types';
+import { IWorkload } from '../../../transmitter/types';
 import { constructWorkloadMetadata } from '../../../transmitter/payload';
 import { buildMetadataForWorkload } from '../../metadata-extractor';
 import { PodPhase } from '../types';
@@ -25,70 +22,7 @@ import { deleteWorkload } from './workload';
 import { k8sApi } from '../../cluster';
 import { paginatedClusterList, paginatedNamespacedList } from './pagination';
 import { trimWorkload } from '../../workload-sanitization';
-
-export interface ImagesToScanQueueData {
-  workloadMetadata: IWorkload[];
-  /** The timestamp when this workload was added to the image scan queue. */
-  enqueueTimestampMs: number;
-}
-
-async function queueWorkerWorkloadScan(
-  task: ImagesToScanQueueData,
-  callback,
-): Promise<void> {
-  const { workloadMetadata, enqueueTimestampMs } = task;
-  /** Represents how long this workload spent waiting in the queue to be processed. */
-  const enqueueDurationMs = Date.now() - enqueueTimestampMs;
-  const telemetry: Partial<Telemetry> = {
-    enqueueDurationMs,
-    queueSize: workloadsToScanQueue.length(),
-  };
-  try {
-    await processWorkload(workloadMetadata, telemetry);
-  } catch (err) {
-    logger.error(
-      { err, task },
-      'error processing a workload in the pod handler 2',
-    );
-    const imageIds = workloadMetadata.map((workload) => workload.imageId);
-    const workload = {
-      // every workload metadata references the same workload, grab it from the first one
-      ...workloadMetadata[0],
-      imageIds,
-    };
-    await deleteWorkloadImagesAlreadyScanned(workload);
-  }
-}
-
-const workloadsToScanQueue = async.queue<ImagesToScanQueueData>(
-  queueWorkerWorkloadScan,
-  config.WORKERS_COUNT,
-);
-
-workloadsToScanQueue.error(function (err, task) {
-  logger.error(
-    { err, task },
-    'error processing a workload in the pod handler 1',
-  );
-});
-
-function reportQueueSize(): void {
-  try {
-    const queueDataToReport: { [key: string]: any } = {};
-    queueDataToReport.workloadsToScanLength = workloadsToScanQueue.length();
-    logger.debug(queueDataToReport, 'queue sizes report');
-  } catch (err) {
-    logger.debug({ err }, 'failed logging queue sizes');
-  }
-}
-
-// Report the queue size shortly after the snyk-monitor starts.
-setTimeout(reportQueueSize, 1 * 60 * 1000).unref();
-// Additionally, periodically report every X minutes.
-setInterval(
-  reportQueueSize,
-  config.QUEUE_LENGTH_LOG_FREQUENCY_MINUTES * 60 * 1000,
-).unref();
+import { deleteWorkloadFromScanQueue, workloadsToScanQueue } from './queue';
 
 async function handleReadyPod(workloadMetadata: IWorkload[]): Promise<void> {
   const workloadToScan: IWorkload[] = [];
@@ -111,8 +45,10 @@ async function handleReadyPod(workloadMetadata: IWorkload[]): Promise<void> {
     workloadToScan.push(workload);
   }
 
+  const workload = workloadToScan[0];
   if (workloadToScan.length > 0) {
     workloadsToScanQueue.push({
+      key: workload.uid,
       workloadMetadata: workloadToScan,
       enqueueTimestampMs: Date.now(),
     });
@@ -223,6 +159,7 @@ export async function podDeletedHandler(pod: V1Pod): Promise<void> {
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

src/supervisor/watchers/handlers/queue.ts

@@ -0,0 +1,81 @@
+import * as async from 'async';
+import { config } from '../../../common/config';
+
+import { logger } from '../../../common/logger';
+import { processWorkload } from '../../../scanner';
+import { deleteWorkloadImagesAlreadyScanned } from '../../../state';
+import type { IWorkload, Telemetry } from '../../../transmitter/types';
+
+interface ImagesToScanQueueData {
+  /** Identifies the workload in the queue. */
+  key: string;
+  workloadMetadata: IWorkload[];
+  /** The timestamp when this workload was added to the image scan queue. */
+  enqueueTimestampMs: number;
+}
+
+export const workloadsToScanQueue = async.queue<ImagesToScanQueueData>(
+  queueWorkerWorkloadScan,
+  config.WORKERS_COUNT,
+);
+
+export async function deleteWorkloadFromScanQueue(workload: {
+  uid: string;
+}): Promise<void> {
+  workloadsToScanQueue.remove(
+    (item) => item.data && item.data.key === workload.uid,
+  );
+}
+
+workloadsToScanQueue.error(function (err, task) {
+  logger.error(
+    { err, task },
+    'error processing a workload in the pod handler 1',
+  );
+});
+
+async function queueWorkerWorkloadScan(
+  task: ImagesToScanQueueData,
+  callback,
+): Promise<void> {
+  const { workloadMetadata, enqueueTimestampMs } = task;
+  /** Represents how long this workload spent waiting in the queue to be processed. */
+  const enqueueDurationMs = Date.now() - enqueueTimestampMs;
+  const telemetry: Partial<Telemetry> = {
+    enqueueDurationMs,
+    queueSize: workloadsToScanQueue.length(),
+  };
+  try {
+    await processWorkload(workloadMetadata, telemetry);
+  } catch (err) {
+    logger.error(
+      { err, task },
+      'error processing a workload in the pod handler 2',
+    );
+    const imageIds = workloadMetadata.map((workload) => workload.imageId);
+    const workload = {
+      // every workload metadata references the same workload, grab it from the first one
+      ...workloadMetadata[0],
+      imageIds,
+    };
+    await deleteWorkloadImagesAlreadyScanned(workload);
+  }
+}
+
+function reportQueueSize(): void {
+  try {
+    const queueDataToReport: { [key: string]: any } = {};
+    queueDataToReport.workloadsToScanLength = workloadsToScanQueue.length();
+    logger.debug(queueDataToReport, 'queue sizes report');
+  } catch (err) {
+    logger.debug({ err }, 'failed logging queue sizes');
+  }
+}
+
+// Report the queue size shortly after the snyk-monitor starts.
+setTimeout(reportQueueSize, 1 * 60 * 1000).unref();
+// Additionally, periodically report every X minutes.
+setInterval(
+  reportQueueSize,
+  config.QUEUE_LENGTH_LOG_FREQUENCY_MINUTES * 60 * 1000,
+).unref();

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

@@ -11,6 +11,7 @@ import {
   kubernetesObjectToWorkloadAlreadyScanned,
 } from '../../../state';
 import { trimWorkload } from '../../workload-sanitization';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedReplicaSetList(
   namespace: string,
@@ -72,6 +73,7 @@ export async function replicaSetWatchHandler(
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

src/supervisor/watchers/handlers/replication-controller.ts

@@ -14,6 +14,7 @@ import {
   kubernetesObjectToWorkloadAlreadyScanned,
 } from '../../../state';
 import { trimWorkload } from '../../workload-sanitization';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedReplicationControllerList(
   namespace: string,
@@ -80,6 +81,7 @@ export async function replicationControllerWatchHandler(
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }
 

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

@@ -11,6 +11,7 @@ import {
   kubernetesObjectToWorkloadAlreadyScanned,
 } from '../../../state';
 import { trimWorkload } from '../../workload-sanitization';
+import { deleteWorkloadFromScanQueue } from './queue';
 
 export async function paginatedNamespacedStatefulSetList(
   namespace: string,
@@ -71,6 +72,7 @@ export async function statefulSetWatchHandler(
           .filter((container) => container.image !== undefined)
           .map((container) => container.image!),
       }),
+      deleteWorkloadFromScanQueue(workloadAlreadyScanned),
     ]);
   }