Merge pull request #265 from snyk/feat/workload_info_cache

feat: workload info caching

Author: shaimendel

config.default.json

@@ -7,6 +7,10 @@
     "MAX_SIZE": 10000,
     "MAX_AGE_MS": 86400000
   },
+  "WORKLOADS_SCANNED_CACHE": {
+    "MAX_SIZE": 10000,
+    "MAX_AGE_MS": 60000
+  },
   "WORKLOADS_TO_SCAN_QUEUE_WORKER_COUNT": 10,
   "METADATA_TO_SEND_QUEUE_WORKER_COUNT": 10,
   "INTEGRATION_ID": "203210a3-1de0-4ed3-b6a6-3acd24b71639",

package-lock.json

@@ -31,6 +31,7 @@
     "@types/lru-cache": "^5.1.0",
     "@types/needle": "^2.0.4",
     "@types/node": "^10.12.5",
+    "@types/sinon": "^7.5.1",
     "async": "^2.6.2",
     "bunyan": "^1.8.12",
     "child-process-promise": "^2.2.1",
@@ -49,6 +50,7 @@
     "@typescript-eslint/parser": "^2.6.1",
     "eslint": "^6.6.0",
     "eslint-config-prettier": "^6.5.0",
+    "sinon": "^8.0.1",
     "tap": "github:snyk/node-tap#alternative-runtimes",
     "ts-node": "^8.1.0",
     "tsc-watch": "^1.0.30"

package.json

@@ -98,7 +98,14 @@ 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);
-    metadataToSendQueue.push({workloadMetadataPayload});
+    const workloadLocator = workloadMetadataPayload.workloadLocator;
+    const workloadKey = `${workloadLocator.namespace}/${workloadLocator.type}/${workloadLocator.name}`;
+    const workloadRevision = workloadMember.revision ? workloadMember.revision.toString() : ''; // this is actually the observed generation
+    if (state.workloadsAlreadyScanned.get(workloadKey) !== workloadRevision) { // either not exists or different
+      state.workloadsAlreadyScanned.set(workloadKey, workloadRevision); // empty string takes zero bytes and is !== undefined
+      metadataToSendQueue.push({workloadMetadataPayload});
+    }
+
     const workloadName = workloadMember.name;
     const workloadWorker = new WorkloadWorker(workloadName);
     handleReadyPod(workloadWorker, workloadMetadata);

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

@@ -2,7 +2,7 @@
 import * as lruCache from 'lru-cache';
 import config = require('./common/config');
 
-const lruCacheOptions = {
+const imagesLruCacheOptions = {
   // limit cache size so we don't exceed memory limit
   max: config.IMAGES_SCANNED_CACHE.MAX_SIZE,
   // limit cache life so if our backend loses track of an image's data,
@@ -11,8 +11,18 @@ const lruCacheOptions = {
   updateAgeOnGet: false,
 };
 
+const workloadsLruCacheOptions = {
+  // limit cache size so we don't exceed memory limit
+  max: config.WORKLOADS_SCANNED_CACHE.MAX_SIZE,
+  // limit cache life so if our backend loses track of an image's data,
+  // eventually we will report again for that image, if it's still relevant
+  maxAge: config.WORKLOADS_SCANNED_CACHE.MAX_AGE_MS,
+  updateAgeOnGet: false,
+};
+
 const state = {
-  imagesAlreadyScanned: new lruCache<string, string>(lruCacheOptions),
+  imagesAlreadyScanned: new lruCache<string, string>(imagesLruCacheOptions),
+  workloadsAlreadyScanned: new lruCache<string, string>(workloadsLruCacheOptions),
 };
 
 export = state;

src/state.ts

@@ -0,0 +1,60 @@
+import * as tap from 'tap';
+import sinon = require('sinon');
+import * as sleep from 'sleep-promise';
+import * as fs from 'fs';
+import * as YAML from 'yaml';
+import async = require('async');
+
+import { V1PodSpec, V1Pod } from '@kubernetes/client-node';
+import transmitterTypes = require('../../src/transmitter/types');
+import * as metadataExtractor from '../../src/kube-scanner/metadata-extractor';
+
+let pushCallCount = 0;
+sinon.stub(async, 'queue').returns({ error: () => { }, push: () => pushCallCount++ } as any);
+
+import * as pod from '../../src/kube-scanner/watchers/handlers/pod';
+
+tap.test('image and workload image cache', async (t) => {
+  const podSpecFixture = fs.readFileSync('./test/fixtures/pod-spec.json', 'utf8');
+  const podSpec: V1PodSpec = YAML.parse(podSpecFixture);
+  const workloadMetadata: transmitterTypes.IWorkload[] = [
+    {
+      type: 'type',
+      name: 'workloadName',
+      namespace: 'spacename',
+      labels: undefined,
+      annotations: undefined,
+      uid: 'udi',
+      specLabels: undefined,
+      specAnnotations: undefined,
+      containerName: 'contener',
+      imageName: 'myImage:tag',
+      imageId: 'does this matter?',
+      cluster: 'grapefruit',
+      revision: 1,
+      podSpec,
+    },
+  ];
+
+  sinon.stub(metadataExtractor, 'buildMetadataForWorkload').resolves(workloadMetadata);
+
+  t.teardown(() => {
+    (async['queue'] as any).restore();
+    (metadataExtractor['buildMetadataForWorkload'] as any).restore();
+  });
+
+  const podFixture = fs.readFileSync('./test/fixtures/sidecar-containers/pod.yaml', 'utf8');
+  const podObject: V1Pod = YAML.parse(podFixture);
+  await pod.podWatchHandler(podObject);
+  await sleep(500);
+  t.equals(pushCallCount, 2, 'pushed data to send');
+
+  await pod.podWatchHandler(podObject);
+  await sleep(500);
+  t.equals(pushCallCount, 2, 'cached info, no data pushed to send');
+
+  workloadMetadata[0].imageId = 'newImageName';
+  await pod.podWatchHandler(podObject);
+  await sleep(1000);
+  t.equals(pushCallCount, 3, 'new image parsed, workload is cached');
+});