Merge pull request #990 from snyk/fix/retry-informer

fix: retry on network errors in K8s API informers

Author: ivanstanev

src/supervisor/kuberenetes-api-wrappers.ts

@@ -3,18 +3,13 @@ import sleep from 'sleep-promise';
 
 import { logger } from '../common/logger';
 import { IRequestError } from './types';
+import { RETRYABLE_NETWORK_ERRORS } from './watchers/types';
 
 export const ATTEMPTS_MAX = 3;
 export const DEFAULT_SLEEP_SEC = 1;
 export const MAX_SLEEP_SEC = 5;
 type IKubernetesApiFunction<ResponseType> = () => Promise<ResponseType>;
 
-const RETRYABLE_NETWORK_ERRORS: string[] = [
-  'ECONNREFUSED',
-  'ETIMEDOUT',
-  'ECONNRESET',
-];
-
 export async function retryKubernetesApiRequest<ResponseType>(
   func: IKubernetesApiFunction<ResponseType>,
 ): Promise<ResponseType> {

src/supervisor/watchers/handlers/index.ts

@@ -30,7 +30,7 @@ import {
 import { k8sApi, kubeConfig } from '../../cluster';
 import * as kubernetesApiWrappers from '../../kuberenetes-api-wrappers';
 import { IWorkloadWatchMetadata, FALSY_WORKLOAD_NAME_MARKER } from './types';
-import { ECONNRESET_ERROR_CODE } from '../types';
+import { RETRYABLE_NETWORK_ERRORS } from '../types';
 
 /**
  * This map is used in combination with the kubernetes-client Informer API
@@ -173,10 +173,11 @@ export async function setupInformer(
   namespace: string,
   workloadKind: WorkloadKind,
 ): Promise<void> {
+  const logContext: Record<string, unknown> = { namespace, workloadKind };
   const isSupported = await isSupportedWorkload(namespace, workloadKind);
   if (!isSupported) {
     logger.info(
-      { namespace, workloadKind },
+      logContext,
       'The Kubernetes cluster does not support this workload',
     );
     return;
@@ -196,7 +197,7 @@ export async function setupInformer(
       );
     } catch (err) {
       logger.error(
-        { err, namespace, workloadKind },
+        { ...logContext, err },
         'error while listing entities on namespace',
       );
       throw err;
@@ -211,18 +212,22 @@ export async function setupInformer(
 
   informer.on(ERROR, (err) => {
     // Types from client library insists that callback is of type KubernetesObject
-    if ((err as any).code === ECONNRESET_ERROR_CODE) {
+    const code = (err as any).code || '';
+    if (RETRYABLE_NETWORK_ERRORS.includes(code)) {
       logger.debug(
-        {},
-        `informer ${ECONNRESET_ERROR_CODE} occurred, restarting informer`,
+        logContext,
+        `informer ${code} occurred, restarting informer`,
       );
 
       // Restart informer after 1sec
       setTimeout(async () => {
         await informer.start();
       }, 1000);
     } else {
-      logger.error({ err }, 'unexpected informer error event occurred');
+      logger.error(
+        { ...logContext, err },
+        'unexpected informer error event occurred',
+      );
     }
   });
 
@@ -235,7 +240,7 @@ export async function setupInformer(
           (watchedWorkload.metadata && watchedWorkload.metadata.name) ||
           FALSY_WORKLOAD_NAME_MARKER;
         logger.warn(
-          { error, namespace, name, workloadKind },
+          { ...logContext, error, name },
           'could not execute the informer handler for a workload',
         );
       }

src/supervisor/watchers/index.ts

@@ -4,7 +4,6 @@ import { V1Namespace } from '@kubernetes/client-node';
 import { logger } from '../../common/logger';
 import { config } from '../../common/config';
 import { WorkloadKind } from '../types';
-import { ECONNRESET_ERROR_CODE } from './types';
 import { setupInformer } from './handlers';
 import { kubeConfig, k8sApi } from '../cluster';
 import * as kubernetesApiWrappers from '../kuberenetes-api-wrappers';
@@ -13,6 +12,7 @@ import {
   openshiftInternalNamespaces,
 } from './internal-namespaces';
 import { state } from '../../state';
+import { RETRYABLE_NETWORK_ERRORS } from './types';
 
 async function setupWatchesForNamespace(namespace: V1Namespace): Promise<void> {
   const namespaceName = extractNamespaceName(namespace);
@@ -73,10 +73,9 @@ async function setupWatchesForCluster(): Promise<void> {
 
   informer.on(ERROR, (err) => {
     // Types from client library insists that callback is of type V1Namespace
-    if ((err as any).code === ECONNRESET_ERROR_CODE) {
-      logger.debug(
-        `namespace informer ${ECONNRESET_ERROR_CODE} occurred, restarting informer`,
-      );
+    const code = (err as any).code || '';
+    if (RETRYABLE_NETWORK_ERRORS.includes(code)) {
+      logger.debug(`namespace informer ${code} occurred, restarting informer`);
 
       // Restart informer after 1sec
       setTimeout(async () => {

src/supervisor/watchers/types.ts

@@ -11,4 +11,8 @@ export enum PodPhase {
   Unknown = 'Unknown',
 }
 
-export const ECONNRESET_ERROR_CODE = 'ECONNRESET';
+export const RETRYABLE_NETWORK_ERRORS: readonly string[] = [
+  'ECONNREFUSED',
+  'ETIMEDOUT',
+  'ECONNRESET',
+];