Merge pull request #472 from snyk/fix/include_errors_for_k8s_api_calls

Fix/include errors for k8s api calls

Author: agranado2k

src/supervisor/kuberenetes-api-wrappers.ts

@@ -46,6 +46,14 @@ function shouldRetryRequest(err: IRequestError, attempt: number): boolean {
     return false;
   }
 
+  if (err.code === 'ECONNREFUSED') {
+    return true;
+  }
+
+  if (err.code === 'ETIMEDOUT') {
+    return true;
+  }
+
   if (!(err.response)) {
     return false;
   }

src/supervisor/types.ts

@@ -14,6 +14,7 @@ export enum WorkloadKind {
 }
 
 export interface IRequestError {
+  code?: string;
   response?: IncomingMessage;
 }
 

src/supervisor/watchers/handlers/index.ts

@@ -10,6 +10,7 @@ import { replicaSetWatchHandler } from './replica-set';
 import { replicationControllerWatchHandler } from './replication-controller';
 import { statefulSetWatchHandler } from './stateful-set';
 import { k8sApi, kubeConfig } from '../../cluster';
+import * as kubernetesApiWrappers from '../../kuberenetes-api-wrappers';
 import { IWorkloadWatchMetadata, FALSY_WORKLOAD_NAME_MARKER } from './types';
 
 /**
@@ -98,7 +99,8 @@ export function setupInformer(namespace: string, workloadKind: WorkloadKind) {
   const listMethod = workloadMetadata.listFactory(namespace);
   const loggedListMethod = async () => {
     try {
-      return await listMethod();
+      return await kubernetesApiWrappers.retryKubernetesApiRequest(
+        () => listMethod());
     } catch (err) {
       logger.error({err, namespace, workloadKind}, 'error while listing entities on namespace');
       throw err;

src/supervisor/watchers/index.ts

@@ -6,6 +6,7 @@ import logger = require('../../common/logger');
 import { WorkloadKind } from '../types';
 import { setupInformer } from './handlers';
 import { kubeConfig, k8sApi } from '../cluster';
+import * as kubernetesApiWrappers from '../kuberenetes-api-wrappers';
 import { kubernetesInternalNamespaces } from './internal-namespaces';
 
 /**
@@ -52,7 +53,8 @@ function setupWatchesForCluster(): void {
     '/api/v1/namespaces',
     async () => {
       try {
-        return await k8sApi.coreClient.listNamespace();
+        return await kubernetesApiWrappers.retryKubernetesApiRequest(
+          () => k8sApi.coreClient.listNamespace());
       } catch (err) {
         logger.error({err}, 'error while listing namespaces');
         throw err;

test/system/kind.test.ts

@@ -69,6 +69,30 @@ tap.test('Kubernetes-Monitor with KinD', async (t) => {
   ]);
 
   // Setup nocks
+  nock(/https\:\/\/127\.0\.0\.1\:\d+/, { allowUnmocked: true})
+    .get('/api/v1/namespaces')
+    .times(1)
+    .replyWithError({
+      code: 'ECONNREFUSED'
+    })
+    .get('/api/v1/namespaces')
+    .times(1)
+    .replyWithError({
+      code: 'ETIMEDOUT'
+    });
+
+  nock(/https\:\/\/127\.0\.0\.1\:\d+/, { allowUnmocked: true})
+    .get('/apis/apps/v1/namespaces/snyk-monitor/deployments')
+    .times(1)
+    .replyWithError({
+      code: 'ECONNREFUSED'
+    })
+    .get('/apis/apps/v1/namespaces/snyk-monitor/deployments')
+    .times(1)
+    .replyWithError({
+      code: 'ETIMEDOUT'
+    });
+
   nock('https://kubernetes-upstream.snyk.io')
     .post('/api/v1/workload')
     .times(1)

test/unit/supervisor/kubernetes-api-wrappers.test.ts

@@ -46,6 +46,84 @@ tap.test('calculateSleepSeconds', async (t) => {
   );
 });
 
+tap.test('retryKubernetesApiRequest for ECONNREFUSED error', async (t) => {
+  const retryableErrorResponse = {code: 'ECONNREFUSED'};
+
+  await t.rejects(
+    () => kubernetesApiWrappers.retryKubernetesApiRequest(() => Promise.reject(retryableErrorResponse)),
+    'eventually throws on repeated retryable error responses',
+  );
+
+  let failures = 0;
+  const functionThatFailsJustEnoughTimes = () => {
+    if (failures < kubernetesApiWrappers.ATTEMPTS_MAX - 1) {
+      failures +=1;
+      return Promise.reject(retryableErrorResponse);
+    }
+    return Promise.resolve('egg');
+  };
+
+  const successfulResponse = await kubernetesApiWrappers.retryKubernetesApiRequest(functionThatFailsJustEnoughTimes);
+  t.equals(
+    successfulResponse,
+    'egg',
+    'keeps retrying on ECONNREFUSED as long as we don\'t cross max attempts',
+  );
+
+  failures = 0;
+  const functionThatFailsOneTooManyTimes = () => {
+    if (failures < kubernetesApiWrappers.ATTEMPTS_MAX) {
+      failures +=1;
+      return Promise.reject(retryableErrorResponse);
+    }
+    return Promise.resolve('egg');
+  };
+
+  await t.rejects(
+    () => kubernetesApiWrappers.retryKubernetesApiRequest(functionThatFailsOneTooManyTimes),
+    'failure more than the maximum, rejects, even for a retryable error response',
+  );
+});
+
+tap.test('retryKubernetesApiRequest for ETIMEDOUT error', async (t) => {
+  const retryableErrorResponse = {code: 'ETIMEDOUT'};
+
+  await t.rejects(
+    () => kubernetesApiWrappers.retryKubernetesApiRequest(() => Promise.reject(retryableErrorResponse)),
+    'eventually throws on repeated retryable error responses',
+  );
+
+  let failures = 0;
+  const functionThatFailsJustEnoughTimes = () => {
+    if (failures < kubernetesApiWrappers.ATTEMPTS_MAX - 1) {
+      failures +=1;
+      return Promise.reject(retryableErrorResponse);
+    }
+    return Promise.resolve('egg');
+  };
+
+  const successfulResponse = await kubernetesApiWrappers.retryKubernetesApiRequest(functionThatFailsJustEnoughTimes);
+  t.equals(
+    successfulResponse,
+    'egg',
+    'keeps retrying on ETIMEDOUT as long as we don\'t cross max attempts',
+  );
+
+  failures = 0;
+  const functionThatFailsOneTooManyTimes = () => {
+    if (failures < kubernetesApiWrappers.ATTEMPTS_MAX) {
+      failures +=1;
+      return Promise.reject(retryableErrorResponse);
+    }
+    return Promise.resolve('egg');
+  };
+
+  await t.rejects(
+    () => kubernetesApiWrappers.retryKubernetesApiRequest(functionThatFailsOneTooManyTimes),
+    'failure more than the maximum, rejects, even for a retryable error response',
+  );
+});
+
 tap.test('retryKubernetesApiRequest for retryable errors', async (t) => {
   const retryableErrorResponse = {response: {statusCode: 429}};