fix: convert chunks to text lines before parsing them

Author: vasil-cf

lib/ContainerLogger.js

@@ -3,13 +3,13 @@ const Q = require('q');
 const promiseRetry = require('promise-retry');
 const logger = require('cf-logs').Logger('codefresh:containerLogger');
 const CFError = require('cf-errors');
-const { Transform, PassThrough } = require('stream');
+const { Transform } = require('stream');
 
 const _ = require('lodash');
 const { LoggerStrategy } = require('./enums');
 
 // eslint-disable-next-line import/no-unresolved
-const deprecatedImagesCollector = require('./metric/deprecated-images/deprecated-images.collector').default;
+const { DeprecatedImagesInterceptorStream } = require('./metric/deprecated-images/deprecated-images-interceptor.stream');
 
 const CONTAINER_START_RETRY_TIMEOUT_SECONDS = 1;
 const CONTAINER_START_RETRY_LIMIT = 10;
@@ -78,9 +78,10 @@ class ContainerLogger extends EventEmitter {
 
                             this._registerToTtyStreams(stdout, stderr);
                         } else {
-                            // TODO: parse deprecated images here
+                            stdout = stdout.pipe(new DeprecatedImagesInterceptorStream());
                             this._handleNonTtyStream(stdout, false);
                             if (stderr) {
+                                stderr = stderr.pipe(new DeprecatedImagesInterceptorStream());
                                 this._handleNonTtyStream(stderr, true);
                             }
                         }
@@ -136,7 +137,7 @@ class ContainerLogger extends EventEmitter {
         // { end = false } on the stepLoggerWritableStream because there is only one instance of it for all the steps.
         this.handledStreams++;
         let stdoutStream = stdout
-            .pipe(this._interceptDeprecatedImagesStream())
+            .pipe(new DeprecatedImagesInterceptorStream())
             .pipe(this._logSizeLimitStream())
             .pipe(this.stepLogger.createMaskingStream());
 
@@ -154,7 +155,7 @@ class ContainerLogger extends EventEmitter {
 
         this.handledStreams++;
         let stderrStream = stderr
-            .pipe(this._interceptDeprecatedImagesStream())
+            .pipe(new DeprecatedImagesInterceptorStream())
             .pipe(this._logSizeLimitStream())
             .pipe(this._errorTransformerStream())
             .pipe(this.stepLogger.createMaskingStream());
@@ -174,9 +175,11 @@ 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}`);
@@ -189,9 +192,7 @@ class ContainerLogger extends EventEmitter {
         this.handledStreams++;
         stream.on('end', this._handleFinished.bind(this));
         stream.on('data', (chunk) => {
-            const message = Buffer.from(chunk).toString('utf-8');
-            deprecatedImagesCollector.catchDeprecatedImage(message);
-            this._logMessage(message, isError);
+            this._logMessage(Buffer.from(chunk).toString('utf-8'), isError);
         });
         logger.info(`Listening on stream 'data' event for container: ${this.containerId}`);
     }
@@ -242,11 +243,6 @@ class ContainerLogger extends EventEmitter {
         this.emit('message.logged', curLogSize);
     }
 
-    _interceptDeprecatedImagesStream() {
-        return new PassThrough()
-            .on('data', (chunk) => deprecatedImagesCollector.catchDeprecatedImage(chunk.toString('utf8')));
-    }
-
     _errorTransformerStream() {
         return new Transform({
             transform: (data, encoding, done) => {

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

@@ -0,0 +1,52 @@
+import { Transform, TransformCallback } from 'stream';
+
+// eslint-disable-next-line import/no-unresolved
+import deprecatedImagesCollector from './deprecated-images.collector';
+
+export class DeprecatedImagesInterceptorStream extends Transform {
+    private _lastChunk: Buffer;
+
+    constructor() {
+        super();
+        this._lastChunk = Buffer.alloc(0);
+    }
+
+    _transform(chunk: Buffer, encoding: BufferEncoding, callback: TransformCallback): void {
+        try {
+
+            const text = Buffer
+                .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');
+
+            for (const line of lines) {
+                deprecatedImagesCollector.catchDeprecatedImage(line.trim());
+            }
+
+            this.push(chunk);
+            callback();
+        } catch (error) {
+            callback(error as any);
+        }
+    }
+
+    /**
+     * _flush() is called when there is no more incoming data.
+     * If we still have leftover data that didn't end with a newline,
+     * treat it as a final line to be processed.
+     */
+    _flush(callback: TransformCallback): void {
+        try {
+            const finalLine = this._lastChunk.toString('utf8');
+            deprecatedImagesCollector.catchDeprecatedImage(finalLine.trim());
+            callback();
+        } catch (error) {
+            callback(error as any);
+        }
+    }
+}