fix: use getObjectStream to address deprecation warning in kubernetes-client (#664)

* fix: use getObjectStream to address deprecation warning in kubernetes-client

Signed-off-by: Markus Maga <[email protected]>

* fix: end stream normally and update tests

Signed-off-by: Markus Maga <[email protected]>

* fix: mock stream end

Signed-off-by: Markus Maga <[email protected]>

Author: Flydiverny

lib/external-secret.js

@@ -1,7 +1,5 @@
 'use strict'
 
-const JSONStream = require('json-stream')
-
 /**
  * Creates an FIFO queue which you can put to and take from.
  * If theres nothing to take it will wait with resolving until
@@ -37,14 +35,11 @@ async function startWatcher ({
     while (true) {
       logger.debug('Starting watch stream for namespace %s', loggedNamespaceName)
 
-      const stream = kubeClient
+      const stream = await kubeClient
         .apis[customResourceManifest.spec.group]
         .v1.watch
         .namespaces(namespace)[customResourceManifest.spec.names.plural]
-        .getStream()
-
-      const jsonStream = new JSONStream()
-      stream.pipe(jsonStream)
+        .getObjectStream()
 
       let timeout
       const restartTimeout = () => {
@@ -55,23 +50,23 @@ async function startWatcher ({
         const timeMs = watchTimeout
         timeout = setTimeout(() => {
           logger.info(`No watch event for ${timeMs} ms, restarting watcher for ${loggedNamespaceName}`)
-          stream.abort()
+          stream.end()
         }, timeMs)
         timeout.unref()
       }
 
-      jsonStream.on('data', (evt) => {
+      stream.on('data', (evt) => {
         eventQueue.put(evt)
         restartTimeout()
       })
 
-      jsonStream.on('error', (err) => {
+      stream.on('error', (err) => {
         logger.warn(err, 'Got error on stream for namespace %s', loggedNamespaceName)
         deathQueue.put('ERROR')
         clearTimeout(timeout)
       })
 
-      jsonStream.on('end', () => {
+      stream.on('end', () => {
         deathQueue.put('END')
         clearTimeout(timeout)
       })
@@ -80,8 +75,7 @@ async function startWatcher ({
 
       logger.info('Stopping watch stream for namespace %s due to event: %s', loggedNamespaceName, deathEvent)
       eventQueue.put({ type: 'DELETED_ALL' })
-
-      stream.abort()
+      stream.end()
     }
   } catch (err) {
     logger.error(err, 'Watcher for namespace %s crashed', loggedNamespaceName)

lib/external-secret.test.js

@@ -10,10 +10,9 @@ const { getExternalSecretEvents } = require('./external-secret')
 describe('getExternalSecretEvents', () => {
   let kubeClientMock
   let watchedNamespaces
-  let externalSecretsApiMock
   let fakeCustomResourceManifest
   let loggerMock
-  let mockedStream
+  let externalsecrets
 
   beforeEach(() => {
     fakeCustomResourceManifest = {
@@ -24,12 +23,10 @@ describe('getExternalSecretEvents', () => {
         }
       }
     }
-    externalSecretsApiMock = sinon.mock()
 
-    mockedStream = new Readable()
-    mockedStream._read = () => { }
-
-    externalSecretsApiMock.get = sinon.stub()
+    externalsecrets = {
+      getObjectStream: () => undefined
+    }
 
     kubeClientMock = {
       apis: {
@@ -38,9 +35,7 @@ describe('getExternalSecretEvents', () => {
             watch: {
               namespaces: () => {
                 return {
-                  externalsecrets: {
-                    getStream: () => mockedStream
-                  }
+                  externalsecrets
                 }
               }
             }
@@ -69,6 +64,27 @@ describe('getExternalSecretEvents', () => {
       spec: { backendType: 'secretsManager', data: [] }
     }
 
+    const fakeStream = Readable.from([
+      {
+        type: 'MODIFIED',
+        object: fakeExternalSecretObject
+      },
+      {
+        type: 'ADDED',
+        object: fakeExternalSecretObject
+      },
+      {
+        type: 'DELETED',
+        object: fakeExternalSecretObject
+      },
+      {
+        type: 'DELETED_ALL'
+      }
+    ])
+
+    fakeStream.end = sinon.stub()
+    externalsecrets.getObjectStream = () => fakeStream
+
     const events = getExternalSecretEvents({
       kubeClient: kubeClientMock,
       watchedNamespaces: watchedNamespaces,
@@ -77,25 +93,6 @@ describe('getExternalSecretEvents', () => {
       watchTimeout: 5000
     })
 
-    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)

package-lock.json

@@ -38,7 +38,6 @@
     "aws-sdk": "^2.628.0",
     "express": "^4.17.1",
     "js-yaml": "^3.14.1",
-    "json-stream": "^1.0.0",
     "kubernetes-client": "^9.0.0",
     "lodash": "^4.17.21",
     "make-promises-safe": "^5.1.0",