feat: restart watcher if no events seen for specified period (default 60 sec) (#532)

Author: Flydiverny

bin/daemon.js

@@ -25,7 +25,8 @@ const {
   pollingDisabled,
   rolePermittedAnnotation,
   namingPermittedAnnotation,
-  enforceNamespaceAnnotation
+  enforceNamespaceAnnotation,
+  watchTimeout
 } = require('../config')
 
 async function main () {
@@ -37,7 +38,8 @@ async function main () {
   const externalSecretEvents = getExternalSecretEvents({
     kubeClient,
     customResourceManifest,
-    logger
+    logger,
+    watchTimeout
   })
 
   const registry = Prometheus.register

charts/kubernetes-external-secrets/README.md

@@ -83,6 +83,7 @@ The following table lists the configurable parameters of the `kubernetes-externa
 | `env.POLLER_INTERVAL_MILLISECONDS`        | Set POLLER_INTERVAL_MILLISECONDS in Deployment Pod                                                                                | `10000`                               |
 | `env.VAULT_ADDR`                          | Endpoint for the Vault backend, if using Vault                                                                                    | `http://127.0.0.1:8200`                |
 | `env.DISABLE_POLLING`                     | Disables backend polling and only updates secrets when ExternalSecret is modified, setting this to any value will disable polling | `nil`                                 |
+| `env.WATCH_TIMEOUT`                     | Restarts the external secrets resource watcher if no events have been seen in this time period (miliseconds) | `60000`                                 |
 | `envVarsFromSecret.AWS_ACCESS_KEY_ID`     | Set AWS_ACCESS_KEY_ID (from a secret) in Deployment Pod                                                                           |                                       |
 | `envVarsFromSecret.AWS_SECRET_ACCESS_KEY` | Set AWS_SECRET_ACCESS_KEY (from a secret) in Deployment Pod                                                                       |                                       |
 | `envVarsFromSecret.AZURE_TENANT_ID`       | Set AZURE_TENANT_ID (from a secret) in Deployment Pod                                                                             |                                       |

charts/kubernetes-external-secrets/values.yaml

@@ -15,6 +15,7 @@ env:
   AWS_REGION: us-west-2
   AWS_DEFAULT_REGION: us-west-2
   POLLER_INTERVAL_MILLISECONDS: 10000  # Caution, setting this frequency may incur additional charges on some platforms
+  WATCH_TIMEOUT: 60000
   LOG_LEVEL: info
   LOG_MESSAGE_KEY: 'msg'
   # Print logs level as string ("info") rather than integer (30)

config/environment.js

@@ -39,6 +39,7 @@ const enforceNamespaceAnnotation = 'ENFORCE_NAMESPACE_ANNOTATIONS' in process.en
 const metricsPort = process.env.METRICS_PORT || 3001
 
 const customResourceManagerDisabled = 'DISABLE_CUSTOM_RESOURCE_MANAGER' in process.env
+const watchTimeout = process.env.WATCH_TIMEOUT ? parseInt(process.env.WATCH_TIMEOUT) : 60000
 
 module.exports = {
   vaultEndpoint,
@@ -56,5 +57,6 @@ module.exports = {
   logLevel,
   customResourceManagerDisabled,
   useHumanReadableLogLevels,
-  logMessageKey
+  logMessageKey,
+  watchTimeout
 }

lib/external-secret.js

@@ -26,7 +26,8 @@ async function startWatcher ({
   kubeClient,
   customResourceManifest,
   logger,
-  eventQueue
+  eventQueue,
+  watchTimeout
 }) {
   const deathQueue = createEventQueue()
 
@@ -42,20 +43,39 @@ async function startWatcher ({
       const jsonStream = new JSONStream()
       stream.pipe(jsonStream)
 
-      jsonStream.on('data', eventQueue.put)
+      let timeout
+      const restartTimeout = () => {
+        if (timeout) {
+          clearTimeout(timeout)
+        }
+
+        const timeMs = watchTimeout
+        timeout = setTimeout(() => {
+          logger.info(`No watch event for ${timeMs} ms, restarting watcher`)
+          stream.abort()
+        }, timeMs)
+        timeout.unref()
+      }
+
+      jsonStream.on('data', (evt) => {
+        eventQueue.put(evt)
+        restartTimeout()
+      })
 
       jsonStream.on('error', (err) => {
         logger.warn(err, 'Got error on stream')
         deathQueue.put('ERROR')
+        clearTimeout(timeout)
       })
 
       jsonStream.on('end', () => {
         deathQueue.put('END')
+        clearTimeout(timeout)
       })
 
-      await deathQueue.take()
+      const deathEvent = await deathQueue.take()
 
-      logger.debug('Stopping watch stream')
+      logger.info('Stopping watch stream due to event: %s', deathEvent)
       eventQueue.put({ type: 'DELETED_ALL' })
 
       stream.abort()
@@ -75,7 +95,8 @@ async function startWatcher ({
 function getExternalSecretEvents ({
   kubeClient,
   customResourceManifest,
-  logger
+  logger,
+  watchTimeout
 }) {
   return (async function * () {
     const eventQueue = createEventQueue()
@@ -84,7 +105,8 @@ function getExternalSecretEvents ({
       kubeClient,
       customResourceManifest,
       logger,
-      eventQueue
+      eventQueue,
+      watchTimeout
     })
 
     while (true) {

lib/external-secret.test.js

@@ -27,6 +27,7 @@ describe('getExternalSecretEvents', () => {
 
     mockedStream = new Readable()
     mockedStream._read = () => {}
+    mockedStream.abort = () => {}
 
     externalSecretsApiMock.get = sinon.stub()
     kubeClientMock = sinon.mock()
@@ -60,7 +61,8 @@ describe('getExternalSecretEvents', () => {
     const events = getExternalSecretEvents({
       kubeClient: kubeClientMock,
       customResourceManifest: fakeCustomResourceManifest,
-      logger: loggerMock
+      logger: loggerMock,
+      watchTimeout: 5000
     })
 
     mockedStream.push(`${JSON.stringify({