Saas-support-tasklogger-streams-2 (#35)

support parallel steps

Author: noamt-codefresh

lib/ContainerLogger.js

@@ -5,7 +5,7 @@ const Q              = require('q');
 const logger         = require('cf-logs').Logger('codefresh:containerLogger');
 const CFError        = require('cf-errors');
 const LoggerStrategy = require('./enums').LoggerStrategy;
-const { Transform, pipeline } = require('stream');
+const { Transform } = require('stream');
 
 class ContainerLogger extends EventEmitter {
 
@@ -99,32 +99,25 @@ class ContainerLogger extends EventEmitter {
     }
 
     _streamTty(stdout, stderr) {
+        logger.info(`Piping stdout and stderr step streams`);
 
-        pipeline(stdout, this._logSizeLimitStream(), this.stepLogger.writeStream(), (err) => {
-            if (err) {
-                logger.error(`Stdout streams pipeline failed on: ${err}`);
-                return;
-            }
-            logger.info('Stdout streams pipeline succeeded.');
-        });
+        const stepLoggerWritableStream = this.stepLogger.writeStream();
+        stepLoggerWritableStream.on('error', err => logger.error(`stepLoggerWritableStream: ${err}`));
+
+        // Attention(!) all streams piped to step logger writable stream must be a new streams(!) in order to avoid message piping twice to writable stream.
+        stdout.pipe(this._logSizeLimitStream(), {end: false}).pipe(this.stepLogger.stepNameTransformStream(), {end: false}).pipe(stepLoggerWritableStream, {end: false});
 
         if (!stderr) {
             return;
         }
 
-        pipeline(stderr, this._logSizeLimitStream(), this._errorTransformerStream(), this.stepLogger.writeStream(), (err) => {
-            if (err) {
-                logger.error(`Stderr streams pipeline failed on: ${err}`);
-                return;
-            }
-            logger.info('Stderr streams pipeline succeeded.');
-        });
+        stderr.pipe(this._logSizeLimitStream(), {end: false})
+            .pipe(this._errorTransformerStream(), {end: false}).pipe(this.stepLogger.stepNameTransformStream(), {end: false}).pipe(stepLoggerWritableStream, {end: false});
 
         stderr.once('end', () => {
             this.stepFinished = true;
             logger.info(`stderr end event was fired for container: ${this.containerId}`);
         });
-
     }
 
     _registerToTtyStreams(stdout, stderr) {

package.json

@@ -6,7 +6,7 @@
     "cf-container-logger"
   ],
   "dependencies": {
-    "@codefresh-io/task-logger": "1.5.3",
+    "@codefresh-io/task-logger": "1.5.6",
     "cf-errors": "^0.1.11",
     "cf-logs": "^1.1.0",
     "docker-events": "0.0.2",

service.yaml

@@ -1 +1 @@
-version: 1.0.37
+version: 1.1.0