Merge pull request #242 from snyk/fix/queue-workload-metadata

Fix/queue workload metadata

Author: Amir Moualem

config.default.json

@@ -8,6 +8,7 @@
     "MAX_AGE_MS": 86400000
   },
   "WORKLOADS_TO_SCAN_QUEUE_WORKER_COUNT": 10,
+  "METADATA_TO_SEND_QUEUE_WORKER_COUNT": 10,
   "INTEGRATION_ID": "203210a3-1de0-4ed3-b6a6-3acd24b71639",
   "DEFAULT_KUBERNETES_UPSTREAM_URL": "https://kubernetes-upstream.snyk.io"
 }

src/kube-scanner/watchers/handlers/pod.ts

@@ -27,7 +27,7 @@ function deleteFailedKeysFromState(keys): void {
   }
 }
 
-async function queueWorker(task, callback): Promise<void> {
+async function queueWorkerWorkloadScan(task, callback): Promise<void> {
   const {workloadWorker, workloadMetadata, imageKeys} = task;
   try {
     await workloadWorker.process(workloadMetadata);
@@ -37,12 +37,23 @@ async function queueWorker(task, callback): Promise<void> {
   }
 }
 
-const workloadsToScanQueue = async.queue(queueWorker, config.WORKLOADS_TO_SCAN_QUEUE_WORKER_COUNT);
+const workloadsToScanQueue = async.queue(queueWorkerWorkloadScan, config.WORKLOADS_TO_SCAN_QUEUE_WORKER_COUNT);
+
+async function queueWorkerMetadataSender(task, callback): Promise<void> {
+  const {workloadMetadataPayload} = task;
+  await sendWorkloadMetadata(workloadMetadataPayload);
+}
+
+const metadataToSendQueue = async.queue(queueWorkerMetadataSender, config.METADATA_TO_SEND_QUEUE_WORKER_COUNT);
 
 workloadsToScanQueue.error(function(err, task) {
   logger.error({err, task}, 'error processing a workload in the pod handler 1');
 });
 
+metadataToSendQueue.error(function(err, task) {
+  logger.error({err, task}, 'error processing a workload metadata send task');
+});
+
 function handleReadyPod(workloadWorker: WorkloadWorker, workloadMetadata: IWorkload[]): void {
   const imagesToScan: IWorkload[] = [];
   const imageKeys: string[] = [];
@@ -87,7 +98,7 @@ export async function podWatchHandler(pod: V1Pod): Promise<void> {
     // every element contains the workload information, so we can get it from the first one
     const workloadMember = workloadMetadata[0];
     const workloadMetadataPayload = constructHomebaseWorkloadMetadataPayload(workloadMember);
-    await sendWorkloadMetadata(workloadMetadataPayload);
+    metadataToSendQueue.push({workloadMetadataPayload});
     const workloadName = workloadMember.name;
     const workloadWorker = new WorkloadWorker(workloadName);
     handleReadyPod(workloadWorker, workloadMetadata);