Merge pull request #664 from snyk/RUN-1388/throttle-homebase-requests

Author: jmickey

.circleci/config.yml

@@ -812,7 +812,8 @@ jobs:
                 command: |
                     npm run lint &&
                     npm run build &&
-                    npm run test:unit
+                    npm run test:unit:tap &&
+                    npm run test:unit:jest
                 name: Unit tests
             - run:
                 command: |

.circleci/config/jobs/@jobs.yml

@@ -92,7 +92,8 @@ unit_tests:
         command: |
           npm run lint &&
           npm run build &&
-          npm run test:unit
+          npm run test:unit:tap &&
+          npm run test:unit:jest
     - run:
         name: Notify Slack on failure
         command: |

.tool-versions

@@ -0,0 +1 @@
+nodejs lts-erbium

config.default.json

@@ -12,7 +12,7 @@
     "MAX_AGE_MS": 60000
   },
   "WORKLOADS_TO_SCAN_QUEUE_WORKER_COUNT": 10,
-  "METADATA_TO_SEND_QUEUE_WORKER_COUNT": 10,
+  "REQUEST_QUEUE_LENGTH": 2,
   "QUEUE_LENGTH_LOG_FREQUENCY_MINUTES": 15,
   "INTEGRATION_ID": "",
   "DEFAULT_KUBERNETES_UPSTREAM_URL": "https://kubernetes-upstream.snyk.io"

package-lock.json

@@ -4,8 +4,9 @@
   "main": "dist/index.js",
   "scripts": {
     "pretest": "./scripts/docker/build-image.sh",
-    "test": "npm run lint && npm run build && npm run test:unit && npm run test:integration:kind:helm",
-    "test:unit": "NODE_ENV=test tap test/unit --timeout=300",
+    "test": "npm run lint && npm run build && npm run test:unit:tap && npm run test:unit:jest && npm run test:integration:kind:helm",
+    "test:unit:tap": "NODE_ENV=test tap test/unit/*.test.ts test/unit/**/*.test.ts --timeout=300",
+    "test:unit:jest": "jest --logHeapUsage --ci test/unit",
     "test:system": "jest --logHeapUsage --ci --maxWorkers=1 test/system",
     "test:integration:kind:yaml": "DEPLOYMENT_TYPE=YAML TEST_PLATFORM=kind CREATE_CLUSTER=true jest --logHeapUsage --ci --maxWorkers=1 test/integration/kubernetes.spec.ts",
     "test:integration:kind:helm": "DEPLOYMENT_TYPE=Helm TEST_PLATFORM=kind CREATE_CLUSTER=true jest --logHeapUsage --ci --maxWorkers=1 test/integration/kubernetes.spec.ts",
@@ -29,7 +30,6 @@
   "dependencies": {
     "@kubernetes/client-node": "^0.14.0",
     "@snyk/dep-graph": "^1.21.0",
-    "@types/async": "^3.2.3",
     "@types/child-process-promise": "^2.2.1",
     "@types/lru-cache": "^5.1.0",
     "@types/needle": "^2.0.4",
@@ -56,6 +56,7 @@
     "@types/jest": "^26.0.20",
     "@typescript-eslint/eslint-plugin": "^2.22.0",
     "@typescript-eslint/parser": "^2.22.0",
+    "@types/async": "^3.2.5",
     "eslint": "^6.8.0",
     "eslint-config-prettier": "^6.10.0",
     "jest": "^26.6.3",

package.json

@@ -39,26 +39,14 @@ async function queueWorkerWorkloadScan(task, callback): Promise<void> {
 
 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');
-});
-
 setInterval(() => {
   try {
     const queueDataToReport: {[key: string]: any} = {};
     queueDataToReport.workloadsToScanLength = workloadsToScanQueue.length();
-    queueDataToReport.metadataToSendLength = metadataToSendQueue.length();
     logger.info(queueDataToReport, 'queue sizes report');
   } catch (err) {
     logger.warn({err}, 'failed logging queue sizes');
@@ -114,7 +102,7 @@ export async function podWatchHandler(pod: V1Pod): Promise<void> {
     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});
+      await sendWorkloadMetadata(workloadMetadataPayload);
     }
 
     handleReadyPod(workloadMetadata);

src/supervisor/watchers/handlers/pod.ts

@@ -1,3 +1,4 @@
+import { queue } from 'async';
 import * as needle from 'needle';
 import { NeedleResponse, NeedleHttpVerbs, NeedleOptions } from 'needle';
 import * as sleep from 'sleep-promise';
@@ -14,15 +15,37 @@ import {
 } from './types';
 import { getProxyAgent } from './proxy';
 
+interface HomebaseRequest {
+  method: NeedleHttpVerbs;
+  url: string;
+  payload:
+    IDependencyGraphPayload |
+    ScanResultsPayload |
+    IWorkloadMetadataPayload |
+    IDeleteWorkloadPayload;
+}
+
 const upstreamUrl = config.INTEGRATION_API || config.DEFAULT_KUBERNETES_UPSTREAM_URL;
 
+// Async queue wraps around the call to retryRequest in order to limit
+// the number of requests in flight to Homebase at any one time.
+const reqQueue = queue(async function(req: HomebaseRequest) {
+  return await retryRequest(req.method, req.url, req.payload);
+}, config.REQUEST_QUEUE_LENGTH);
+
 export async function sendDepGraph(...payloads: IDependencyGraphPayload[]): Promise<void> {
   for (const payload of payloads) {
     // Intentionally removing dependencyGraph as it would be too big to log
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
     const { dependencyGraph, ...payloadWithoutDepGraph } = payload;
     try {
-      const {response, attempt} = await retryRequest('post', `${upstreamUrl}/api/v1/dependency-graph`, payload);
+      const request: HomebaseRequest = {
+        method: 'post',
+        url: `${upstreamUrl}/api/v1/dependency-graph`,
+        payload: payload,
+      };
+
+      const { response, attempt } = await reqQueue.pushAsync(request);
       if (!isSuccessStatusCode(response.statusCode)) {
         throw new Error(`${response.statusCode} ${response.statusMessage}`);
       } else {
@@ -40,7 +63,13 @@ export async function sendScanResults(payloads: ScanResultsPayload[]): Promise<b
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
     const { scanResults, ...payloadWithoutScanResults } = payload;
     try {
-      const {response, attempt} = await retryRequest('post', `${upstreamUrl}/api/v1/scan-results`, payload);
+      const request: HomebaseRequest = {
+        method: 'post',
+        url: `${upstreamUrl}/api/v1/scan-results`,
+        payload: payload,
+      };
+
+      const { response, attempt } = await reqQueue.pushAsync(request);
       if (!isSuccessStatusCode(response.statusCode)) {
         throw new Error(`${response.statusCode} ${response.statusMessage}`);
       } else {
@@ -59,7 +88,13 @@ export async function sendWorkloadMetadata(payload: IWorkloadMetadataPayload): P
     try {
       logger.info({workloadLocator: payload.workloadLocator}, 'attempting to send workload metadata upstream');
 
-      const {response, attempt} = await retryRequest('post', `${upstreamUrl}/api/v1/workload`, payload);
+      const request: HomebaseRequest = {
+        method: 'post',
+        url: `${upstreamUrl}/api/v1/workload`,
+        payload: payload,
+      };
+
+      const { response, attempt } = await reqQueue.pushAsync(request);
       if (!isSuccessStatusCode(response.statusCode)) {
         throw new Error(`${response.statusCode} ${response.statusMessage}`);
       } else {
@@ -96,7 +131,13 @@ export async function sendWorkloadAutoImportPolicy(payload: WorkloadAutoImportPo
 
 export async function deleteWorkload(payload: IDeleteWorkloadPayload): Promise<void> {
   try {
-    const {response, attempt} = await retryRequest('delete', `${upstreamUrl}/api/v1/workload`, payload);
+    const request: HomebaseRequest = {
+      method: 'delete',
+      url: `${upstreamUrl}/api/v1/workload`,
+      payload: payload,
+    };
+
+    const { response, attempt } = await reqQueue.pushAsync(request);
     if (response.statusCode === 404) {
       // TODO: maybe we're still building it?
       const msg = 'attempted to delete a workload the Upstream service could not find';

src/transmitter/index.ts

@@ -1,20 +1,18 @@
-import * as tap from 'tap';
-import * as sinon from 'sinon';
-import * as sleep from 'sleep-promise';
+import * as async from 'async';
 import * as fs from 'fs';
+import * as sleep from 'sleep-promise';
 import * as YAML from 'yaml';
-import async = require('async');
 
 import { V1PodSpec, V1Pod } from '@kubernetes/client-node';
 import { IWorkload } from '../../../src/transmitter/types';
 import * as metadataExtractor from '../../../src/supervisor/metadata-extractor';
 
 let pushCallCount = 0;
-sinon.stub(async, 'queue').returns({ error: () => { }, push: () => pushCallCount++ } as any);
+const asyncQueueSpy = jest.spyOn(async, 'queue').mockReturnValue({ error: () => {}, push: () => pushCallCount++ } as any);
 
 import * as pod from '../../../src/supervisor/watchers/handlers/pod';
 
-tap.test('image and workload image cache', async (t) => {
+describe('image and workload image cache', () => {
   const podSpecFixture = fs.readFileSync('./test/fixtures/pod-spec.json', 'utf8');
   const podSpec: V1PodSpec = YAML.parse(podSpecFixture);
   const workloadMetadata: IWorkload[] = [
@@ -36,25 +34,33 @@ tap.test('image and workload image cache', async (t) => {
     },
   ];
 
-  sinon.stub(metadataExtractor, 'buildMetadataForWorkload').resolves(workloadMetadata);
 
-  t.teardown(() => {
-    (async['queue'] as any).restore();
-    (metadataExtractor['buildMetadataForWorkload'] as any).restore();
+  const buildMetadataSpy = jest.spyOn(metadataExtractor, 'buildMetadataForWorkload').mockResolvedValue(workloadMetadata);
+
+  afterAll(() => {
+    asyncQueueSpy.mockRestore();
+    buildMetadataSpy.mockRestore();
   });
 
   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');
+
+  it('pushed data to send', async () => {
+    await pod.podWatchHandler(podObject);
+    await sleep(500);
+    expect(pushCallCount).toEqual(1);
+  });
+
+  it('cached info, no data pushed to send', async () => {
+    await pod.podWatchHandler(podObject);
+    await sleep(500);
+    expect(pushCallCount).toEqual(1);
+  });
+
+  it('new image parsed, workload is cached', async () => {
+    workloadMetadata[0].imageId = 'newImageName';
+    await pod.podWatchHandler(podObject);
+    await sleep(1000);
+    expect(pushCallCount).toEqual(2);
+  });
 });