container logger will wait for all logs to be sent (#41)

* container logger will wait for all logs to be sent
* CONTAINER_LOGGER_SHOW_PROGRESS - if true will print progression info
* added more info to health-check

Author: roi-codefresh

gulpfile.js

@@ -19,7 +19,7 @@ gulp.task('no.onlys', function (callback) {
 
 
 gulp.task('lint', ['clean'], function () {
-	return gulp.src(['**/*.js', '!**/node_modules/**', '!**/server/migration/**', '!coverage/**/*.js', '!test/logger.unit.spec.js'])
+	return gulp.src(['**/*.js', '!**/node_modules/**', '!**/server/migration/**', '!coverage/**/*.js', '!test/*.unit.spec.js'])
 		.pipe(jshint({lookup: true}))
 		.pipe(jshint.reporter('default'))
 		.pipe(jshint.reporter('fail'));

lib/ContainerLogger.js

@@ -27,6 +27,8 @@ class ContainerLogger extends EventEmitter {
         this.logSize                   = 0;
         this.isWorkflowLogSizeExceeded = isWorkflowLogSizeExceeded;
         this.stepFinished              = false;
+        this.finishedStreams = 0;
+        this.handledStreams = 0;
     }
 
     start() {
@@ -105,14 +107,24 @@ class ContainerLogger extends EventEmitter {
         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});
-
+        // { end = false } on the stepLoggerWritableStream because there is only one instance of it for all the steps.
+        this.handledStreams++;
+        stdout
+            .pipe(this._logSizeLimitStream())
+            .pipe(this.stepLogger.stepNameTransformStream().once('end', this._handleFinished.bind(this)))
+            .pipe(stepLoggerWritableStream, {end: false});
+
+        
         if (!stderr) {
             return;
         }
 
-        stderr.pipe(this._logSizeLimitStream(), {end: false})
-            .pipe(this._errorTransformerStream(), {end: false}).pipe(this.stepLogger.stepNameTransformStream(), {end: false}).pipe(stepLoggerWritableStream, {end: false});
+        this.handledStreams++;
+        stderr
+            .pipe(this._logSizeLimitStream())
+            .pipe(this._errorTransformerStream())
+            .pipe(this.stepLogger.stepNameTransformStream().once('end', this._handleFinished.bind(this)))
+            .pipe(stepLoggerWritableStream, {end: false});
 
         stderr.once('end', () => {
             this.stepFinished = true;
@@ -133,6 +145,8 @@ class ContainerLogger extends EventEmitter {
     }
 
     _handleTtyStream(stream, isError) {
+        this.handledStreams++;
+        stream.on('end', this._handleFinished.bind(this));
         stream.on('data', (chunk) => {
             const buf     = new Buffer(chunk);
             const message = buf.toString('utf8');
@@ -142,6 +156,7 @@ class ContainerLogger extends EventEmitter {
     }
 
     _handleNonTtyStream(stream, isError) {
+        this.handledStreams++;
         stream.on('readable', () => {
             let header = stream.read(8);
             while (header !== null) {
@@ -153,6 +168,7 @@ class ContainerLogger extends EventEmitter {
                 header = stream.read(8);
             }
         });
+        stream.on('end', this._handleFinished.bind(this));
         logger.info(`Listening on stream 'readable' event for container: ${this.containerId}`);
     }
 
@@ -218,6 +234,14 @@ class ContainerLogger extends EventEmitter {
         });
     }
 
+    _handleFinished() {
+        this.finishedStreams++;
+
+        if (this.finishedStreams === this.handledStreams) {
+            // the emission of this event reflects the ending of all streams handled by this container logger
+            this.emit('end');
+        }
+    }
 }
 
 module.exports = ContainerLogger;

lib/index.js

@@ -2,6 +2,7 @@
 
 const path   = require('path');
 const cflogs = require('cf-logs');
+const Q = require('q');
 
 const loggerOptions = {
     filePath: path.join(__dirname, '../logs', 'logs.log'),
@@ -15,17 +16,25 @@ const loggerOptions = {
 cflogs.init(loggerOptions);
 
 const Logger = require('./logger');
+const buildFinishedPromise = Q.defer();
 
 const logger = new Logger({
     loggerId: process.env.LOGGER_ID,
     taskLoggerConfig: JSON.parse(process.env.TASK_LOGGER_CONFIG),
     findExistingContainers: process.env.LISTEN_ON_EXISTING,
     logSizeLimit: process.env.LOG_SIZE_LIMIT ? (parseInt(process.env.LOG_SIZE_LIMIT) * 1000000) : undefined,
+    buildFinishedPromise: buildFinishedPromise.promise,
+    showProgress: process.env.SHOW_PROGRESS === 'true' ? true : false,
 });
 
 logger.validate();
 logger.start();
 
+process.on('SIGUSR2', () => {
+    console.log('GOT SIGUSR2 --- engine signaling build is finished');
+    buildFinishedPromise.resolve();
+});
+
 process.on('beforeExit', (code) => {
     console.log(`beforeExit: ${code}`);
     logger.state.beforeExitCode = code;

lib/logger.js

@@ -1,7 +1,9 @@
 'use strict';
 
 const fs                      = require('fs');
+const { EventEmitter }        = require('events');
 const _                       = require('lodash');
+const Q                       = require('q');
 const Docker                  = require('dockerode');
 const DockerEvents            = require('docker-events');
 const CFError                 = require('cf-errors');
@@ -12,15 +14,16 @@ const ContainerHandlingStatus = require('./enums').ContainerHandlingStatus;
 const ContainerLogger         = require('./ContainerLogger');
 const { TaskLogger }          = require('@codefresh-io/task-logger');
 
-
-const initialState = { status: 'init', lastLogsDate: Date.now() , failedHealthChecks: [] , restartCounter: 0, containers: {} };
+const initialState = { pid: process.pid, status: 'init', lastLogsDate: new Date() , failedHealthChecks: [] , restartCounter: 0, containers: {} };
 class Logger {
 
     constructor({
         loggerId,
         taskLoggerConfig,
         findExistingContainers,
-        logSizeLimit
+        logSizeLimit,
+        buildFinishedPromise,
+        showProgress,
     }) {
         this.taskLoggerConfig       = taskLoggerConfig;
         this.loggerId               = loggerId;
@@ -29,6 +32,10 @@ class Logger {
         this.containerLoggers       = [];
         this.logSize                = 0;
         this.taskLogger             = undefined;
+        this.buildFinishedPromise = buildFinishedPromise || Q.resolve();
+        this.finishedContainers = 0;
+        this.finishedContainersEmitter = new EventEmitter();
+        this.showProgress = showProgress;
 
         let dockerSockPath;
         if (fs.existsSync('/var/run/codefresh/docker.sock')) {
@@ -43,6 +50,8 @@ class Logger {
             socketPath: dockerSockPath,
         });
         this._readState();
+        this._handleBuildFinished();
+        this._updateStateInterval = setInterval(this._updateStateFile.bind(this), 1000);
     }
 
     /**
@@ -70,6 +79,7 @@ class Logger {
 
         TaskLogger(this.taskLoggerConfig.task, this.taskLoggerConfig.opts)
             .then((taskLogger) => {
+                this.taskLogger = taskLogger;
                 taskLogger.on('error', (err) => {
                     logger.error(err.stack);
                 });
@@ -81,12 +91,16 @@ class Logger {
                     }else {
                         this.state.healthCheckStatus = status; 
                     }
+                    
                     this._writeNewState();
                 });
-
-                this.taskLogger = taskLogger;
+                taskLogger.on('flush', () => {
+                    this._updateMissingLogs();
+                    this._updateLastLoggingDate();
+                });
+                this.state.logsStatus = this.taskLogger.getStatus();
                 logger.info(`taskLogger successfully created`);
-
+                
                 this._listenForNewContainers();
 
                 this.state.status = 'ready';
@@ -108,8 +122,9 @@ class Logger {
     _readState() {
         const filePath = `${__dirname}/state.json`;
         if (fs.existsSync(filePath)) {
-            this.state = _.omit(JSON.parse(fs.readFileSync(filePath, 'utf8'), 'containers'));
+            this.state = _.omit(JSON.parse(fs.readFileSync(filePath, 'utf8'), ['containers', 'pid']));
             this.state.containers = {};
+            this.state.pid = process.pid;
             let restartCounter =  _.get(this.state, 'restartCounter', 0);
              restartCounter++;
              this.state.restartCounter = restartCounter;
@@ -143,10 +158,9 @@ class Logger {
             } else if (!disableLog) {
                 logger.info(`State: ${currentState} updated and written to file: ${filePath}`);
             }
-        });
+        }); 
     }
 
-
     logLimitExceeded() {
         // TODO in the future when we allow a workflow to use multuple dinds, this will not be correct
         // we need to get the total size of logs from all dinds
@@ -238,7 +252,7 @@ class Logger {
         });
         this.containerLoggers.push(containerLogger);
         containerLogger.on('message.logged', this._updateTotalLogSize.bind(this));
-        containerLogger.on('message.logged', this._updateLastLoggingDate.bind(this));
+        containerLogger.once('end', this._handleContainerStreamEnd.bind(this));
 
         containerLogger.start()
             .done(() => {
@@ -254,14 +268,33 @@ class Logger {
             });
     }
 
+    _updateMissingLogs() {
+        const resolvedCalls = _.get(this, 'state.logsStatus.resolvedCalls', 0);
+        const writeCalls = _.get(this, 'state.logsStatus.writeCalls', 0);
+        const rejectedCalls = _.get(this, 'state.logsStatus.rejectedCalls', 0);
+
+        _.set(this, 'state.logsStatus.missingLogs', writeCalls - resolvedCalls - rejectedCalls);
+    }
+
     _updateTotalLogSize() {
         this.logSize = this._getTotalLogSize();
         this.taskLogger.setLogSize(this.logSize);
     }
 
     _updateLastLoggingDate() {
-        this.state.lastLogsDate = Date.now();
-        this._writeNewState(true);
+        this.state.lastLogsDate = new Date();
+    }
+
+    _updateStateFile() {
+        if (this.state.status === 'done') {
+            clearInterval(this._updateStateInterval);
+        } else {
+            this._writeNewState(true);
+
+            if (this.showProgress) {
+                logger.debug(`logger progress update: ${JSON.stringify(this.state.logsStatus)}`);
+            }
+        }
     }
 
     /**
@@ -304,6 +337,42 @@ class Logger {
         });
     }
 
+    _handleContainerStreamEnd() {
+        this.finishedContainers++;
+        this.finishedContainersEmitter.emit('end');
+    }
+
+    // do not call before build is finished
+    _awaitAllStreamsClosed() {
+        const deferred = Q.defer();
+        this._checkAllStreamsClosed(deferred);
+        this.finishedContainersEmitter.on('end', this._checkAllStreamsClosed.bind(this, deferred));
+        return deferred.promise;
+    }
+
+    _checkAllStreamsClosed(deferred) {
+        if (this.finishedContainers === this.containerLoggers.length) {
+            deferred.resolve();
+        }
+    }
+
+    _handleBuildFinished() {
+        this.buildFinishedPromise
+            .then(() => {
+                logger.info('=== build is finished ===');
+                return this._awaitAllStreamsClosed();
+            })
+            .then(() => {
+                logger.info('=== all streams have been closed ===');
+                return this.taskLogger.awaitLogsFlushed();
+            })
+            .then(() => {
+                logger.info('=== All logs flushed. Container logger finished. ===');
+                this.state.logsStatus = this.taskLogger.getStatus();
+                this.state.status = 'done';
+                this._writeNewState();
+            });
+    }
 }
 
 module.exports = Logger;