Repository: godaddy/kubernetes-external-secrets
Pull request #107 (2019-06-30 16:10:21)

refactor: use watch api and instant poll new or modified secrets
Currently you can't set a lower events interval than poll interval, as this would result in never polling anything as the poller would be recreated everytime the events are polled.

Not sure if this is the best way to go about it but I tested refactoring so the events poller tracks which external secrets it has seen and yields appropriate events instead of always doing a delete all.

It would be nice if we could in some way instantly poll if an external secret has been modified or added. As if you have a long poll time you would currently have to wait for that to trigger, which feels slow when adding a new external secret. Not sure how to deal with pod restarts tho, as you wouldn't necessarily want to poll instantly when that happens, one alternative would be for the poller to check if the secret exists when started, and if it doesn't then trigger a poll right away.

Let me know what you think :) 

ping @silasbw

silasbw: @Flydiverny an alternative to polling is using the Kubernetes API watch endpoints. I think we want to use those eventually, but hadn't gotten around to the implementation. If we're going to make improvements to this code, it would be good for those improvements to work with the watch implementation (or maybe we do that one first?).

@satish-ravi, thoughts?
Flydiverny: yeah I was wondering why it didn't use watch, but still tried to simulate it somehow? :D So I went with yielding event types based on what watch should provide, not sure about the payload tho. I'm not sure how to use watch with the kubernetes-client though 

would you do something like?
```js
    const stream = kubeClient
          .apis[customResourceManifest.spec.group]
          .v1[customResourceManifest.spec.names.plural]
          .get({ watch: true })
    const jsonStream = new JSONStream()
    stream.pipe(jsonStream)
    jsonStream.on('data', object => {
      // do stuff
    })
```
Flydiverny: @silasbw made some updates, now using the watch api instead.
jeffpearce: Hi @Flydiverny. I'm taking a look at this today
jeffpearce: @satish-ravi 
Flydiverny: Yeah, I'll update the test to work with the stream.
I'm not sure about how to deal with error handling tho, what happens when it loses connection, how to get it to recover?
Think this makes it harder as well https://github.com/godaddy/kubernetes-client/issues/248?  kubernetes-client is also deprecating `getStream` but there's no `getObjectStream` for the CRDs.
silasbw: Yes, we should add `getObjectStream`.
Flydiverny: Please check again if there is something more to be done, or that you want reverted :) 

Patch:
diff --git a/README.md b/README.md
index c6eb0c7..09c2875 100644
--- a/README.md
+++ b/README.md
@@ -50,23 +50,25 @@ helm delete kubernetes-external-secrets
 
 The following table lists the configurable parameters of the `kubernetes-external-secrets` chart and their default values.
 
-| Parameter                            | Description                                                  | Default                                                 |
-| ------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------- |
-| `env.AWS_REGION`                     | Set AWS_REGION in Deployment Pod                             | `us-west-2`                                             |
-| `env.EVENTS_INTERVAL_MILLISECONDS`   | Set EVENTS_INTERVAL_MILLISECONDS in Deployment Pod           | `60000`                                                 |
-| `env.POLLER_INTERVAL_MILLISECONDS`   | Set POLLER_INTERVAL_MILLISECONDS in Deployment Pod           | `10000`                                                 |
-| `image.repository`                   | kubernetes-external-secrets Image name                       | `godaddy/kubernetes-external-secrets`                   |
-| `image.tag`                          | kubernetes-external-secrets Image tag                        | `1.2.0`                                                 |
-| `image.pullPolicy`                   | Image pull policy                                            | `IfNotPresent`                                          |
-| `rbac.create`                        | Create & use RBAC resources                                  | `true`                                                  |
-| `serviceAccount.create`              | Whether a new service account name should be created.        | `true`                                                  |
-| `serviceAccount.name`                | Service account to be used.                                  | automatically generated
-| `podAnnotations`                     | Annotations to be added to pods                              | `{}`                                                    |
-| `replicaCount`                       | Number of replicas                                           | `1`                                                     |
-| `nodeSelector`                       | node labels for pod assignment                               | `{}`                                                    |
-| `tolerations`                        | List of node taints to tolerate (requires Kubernetes >= 1.6) | `[]`                                                    |
-| `affinity`                           | Affinity for pod assignment                                  | `{}`                                                    |
-| `resources`                          | Pod resource requests & limits                               | `{}`                                                    |
+| Parameter                                 | Description                                                  | Default                                                 |
+| ----------------------------------------- | ------------------------------------------------------------ | ------------------------------------------------------- |
+| `env.AWS_REGION`                          | Set AWS_REGION in Deployment Pod                             | `us-west-2`                                             |
+| `env.EVENTS_INTERVAL_MILLISECONDS`        | Set EVENTS_INTERVAL_MILLISECONDS in Deployment Pod           | `60000`                                                 |
+| `env.POLLER_INTERVAL_MILLISECONDS`        | Set POLLER_INTERVAL_MILLISECONDS in Deployment Pod           | `10000`                                                 |
+| `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  |                                                         |
+| `image.repository`                        | kubernetes-external-secrets Image name                       | `godaddy/kubernetes-external-secrets`                   |
+| `image.tag`                               | kubernetes-external-secrets Image tag                        | `1.2.0`                                                 |
+| `image.pullPolicy`                        | Image pull policy                                            | `IfNotPresent`                                          |
+| `rbac.create`                             | Create & use RBAC resources                                  | `true`                                                  |
+| `serviceAccount.create`                   | Whether a new service account name should be created.        | `true`                                                  |
+| `serviceAccount.name`                     | Service account to be used.                                  | automatically generated                                 |
+| `podAnnotations`                          | Annotations to be added to pods                              | `{}`                                                    |
+| `replicaCount`                            | Number of replicas                                           | `1`                                                     |
+| `nodeSelector`                            | node labels for pod assignment                               | `{}`                                                    |
+| `tolerations`                             | List of node taints to tolerate (requires Kubernetes >= 1.6) | `[]`                                                    |
+| `affinity`                                | Affinity for pod assignment                                  | `{}`                                                    |
+| `resources`                               | Pod resource requests & limits                               | `{}`                                                    |
 
 Specify each parameter using the `--set key=value[,key=value]` argument to `helm install`. For example,
 
@@ -77,6 +79,12 @@ helm install --name kubernetes-external-secrets \
 charts/kubernetes-external-secrets
 ```
 
+### Use IAM credentials for Secrets Manager access
+
+If not running on EKS you will have to use an IAM user (in lieu of a role).
+Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY env vars in the session/pod.
+You can use envVarsFromSecret in the helm chart to create these env vars from existing k8s secrets
+
 ### Add a secret
 
 Add your secret data to your backend. For example, AWS Secrets Manager:
diff --git a/bin/daemon.js b/bin/daemon.js
index f055fe0..486a916 100755
--- a/bin/daemon.js
+++ b/bin/daemon.js
@@ -15,7 +15,6 @@ const {
   kubeClient,
   customResourceManager,
   customResourceManifest,
-  eventsIntervalMilliseconds,
   logger,
   pollerIntervalMilliseconds
 } = require('../config')
@@ -31,7 +30,6 @@ async function main () {
   const externalSecretEvents = getExternalSecretEvents({
     kubeClient,
     customResourceManifest,
-    intervalMilliseconds: eventsIntervalMilliseconds,
     logger
   })
 
diff --git a/charts/kubernetes-external-secrets/README.md b/charts/kubernetes-external-secrets/README.md
new file mode 100644
index 0000000..c54cd39
--- /dev/null
+++ b/charts/kubernetes-external-secrets/README.md
@@ -0,0 +1,123 @@
+# 💂 Kubernetes External Secrets
+
+[Kubernetes External Secrets](https://github.com/godaddy/kubernetes-external-secrets) allows you to use external secret management systems (*e.g.*, [AWS Secrets Manager](https://aws.amazon.com/secrets-manager/)) to securely add secrets in Kubernetes. Read more about the design and motivation for Kubernetes External Secrets on the [GoDaddy Engineering Blog](https://godaddy.github.io/2019/04/16/kubernetes-external-secrets/).
+
+## TL;DR;
+
+```bash
+$ helm install stable/kubernetes-external-secrets
+```
+
+## Prerequisites
+
+* Kubernetes 1.7+
+
+## Installing the Chart
+
+To install the chart with the release named `my-release`:
+
+```bash
+$ helm install --name my-release stable/kubernetes-external-secrets
+```
+
+> **Tip:** A namespace can be specified by the `Helm` option '`--namespace kube-external-secrets`'
+
+## Uninstalling the Chart
+
+To uninstall/delete the deployment:
+
+```bash
+helm delete my-release
+```
+
+## Configuration
+
+The following table lists the configurable parameters of the `kubernetes-external-secrets` chart and their default values.
+
+| Parameter                            | Description                                                  | Default                                                 |
+| ------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------- |
+| `env.AWS_REGION`                     | Set AWS_REGION in Deployment Pod                             | `us-west-2`                                             |
+| `env.EVENTS_INTERVAL_MILLISECONDS`   | Set EVENTS_INTERVAL_MILLISECONDS in Deployment Pod           | `60000`                                                 |
+| `env.POLLER_INTERVAL_MILLISECONDS`   | Set POLLER_INTERVAL_MILLISECONDS in Deployment Pod           | `10000`                                                 |
+| `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  |                                                         |
+| `image.repository`                   | kubernetes-external-secrets Image name                       | `godaddy/kubernetes-external-secrets`                   |
+| `image.tag`                          | kubernetes-external-secrets Image tag                        | `1.2.0`                                                 |
+| `image.pullPolicy`                   | Image pull policy                                            | `IfNotPresent`                                          |
+| `nameOverride`                   | Override the name of app                                            | `nil`                                          |
+| `fullnameOverride`                   | Override the full name of app                                            | `nil`                                          |
+| `rbac.create`                        | Create & use RBAC resources                                  | `true`                                                  |
+| `serviceAccount.create`              | Whether a new service account name should be created.        | `true`                                                  |
+| `serviceAccount.name`                | Service account to be used.                                  | automatically generated
+| `podAnnotations`                     | Annotations to be added to pods                              | `{}`                                                    |
+| `replicaCount`                       | Number of replicas                                           | `1`                                                     |
+| `nodeSelector`                       | node labels for pod assignment                               | `{}`                                                    |
+| `tolerations`                        | List of node taints to tolerate (requires Kubernetes >= 1.6) | `[]`                                                    |
+| `affinity`                           | Affinity for pod assignment                                  | `{}`                                                    |
+| `resources`                          | Pod resource requests & limits                               | `{}`                                                    |
+
+Specify each parameter using the `--set key=value[,key=value]` argument to `helm install`. For example,
+
+```bash
+helm install stable/kubernetes-external-secrets --name my-releases \
+--set env.POLLER_INTERVAL_MILLISECONDS='300000' \
+--set podAnnotations."iam\.amazonaws\.com/role"='Name-Of-IAM-Role-With-SecretManager-Access'
+```
+
+Alternatively, a YAML file that specifies the values for the parameters can be provided while installing the chart. For example,
+
+```bash
+helm install stable/kubernetes-external-secrets --name my-release -f values.yaml
+```
+
+> **Tip**: You can use the default [values.yaml](values.yaml)
+
+## Add a secret
+
+Add your secret data to your backend. For example, AWS Secrets Manager:
+
+```
+aws secretsmanager create-secret --name hello-service/password --secret-string "1234"
+```
+
+and then create a `hello-service-external-secret.yml` file:
+
+```yml
+apiVersion: 'kubernetes-client.io/v1'
+kind: ExternalSecret
+metadata:
+  name: hello-service
+secretDescriptor:
+  backendType: secretsManager
+  data:
+    - key: hello-service/password
+      name: password
+```
+
+Save the file and run:
+
+```sh
+kubectl apply -f hello-service-external-secret.yml
+```
+
+Wait a few minutes and verify that the associated `Secret` has been created:
+
+```sh
+kubectl get secret hello-service -o=yaml
+```
+
+The `Secret` created by the controller should look like:
+
+```yml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: hello-service
+type: Opaque
+data:
+  password: MTIzNA==
+```
+
+## Further Information
+
+For more in-depth documentation of usage, please see the [Kubernetes External Secrets repo](https://github.com/godaddy/kubernetes-external-secrets)
diff --git a/charts/kubernetes-external-secrets/templates/NOTES.txt b/charts/kubernetes-external-secrets/templates/NOTES.txt
new file mode 100644
index 0000000..965ef29
--- /dev/null
+++ b/charts/kubernetes-external-secrets/templates/NOTES.txt
@@ -0,0 +1,4 @@
+The kubernetes external secrets has been installed. Check its status by running:
+  kubectl --namespace {{ .Release.Namespace }} get pods -l "app.kubernetes.io/name={{ include "kubernetes-external-secrets.name" . }},app.kubernetes.io/instance={{ .Release.Name }}"
+
+Visit https://github.com/godaddy/kubernetes-external-secrets for instructions on how to use kubernetes external secrets
diff --git a/charts/kubernetes-external-secrets/templates/deployment.yaml b/charts/kubernetes-external-secrets/templates/deployment.yaml
index f079e34..c5c56c8 100644
--- a/charts/kubernetes-external-secrets/templates/deployment.yaml
+++ b/charts/kubernetes-external-secrets/templates/deployment.yaml
@@ -33,10 +33,18 @@ spec:
           env:
           {{- range $name, $value := .Values.env }}
           {{- if not (empty $value) }}
-            - name: {{ $name | quote }}
-              value: {{ $value | quote }}
+          - name: {{ $name | quote }}
+            value: {{ $value | quote }}
           {{- end }}
           {{- end }}
+          # Params for env vars populated from k8s secrets
+          {{- range $key, $value := .Values.envVarsFromSecret }}
+          - name: {{ $key }}
+            valueFrom:
+              secretKeyRef:
+                name: {{ $value.secretKeyRef | quote }}
+                key: {{ $value.key | quote }}
+          {{- end }}
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}
diff --git a/charts/kubernetes-external-secrets/values.yaml b/charts/kubernetes-external-secrets/values.yaml
index 565ac8a..de2a6d2 100644
--- a/charts/kubernetes-external-secrets/values.yaml
+++ b/charts/kubernetes-external-secrets/values.yaml
@@ -8,6 +8,15 @@ env:
   EVENTS_INTERVAL_MILLISECONDS: 60000
   POLLER_INTERVAL_MILLISECONDS: 10000
 
+# Create environment variables from exists k8s secrets
+#envVarsFromSecret:
+#  AWS_ACCESS_KEY_ID:
+#    secretKeyRef: aws-credentials
+#    key: id
+#  AWS_SECRET_ACCESS_KEY:
+#    secretKeyRef: aws-credentials
+#    key: key
+
 rbac:
   # Specifies whether RBAC resources should be created
   create: true
diff --git a/config/environment.js b/config/environment.js
index a031d16..4f672d5 100644
--- a/config/environment.js
+++ b/config/environment.js
@@ -16,13 +16,13 @@ if (environment === 'development') {
   require('dotenv').config()
 }
 
-const eventsIntervalMilliseconds = process.env.EVENTS_INTERVAL_MILLISECONDS
-  ? Number(process.env.EVENTS_INTERVAL_MILLISECONDS) : 60000
 const pollerIntervalMilliseconds = process.env.POLLER_INTERVAL_MILLISECONDS
   ? Number(process.env.POLLER_INTERVAL_MILLISECONDS) : 10000
 
+const logLevel = process.env.LOG_LEVEL || 'info'
+
 module.exports = {
   environment,
-  eventsIntervalMilliseconds,
-  pollerIntervalMilliseconds
+  pollerIntervalMilliseconds,
+  logLevel
 }
diff --git a/config/index.js b/config/index.js
index d1cce19..1679695 100644
--- a/config/index.js
+++ b/config/index.js
@@ -20,7 +20,8 @@ const kubeClient = new kube.Client({ backend: kubeBackend })
 const logger = pino({
   serializers: {
     err: pino.stdSerializers.err
-  }
+  },
+  level: envConfig.logLevel
 })
 
 const customResourceManager = new CustomResourceManager({
diff --git a/lib/daemon.js b/lib/daemon.js
index cab8a92..dfad6f5 100644
--- a/lib/daemon.js
+++ b/lib/daemon.js
@@ -36,19 +36,18 @@ class Daemon {
    * @returns {Object} Poller descriptor.
    */
   _createPollerDescriptor (externalSecret) {
-    const { name, namespace, resourceVersion } = externalSecret.metadata
+    const { uid, name, namespace } = externalSecret.metadata
     // NOTE(jdaeli): hash this in case resource version becomes too long?
-    const id = `${name}_${resourceVersion}`
     const secretDescriptor = { ...externalSecret.secretDescriptor, name }
     const ownerReference = {
       apiVersion: externalSecret.apiVersion,
       controller: true,
       kind: externalSecret.kind,
-      name: externalSecret.metadata.name,
-      uid: externalSecret.metadata.uid
+      name,
+      uid
     }
 
-    return { id, namespace, secretDescriptor, ownerReference }
+    return { id: uid, namespace, secretDescriptor, ownerReference }
   }
 
   /**
@@ -56,42 +55,66 @@ class Daemon {
    * @param {String} pollerId - ID of the poller to remove.
    */
   _removePoller (pollerId) {
-    this._logger.info(`stopping and removing poller ${pollerId}`)
-    this._pollers[pollerId].stop()
-    delete this._pollers[pollerId]
+    if (this._pollers[pollerId]) {
+      this._logger.info(`stopping and removing poller ${pollerId}`)
+      this._pollers[pollerId].stop()
+      delete this._pollers[pollerId]
+    }
   }
 
   _removePollers () {
     Object.keys(this._pollers).forEach(pollerId => this._removePoller(pollerId))
   }
 
+  _addPoller (descriptor, forcePoll = false) {
+    this._logger.info('spinning up poller', descriptor)
+
+    const poller = new Poller({
+      backends: this._backends,
+      intervalMilliseconds: this._pollerIntervalMilliseconds,
+      kubeClient: this._kubeClient,
+      logger: this._logger,
+      namespace: descriptor.namespace,
+      secretDescriptor: descriptor.secretDescriptor,
+      ownerReference: descriptor.ownerReference
+    })
+
+    this._pollers[descriptor.id] = poller.start({ forcePoll })
+  }
+
   /**
    * Start daemon and create pollers.
    */
   async start () {
     for await (const event of this._externalSecretEvents) {
-      if (event.type === 'DELETED_ALL') {
-        this._removePollers()
-      } else if (event.type === 'ADDED') {
-        const descriptor = this._createPollerDescriptor(event.object)
-        this._logger.info('spinning up poller', descriptor)
-
-        const poller = new Poller({
-          backends: this._backends,
-          intervalMilliseconds: this._pollerIntervalMilliseconds,
-          kubeClient: this._kubeClient,
-          logger: this._logger,
-          namespace: descriptor.namespace,
-          secretDescriptor: descriptor.secretDescriptor,
-          ownerReference: descriptor.ownerReference
-        })
-
-        // handle duplicate ADDED events
-        if (this._pollers[descriptor.id]) {
+      const descriptor = event.object ? this._createPollerDescriptor(event.object) : null
+
+      switch (event.type) {
+        case 'DELETED': {
+          this._removePoller(descriptor.id)
+          break
+        }
+
+        case 'ADDED': {
+          this._addPoller(descriptor)
+          break
+        }
+
+        case 'MODIFIED': {
           this._removePoller(descriptor.id)
+          this._addPoller(descriptor, true)
+          break
         }
 
-        this._pollers[descriptor.id] = poller.start()
+        case 'DELETED_ALL': {
+          this._removePollers()
+          break
+        }
+
+        default: {
+          this._logger.warn(event, 'Unhandled event type %s', event.type)
+          break
+        }
       }
     }
   }
diff --git a/lib/external-secret.js b/lib/external-secret.js
index 1037518..6cf8b58 100644
--- a/lib/external-secret.js
+++ b/lib/external-secret.js
@@ -1,53 +1,94 @@
 'use strict'
 
+const JSONStream = require('json-stream')
+
 /**
- * Block asynchronous flow.
- * @param {number} milliseconds - Number of milliseconds to block flow operation.
- * @returns {Promise} Promise object representing block flow operation.
+ * Creates an FIFO queue which you can put to and take from.
+ * If theres nothing to take it will wait with resolving until
+ * something is put to the queue.
+ * @returns {Object} Queue instance with put and take methods
  */
-function sleep ({ milliseconds }) {
-  return new Promise(resolve => setTimeout(resolve, milliseconds))
+function createEventQueue () {
+  const queuedEvents = []
+  const waitingResolvers = []
+
+  return {
+    take: () => queuedEvents.length > 0
+      ? Promise.resolve(queuedEvents.shift())
+      : new Promise(resolve => waitingResolvers.push(resolve)),
+    put: (msg) => waitingResolvers.length > 0
+      ? waitingResolvers.shift()(msg)
+      : queuedEvents.push(msg)
+  }
+}
+
+async function startWatcher ({
+  kubeClient,
+  customResourceManifest,
+  logger,
+  eventQueue
+}) {
+  const deathQueue = createEventQueue()
+
+  try {
+    while (true) {
+      logger.debug('Starting watch stream')
+
+      const stream = kubeClient
+        .apis[customResourceManifest.spec.group]
+        .v1.watch[customResourceManifest.spec.names.plural]
+        .getStream()
+
+      const jsonStream = new JSONStream()
+      stream.pipe(jsonStream)
+
+      jsonStream.on('data', eventQueue.put)
+
+      jsonStream.on('error', (err) => {
+        logger.warn(err, 'Got error on stream')
+        deathQueue.put('ERROR')
+      })
+
+      jsonStream.on('end', () => {
+        deathQueue.put('END')
+      })
+
+      await deathQueue.take()
+
+      logger.debug('Stopping watch stream')
+      eventQueue.put({ type: 'DELETED_ALL' })
+
+      stream.abort()
+    }
+  } catch (err) {
+    logger.error(err, 'Watcher crashed')
+  }
 }
 
 /**
  * Get a stream of external secret events. This implementation uses
- * polling/batching and converts it to a stream of events.
+ * watch and yields as a stream of events.
  * @param {Object} kubeClient - Client for interacting with kubernetes cluster.
  * @param {Object} customResourceManifest - Custom resource manifest.
- * @param {number} intervalMilliseconds - Interval time in milliseconds for polling external secrets.
- * @param {Object} logger - Logger for logging stuff.
  * @returns {Object} An async generator that yields externalsecret events.
  */
 function getExternalSecretEvents ({
   kubeClient,
   customResourceManifest,
-  intervalMilliseconds,
   logger
 }) {
   return (async function * () {
+    const eventQueue = createEventQueue()
+
+    startWatcher({
+      kubeClient,
+      customResourceManifest,
+      logger,
+      eventQueue
+    })
+
     while (true) {
-      try {
-        const externalSecrets = await kubeClient
-          .apis[customResourceManifest.spec.group]
-          .v1[customResourceManifest.spec.names.plural]
-          .get()
-
-        // this is analagous to handling a disconnect from a watch stream
-        yield {
-          type: 'DELETED_ALL'
-        }
-
-        // simulate a bunch of ADDED events
-        for (const externalSecret of externalSecrets.body.items) {
-          yield {
-            type: 'ADDED',
-            object: externalSecret
-          }
-        }
-      } catch (err) {
-        logger.warn('Failed to fetch external secrets', err)
-      }
-      await sleep({ milliseconds: intervalMilliseconds })
+      yield await eventQueue.take()
     }
   }())
 }
diff --git a/lib/poller.js b/lib/poller.js
index 54726c4..2e69ef6 100644
--- a/lib/poller.js
+++ b/lib/poller.js
@@ -44,12 +44,13 @@ class Poller {
 
   /**
    * Create Kubernetes secret manifest.
-   * @param {SecretDescriptor} secretDescriptor - Kubernetes secret description.s.
    * @returns {Object} Promise object representing Kubernetes manifest.
    */
-  async _createSecretManifest ({ secretDescriptor }) {
+  async _createSecretManifest () {
+    const secretDescriptor = this._secretDescriptor
     const data = await this._backends[secretDescriptor.backendType]
       .getSecretManifestData({ secretDescriptor })
+
     return {
       apiVersion: 'v1',
       kind: 'Secret',
@@ -71,7 +72,7 @@ class Poller {
   async _poll () {
     this._logger.info('running poll')
     try {
-      await this._upsertKubernetesSecret({ secretDescriptor: this._secretDescriptor })
+      await this._upsertKubernetesSecret()
     } catch (err) {
       this._logger.error('failure while polling the secrets')
       this._logger.error(err)
@@ -80,15 +81,15 @@ class Poller {
 
   /**
    * Create or update Kubernets secret in the cluster.
-   * @param {SecretDescriptor} secretDescriptor - Kubernetes secret description.
    * @returns {Promise} Promise object representing operation result.
    */
-  async _upsertKubernetesSecret ({ secretDescriptor }) {
+  async _upsertKubernetesSecret () {
+    const secretDescriptor = this._secretDescriptor
     const secretName = secretDescriptor.name
-    const secretManifest = await this._createSecretManifest({ secretDescriptor })
+    const secretManifest = await this._createSecretManifest()
     const kubeNamespace = this._kubeClient.api.v1.namespaces(this._namespace)
 
-    this._logger.info(`upserting secret ${secretName}`)
+    this._logger.info(`upserting secret ${secretName} in ${this._namespace}`)
     try {
       return await kubeNamespace.secrets.post({ body: secretManifest })
     } catch (err) {
@@ -97,12 +98,38 @@ class Poller {
     }
   }
 
+  /**
+   * Checks if secret exists, if not trigger a poll
+   */
+  async _checkForSecret () {
+    const secretDescriptor = this._secretDescriptor
+    const secretName = secretDescriptor.name
+    const kubeNamespace = this._kubeClient.api.v1.namespaces(this._namespace)
+
+    try {
+      await kubeNamespace.secrets(secretName).get()
+    } catch (err) {
+      if (err.statusCode === 404) {
+        this._logger.info('Secret does not exist, polling right away')
+        this._poll()
+      }
+    }
+  }
+
   /**
    * Start poller.
+   * @param {boolean} forcePoll - Trigger poll right away
    * @returns {Object} Poller instance.
    */
-  start () {
+  start ({ forcePoll = false } = {}) {
     if (this._interval) return this
+
+    if (forcePoll) {
+      this._poll()
+    } else {
+      this._checkForSecret()
+    }
+
     this._logger.info('starting poller')
     this._interval = setInterval(this._poll.bind(this), this._intervalMilliseconds)
     return this
diff --git a/package.json b/package.json
index be4fd50..d24a605 100644
--- a/package.json
+++ b/package.json
@@ -30,6 +30,7 @@
   },
   "dependencies": {
     "aws-sdk": "^2.433.0",
+    "json-stream": "^1.0.0",
     "kubernetes-client": "^8.3.0",
     "lodash.clonedeep": "^4.5.0",
     "make-promises-safe": "^5.0.0",


Test patch:
diff --git a/lib/external-secret.test.js b/lib/external-secret.test.js
index f314389..ee47ede 100644
--- a/lib/external-secret.test.js
+++ b/lib/external-secret.test.js
@@ -3,6 +3,7 @@
 
 const { expect } = require('chai')
 const sinon = require('sinon')
+const { Readable } = require('stream')
 
 const { getExternalSecretEvents } = require('./external-secret')
 
@@ -11,6 +12,7 @@ describe('getExternalSecretEvents', () => {
   let externalSecretsApiMock
   let fakeCustomResourceManifest
   let loggerMock
+  let mockedStream
 
   beforeEach(() => {
     fakeCustomResourceManifest = {
@@ -22,36 +24,78 @@ describe('getExternalSecretEvents', () => {
       }
     }
     externalSecretsApiMock = sinon.mock()
+
+    mockedStream = new Readable()
+    mockedStream._read = () => {}
+
     externalSecretsApiMock.get = sinon.stub()
     kubeClientMock = sinon.mock()
     kubeClientMock.apis = sinon.mock()
     kubeClientMock.apis['kubernetes-client.io'] = sinon.mock()
     kubeClientMock.apis['kubernetes-client.io'].v1 = sinon.mock()
+    kubeClientMock.apis['kubernetes-client.io'].v1.watch = sinon.mock()
+    kubeClientMock.apis['kubernetes-client.io']
+      .v1.watch.externalsecrets = sinon.mock()
     kubeClientMock.apis['kubernetes-client.io']
-      .v1.externalsecrets = externalSecretsApiMock
+      .v1.watch.externalsecrets.getStream = () => mockedStream
+
     loggerMock = sinon.mock()
+    loggerMock.info = sinon.stub()
+    loggerMock.warn = sinon.stub()
+    loggerMock.error = sinon.stub()
+    loggerMock.debug = sinon.stub()
   })
 
   it('gets a stream of external secret events', async () => {
-    const fakeExternalSecretObject = { type: 'ExternalSecret' }
-    externalSecretsApiMock.get.resolves({
-      body: {
-        items: [fakeExternalSecretObject]
-      }
-    })
+    const fakeExternalSecretObject = {
+      apiVersion: 'kubernetes-client.io/v1',
+      kind: 'ExternalSecret',
+      metadata: {
+        name: 'my-secret',
+        namespace: 'default'
+      },
+      secretDescriptor: { backendType: 'secretsManager', data: [] }
+    }
 
     const events = getExternalSecretEvents({
       kubeClient: kubeClientMock,
       customResourceManifest: fakeCustomResourceManifest,
-      intervalMilliseconds: 0,
       logger: loggerMock
     })
 
-    const deletedAllEvent = await events.next()
-    expect(deletedAllEvent.value.type).is.equal('DELETED_ALL')
+    mockedStream.push(`${JSON.stringify({
+      type: 'MODIFIED',
+      object: fakeExternalSecretObject
+    })}\n`)
+
+    mockedStream.push(`${JSON.stringify({
+      type: 'ADDED',
+      object: fakeExternalSecretObject
+    })}\n`)
+
+    mockedStream.push(`${JSON.stringify({
+      type: 'DELETED',
+      object: fakeExternalSecretObject
+    })}\n`)
+
+    mockedStream.push(`${JSON.stringify({
+      type: 'DELETED_ALL'
+    })}\n`)
+
+    const modifiedEvent = await events.next()
+    expect(modifiedEvent.value.type).is.equal('MODIFIED')
+    expect(modifiedEvent.value.object).is.deep.equal(fakeExternalSecretObject)
+
     const addedEvent = await events.next()
     expect(addedEvent.value.type).is.equal('ADDED')
     expect(addedEvent.value.object).is.deep.equal(fakeExternalSecretObject)
-    events.return(null)
+
+    const deletedEvent = await events.next()
+    expect(deletedEvent.value.type).is.equal('DELETED')
+    expect(deletedEvent.value.object).is.deep.equal(fakeExternalSecretObject)
+
+    const deletedAllEvent = await events.next()
+    expect(deletedAllEvent.value.type).is.equal('DELETED_ALL')
+    expect(deletedAllEvent.value.object).is.deep.equal(undefined)
   })
 })
diff --git a/lib/poller.test.js b/lib/poller.test.js
index 683508f..18f5579 100644
--- a/lib/poller.test.js
+++ b/lib/poller.test.js
@@ -10,7 +10,7 @@ describe('Poller', () => {
   let backendMock
   let kubeClientMock
   let loggerMock
-  let poller
+  let pollerFactory
 
   const ownerReference = {
     apiVersion: 'owner-api/v1',
@@ -28,7 +28,15 @@ describe('Poller', () => {
     loggerMock.info = sinon.stub()
     loggerMock.error = sinon.stub()
 
-    poller = new Poller({
+    pollerFactory = (secretDescriptor = {
+      backendType: 'fakeBackendType',
+      name: 'fakeSecretName',
+      properties: [
+        'fakePropertyName1',
+        'fakePropertyName2'
+      ]
+    }) => new Poller({
+      secretDescriptor,
       backends: {
         fakeBackendType: backendMock
       },
@@ -50,21 +58,21 @@ describe('Poller', () => {
     })
 
     it('creates secret manifest', async () => {
+      const poller = pollerFactory({
+        backendType: 'fakeBackendType',
+        name: 'fakeSecretName',
+        properties: [
+          'fakePropertyName1',
+          'fakePropertyName2'
+        ]
+      })
+
       backendMock.getSecretManifestData.resolves({
         fakePropertyName1: 'ZmFrZVByb3BlcnR5VmFsdWUx', // base 64 value
         fakePropertyName2: 'ZmFrZVByb3BlcnR5VmFsdWUy' // base 64 value
       })
 
-      const secretManifest = await poller._createSecretManifest({
-        secretDescriptor: {
-          backendType: 'fakeBackendType',
-          name: 'fakeSecretName',
-          properties: [
-            'fakePropertyName1',
-            'fakePropertyName2'
-          ]
-        }
-      })
+      const secretManifest = await poller._createSecretManifest()
 
       expect(backendMock.getSecretManifestData.calledWith({
         secretDescriptor: {
@@ -94,12 +102,13 @@ describe('Poller', () => {
   })
 
   describe('_poll', () => {
+    let poller
     beforeEach(() => {
-      poller._secretDescriptor = {
+      poller = pollerFactory({
         backendType: 'fakeBackendType',
         name: 'fakeSecretName1',
         properties: ['fakePropertyName1', 'fakePropertyName2']
-      }
+      })
       poller._upsertKubernetesSecret = sinon.stub()
     })
 
@@ -109,16 +118,7 @@ describe('Poller', () => {
       await poller._poll()
 
       expect(loggerMock.info.calledWith('running poll')).to.equal(true)
-      expect(poller._upsertKubernetesSecret.calledWith({
-        secretDescriptor: {
-          backendType: 'fakeBackendType',
-          name: 'fakeSecretName1',
-          properties: [
-            'fakePropertyName1',
-            'fakePropertyName2'
-          ]
-        }
-      })).to.equal(true)
+      expect(poller._upsertKubernetesSecret.calledWith()).to.equal(true)
     })
 
     it('logs error if storing secret operation fails', async () => {
@@ -132,9 +132,14 @@ describe('Poller', () => {
 
   describe('_upsertKubernetesSecret', () => {
     let kubeNamespaceMock
-    let upsertSecretParams
+    let poller
 
     beforeEach(() => {
+      poller = pollerFactory({
+        backendType: 'fakeBackendType',
+        name: 'fakeSecretName',
+        properties: ['fakePropertyName']
+      })
       kubeNamespaceMock = sinon.mock()
       kubeNamespaceMock.secrets = sinon.stub().returns(kubeNamespaceMock)
       kubeClientMock.api = sinon.mock()
@@ -151,19 +156,12 @@ describe('Poller', () => {
           fakePropertyName: 'ZmFrZVByb3BlcnR5VmFsdWU='
         }
       })
-      upsertSecretParams = {
-        secretDescriptor: {
-          backendType: 'fakeBackendType',
-          name: 'fakeSecretName',
-          properties: ['fakePropertyName']
-        }
-      }
     })
 
     it('creates new secret', async () => {
       kubeNamespaceMock.secrets.post = sinon.stub().resolves()
 
-      await poller._upsertKubernetesSecret(upsertSecretParams)
+      await poller._upsertKubernetesSecret()
 
       expect(kubeNamespaceMock.secrets.post.calledWith({
         body: {
@@ -186,9 +184,10 @@ describe('Poller', () => {
       kubeNamespaceMock.secrets.post = sinon.stub().throws(conflictError)
       kubeNamespaceMock.put = sinon.stub().resolves()
 
-      await poller._upsertKubernetesSecret(upsertSecretParams)
+      await poller._upsertKubernetesSecret()
 
       expect(kubeNamespaceMock.secrets.calledWith('fakeSecretName')).to.equal(true)
+
       expect(kubeNamespaceMock.put.calledWith({
         body: {
           apiVersion: 'v1',
@@ -213,7 +212,7 @@ describe('Poller', () => {
       let error
 
       try {
-        await poller._upsertKubernetesSecret(upsertSecretParams)
+        await poller._upsertKubernetesSecret()
       } catch (err) {
         error = err
       }
@@ -225,8 +224,10 @@ describe('Poller', () => {
 
   describe('start', () => {
     let clock
+    let poller
 
     beforeEach(() => {
+      poller = pollerFactory()
       clock = sinon.useFakeTimers()
       poller._poll = sinon.stub()
     })
@@ -238,7 +239,7 @@ describe('Poller', () => {
 
     it('starts poller', async () => {
       expect(poller._interval).to.equal(null)
-      poller.start()
+      poller.start({ forcePoll: true })
       clock.tick(poller._intervalMilliseconds)
       expect(loggerMock.info.calledWith('starting poller')).to.equal(true)
       expect(poller._interval).to.not.equal(null)
@@ -248,8 +249,10 @@ describe('Poller', () => {
 
   describe('stop', () => {
     let clock
+    let poller
 
     beforeEach(() => {
+      poller = pollerFactory()
       clock = sinon.useFakeTimers()
       poller._poll = sinon.stub()
     })
@@ -259,7 +262,7 @@ describe('Poller', () => {
     })
 
     it('stops poller', async () => {
-      poller.start()
+      poller.start({ forcePoll: true })
       poller.stop()
       expect(loggerMock.info.calledWith('stopping poller')).to.equal(true)
       expect(poller._interval).to.equal(null)