fix: support streams without pipes

Author: vasil-cf

lib/ContainerLogger.js

@@ -78,10 +78,8 @@ class ContainerLogger extends EventEmitter {
 
                             this._registerToTtyStreams(stdout, stderr);
                         } else {
-                            stdout = stdout.pipe(new DeprecatedImagesInterceptorStream());
                             this._handleNonTtyStream(stdout, false);
                             if (stderr) {
-                                stderr = stderr.pipe(new DeprecatedImagesInterceptorStream());
                                 this._handleNonTtyStream(stderr, true);
                             }
                         }
@@ -175,11 +173,9 @@ class ContainerLogger extends EventEmitter {
     }
 
     _registerToTtyStreams(stdout, stderr) {
-        stdout = stdout.pipe(new DeprecatedImagesInterceptorStream());
         this._handleTtyStream(stdout, false);
 
         if (stderr) {
-            stderr = stderr.pipe(new DeprecatedImagesInterceptorStream());
             stderr.once('end', () => {
                 this.stepFinished = true;
                 logger.info(`stderr end event was fired for container: ${this.containerId}`);
@@ -190,27 +186,38 @@ class ContainerLogger extends EventEmitter {
 
     _handleTtyStream(stream, isError) {
         this.handledStreams++;
-        stream.on('end', this._handleFinished.bind(this));
+        const deprecatedImagesInterceptor = new DeprecatedImagesInterceptorStream(true);
+        stream.on('end', () => {
+            this._handleFinished();
+            deprecatedImagesInterceptor.end();
+        });
         stream.on('data', (chunk) => {
+            deprecatedImagesInterceptor.write(chunk);
             this._logMessage(Buffer.from(chunk).toString('utf-8'), isError);
         });
         logger.info(`Listening on stream 'data' event for container: ${this.containerId}`);
     }
 
     _handleNonTtyStream(stream, isError) {
         this.handledStreams++;
+        const deprecatedImagesInterceptor = new DeprecatedImagesInterceptorStream(true);
         stream.on('readable', () => {
             let header = stream.read(8);
             while (header !== null) {
+                deprecatedImagesInterceptor.write(header);
                 const payload = stream.read(header.readUInt32BE(4));
                 if (payload === null) {
                     break;
                 }
+                deprecatedImagesInterceptor.write(payload);
                 this._logMessage(Buffer.from(payload).toString('utf8'), isError);
                 header = stream.read(8);
             }
         });
-        stream.on('end', this._handleFinished.bind(this));
+        stream.on('end', () => {
+            this._handleFinished();
+            deprecatedImagesInterceptor.end();
+        });
         logger.info(`Listening on stream 'readable' event for container: ${this.containerId}`);
     }
 

lib/metric/deprecated-images/deprecated-images-interceptor.stream.ts

@@ -4,31 +4,34 @@ import { Transform, TransformCallback } from 'stream';
 import deprecatedImagesCollector from './deprecated-images.collector';
 
 export class DeprecatedImagesInterceptorStream extends Transform {
-    private _lastChunk: Buffer;
+    private lastChunk: Buffer;
 
-    constructor() {
+    constructor(private readonly noPush = false) {
         super();
-        this._lastChunk = Buffer.alloc(0);
+        this.lastChunk = Buffer.alloc(0);
     }
 
     _transform(chunk: Buffer, encoding: BufferEncoding, callback: TransformCallback): void {
         try {
 
             const text = Buffer
-                .concat([this._lastChunk, chunk])
+                .concat([this.lastChunk, chunk])
                 .toString('utf8');
 
             const lines = text.split('\n');
 
             // the final element in 'lines' may be an incomplete line
-            // save it in _lastChunk for next time
-            this._lastChunk = Buffer.from(lines.pop() ?? '', 'utf8');
+            // save it in lastChunk for next time
+            this.lastChunk = Buffer.from(lines.pop() ?? '', 'utf8');
 
             for (const line of lines) {
                 deprecatedImagesCollector.catchDeprecatedImage(line.trim());
             }
 
-            this.push(chunk);
+            if (!this.noPush) {
+                this.push(chunk);
+            }
+
             callback();
         } catch (error) {
             callback(error as any);
@@ -42,7 +45,7 @@ export class DeprecatedImagesInterceptorStream extends Transform {
      */
     _flush(callback: TransformCallback): void {
         try {
-            const finalLine = this._lastChunk.toString('utf8');
+            const finalLine = this.lastChunk.toString('utf8');
             deprecatedImagesCollector.catchDeprecatedImage(finalLine.trim());
             callback();
         } catch (error) {