Merge pull request #1035 from snyk/feat/delete-workloads-from-scan-queue

[RUN-2125] Feat/delete workloads from scan queue

Author: ivanstanev

config.default.json

@@ -11,9 +11,9 @@
     "MAX_SIZE": 10000,
     "MAX_AGE_MS": 60000
   },
-  "WORKERS_COUNT": 10,
+  "WORKERS_COUNT": 5,
   "REQUEST_QUEUE_LENGTH": 2,
-  "QUEUE_LENGTH_LOG_FREQUENCY_MINUTES": 15,
+  "QUEUE_LENGTH_LOG_FREQUENCY_MINUTES": 1,
   "INTEGRATION_ID": "",
   "DEFAULT_KUBERNETES_UPSTREAM_URL": "https://kubernetes-upstream.snyk.io",
   "MAX_RETRY_BACKOFF_DURATION_SECONDS": 300

src/supervisor/metadata-extractor.ts

@@ -5,11 +5,14 @@ import {
   V1ContainerStatus,
   V1PodSpec,
 } from '@kubernetes/client-node';
-import { IWorkload, ILocalWorkloadLocator } from '../transmitter/types';
 import { currentClusterName } from './cluster';
-import { IKubeObjectMetadata } from './types';
+import { WorkloadKind } from './types';
 import { getSupportedWorkload, getWorkloadReader } from './workload-reader';
 import { logger } from '../common/logger';
+import { config } from '../common/config';
+
+import type { IKubeObjectMetadata } from './types';
+import type { IWorkload, ILocalWorkloadLocator } from '../transmitter/types';
 
 const loopingThreshold = 20;
 
@@ -170,6 +173,17 @@ export async function buildMetadataForWorkload(
     );
   }
 
+  const hasJobOwnerRef = pod.metadata?.ownerReferences?.find(
+    (owner) => owner.kind === WorkloadKind.Job,
+  );
+  if (hasJobOwnerRef && config.SKIP_K8S_JOBS) {
+    logger.info(
+      { podMetadata: pod.metadata },
+      'pod associated with job but jobs are skipped from processing. not building metadata.',
+    );
+    return undefined;
+  }
+
   const podOwner: IKubeObjectMetadata | undefined = await findParentWorkload(
     pod.spec,
     pod.metadata.ownerReferences,

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),
     ]);
   }