Merge pull request #255 from snyk/feat/retry-transmissions

Feat/retry transmissions

Author: Amir Moualem

.eslintrc.json

@@ -11,6 +11,7 @@
     "prettier/@typescript-eslint"
   ],
   "rules": {
+    "@typescript-eslint/no-use-before-define": "off",
     "@typescript-eslint/interface-name-prefix": "off",
     "@typescript-eslint/await-thenable": "error",
     "@typescript-eslint/ban-ts-ignore": "error",

package-lock.json

@@ -36,6 +36,7 @@
     "child-process-promise": "^2.2.1",
     "lru-cache": "^5.1.1",
     "needle": "^2.4.0",
+    "sleep-promise": "^8.0.1",
     "snyk-config": "3.0.0",
     "snyk-docker-plugin": "1.34.0",
     "source-map-support": "^0.5.9",
@@ -48,7 +49,6 @@
     "@typescript-eslint/parser": "^2.6.1",
     "eslint": "^6.6.0",
     "eslint-config-prettier": "^6.5.0",
-    "sleep-promise": "^8.0.1",
     "tap": "github:snyk/node-tap#alternative-runtimes",
     "ts-node": "^8.1.0",
     "tsc-watch": "^1.0.30"

package.json

@@ -1,30 +1,25 @@
-import needle = require('needle');
+import * as needle from 'needle';
+import { NeedleResponse, NeedleHttpVerbs } from 'needle';
+import * as sleep from 'sleep-promise';
 import * as config from '../common/config';
 import logger = require('../common/logger');
-import { IDeleteWorkloadPayload, IDepGraphPayload, IWorkloadMetadataPayload } from './types';
+import { IDeleteWorkloadPayload, IDepGraphPayload, IWorkloadMetadataPayload, IResponseWithAttempts } from './types';
 
 const upstreamUrl = config.INTEGRATION_API || config.DEFAULT_KUBERNETES_UPSTREAM_URL;
 
-function isSuccessStatusCode(statusCode: number | undefined): boolean {
-  return statusCode !== undefined && statusCode > 100 && statusCode < 400;
-}
-
 export async function sendDepGraph(...payloads: IDepGraphPayload[]): 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 result = await needle('post', `${upstreamUrl}/api/v1/dependency-graph`, payload, {
-          json: true,
-          compressed: true,
-        },
-      );
-
-      if (!isSuccessStatusCode(result.statusCode)) {
-        throw new Error(`${result.statusCode} ${result.statusMessage}`);
+      const {response, attempt} = await retryRequest('post', `${upstreamUrl}/api/v1/dependency-graph`, payload);
+      if (!isSuccessStatusCode(response.statusCode)) {
+        throw new Error(`${response.statusCode} ${response.statusMessage}`);
+      } else {
+        logger.info({payload: payloadWithoutDepGraph, attempt}, 'dependency graph sent upstream successfully')
       }
     } catch (error) {
-      // Intentionally removing dependencyGraph as it would be too big to log
-      // eslint-disable-next-line @typescript-eslint/no-unused-vars
-      const { dependencyGraph, ...payloadWithoutDepGraph } = payload;
       logger.error({error, payload: payloadWithoutDepGraph}, 'could not send the dependency scan result to Homebase');
     }
   }
@@ -33,16 +28,12 @@ export async function sendDepGraph(...payloads: IDepGraphPayload[]): Promise<voi
 export async function sendWorkloadMetadata(payload: IWorkloadMetadataPayload): Promise<void> {
     try {
       logger.info({workloadLocator: payload.workloadLocator}, 'attempting to send workload metadata upstream')
-      const result = await needle('post', `${upstreamUrl}/api/v1/workload`, payload, {
-          json: true,
-          compressed: true,
-        },
-      );
 
-      if (!isSuccessStatusCode(result.statusCode)) {
-        throw new Error(`${result.statusCode} ${result.statusMessage}`);
+      const {response, attempt} = await retryRequest('post', `${upstreamUrl}/api/v1/workload`, payload);
+      if (!isSuccessStatusCode(response.statusCode)) {
+        throw new Error(`${response.statusCode} ${response.statusMessage}`);
       } else {
-        logger.info({workloadLocator: payload.workloadLocator}, 'workload metadata sent upstream successfully')
+        logger.info({workloadLocator: payload.workloadLocator, attempt}, 'workload metadata sent upstream successfully')
       }
     } catch (error) {
       logger.error({error, workloadLocator: payload.workloadLocator}, 'could not send workload metadata to Homebase');
@@ -51,22 +42,48 @@ export async function sendWorkloadMetadata(payload: IWorkloadMetadataPayload): P
 
 export async function deleteHomebaseWorkload(payload: IDeleteWorkloadPayload): Promise<void> {
   try {
-    const result = await needle('delete', `${upstreamUrl}/api/v1/workload`, payload, {
-        json: true,
-        compressed: true,
-      },
-    );
-
-    if (result.statusCode === 404) {
+    const {response, attempt} = await retryRequest('delete', `${upstreamUrl}/api/v1/workload`, payload);
+    if (response.statusCode === 404) {
       const msg = 'attempted to delete a workload Homebase could not find, maybe we are still building it?';
       logger.info({payload}, msg);
       return;
     }
-
-    if (!isSuccessStatusCode(result.statusCode)) {
-      throw new Error(`${result.statusCode} ${result.statusMessage}`);
+    if (!isSuccessStatusCode(response.statusCode)) {
+      throw new Error(`${response.statusCode} ${response.statusMessage}`);
+    } else {
+      logger.info({workloadLocator: payload.workloadLocator, attempt}, 'workload deleted successfully')
     }
   } catch (error) {
     logger.error({error, payload}, 'could not send workload to delete to Homebase');
   }
 }
+
+function isSuccessStatusCode(statusCode: number | undefined): boolean {
+  return statusCode !== undefined && statusCode > 100 && statusCode < 400;
+}
+
+async function retryRequest(verb: NeedleHttpVerbs, url: string, payload: object): Promise<IResponseWithAttempts> {
+  const retry = {
+    attempts: 3,
+    intervalSeconds: 2,
+  }
+  const options = {
+    json: true,
+    compressed: true,
+  };
+
+  let response: NeedleResponse;
+  let attempt = 1;
+
+  response = await needle(verb, url, payload, options);
+  for (; attempt <= retry.attempts; attempt++) {
+    if (response.statusCode === 502 && attempt + 1 < retry.attempts) {
+      await sleep(retry.intervalSeconds * 1000);
+      response = await needle(verb, url, payload, options);
+    } else {
+      break;
+    }
+  }
+
+  return {response, attempt};
+}

src/transmitter/index.ts

@@ -1,4 +1,5 @@
 import { V1PodSpec } from '@kubernetes/client-node';
+import { NeedleResponse } from 'needle';
 
 interface StringMap { [key: string]: string; }
 
@@ -66,3 +67,8 @@ export interface IWorkload {
   cluster: string;
   podSpec: V1PodSpec;
 }
+
+export interface IResponseWithAttempts {
+  response: NeedleResponse;
+  attempt: number;
+}