Merge pull request #653 from snyk/feat/restart-informers-if-error-occurred

feat: restart k8s informers on error event (e.g idle connection)

Author: dkontorovskyy

src/supervisor/watchers/handlers/index.ts

@@ -1,4 +1,4 @@
-import { makeInformer, ADD, DELETE, UPDATE, KubernetesObject } from '@kubernetes/client-node';
+import { makeInformer, ADD, DELETE, ERROR, UPDATE, KubernetesObject } from '@kubernetes/client-node';
 import { logger } from '../../../common/logger';
 import { WorkloadKind } from '../../types';
 import { podWatchHandler, podDeletedHandler } from './pod';
@@ -12,6 +12,7 @@ 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';
+import { ECONNRESET_ERROR_CODE } from '../types';
 
 /**
  * This map is used in combination with the kubernetes-client Informer API
@@ -109,6 +110,20 @@ export function setupInformer(namespace: string, workloadKind: WorkloadKind) {
 
   const informer = makeInformer<KubernetesObject>(kubeConfig, namespacedEndpoint, loggedListMethod);
 
+  informer.on(ERROR, (err) => {
+    // Types from client library insists that callback is of type KubernetesObject
+    if ((err as any).code === ECONNRESET_ERROR_CODE) {
+      logger.debug(`informer ${ECONNRESET_ERROR_CODE} occurred, restarting informer`);
+
+      // Restart informer after 1sec
+      setTimeout(() => {
+        informer.start();
+      }, 1000);
+    } else {
+      logger.error({ err }, 'unexpected informer error event occurred');
+    }
+  });
+
   for (const informerVerb of Object.keys(workloadMetadata.handlers)) {
     informer.on(informerVerb, async (watchedWorkload) => {
       try {

src/supervisor/watchers/index.ts

@@ -1,9 +1,10 @@
-import { makeInformer, ADD } from '@kubernetes/client-node';
+import { makeInformer, ADD, ERROR } from '@kubernetes/client-node';
 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';
@@ -67,6 +68,20 @@ function setupWatchesForCluster(): 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`);
+
+      // Restart informer after 1sec
+      setTimeout(() => {
+        informer.start();
+      }, 1000);
+    } else {
+      logger.error({ err }, 'unexpected namespace informer error event occurred');
+    }
+  });
+
   informer.on(ADD, (namespace: V1Namespace) => {
     try {
       const namespaceName = extractNamespaceName(namespace);

src/supervisor/watchers/types.ts

@@ -10,3 +10,5 @@ export enum PodPhase {
   // For some reason the state of the pod could not be obtained.
   Unknown = 'Unknown',
 }
+
+export const ECONNRESET_ERROR_CODE = 'ECONNRESET';