Support reporting to firebase metrics on logs and stop writing new logs when reaching the limit (#11)

Author: itai-codefresh

lib/ContainerLogger.js

@@ -1,19 +1,35 @@
 'use strict';
 
+const EventEmitter   = require('events');
 const Q              = require('q');
 const logger         = require('cf-logs').Logger('codefresh:containerLogger');
 const CFError        = require('cf-errors');
 const LoggerStrategy = require('./enums').LoggerStrategy;
 
-class ContainerLogger {
+class ContainerLogger extends EventEmitter {
 
-    constructor(containerId, containerInterface, firebaseLogger, firebaseLastUpdate, loggerStrategy) {
-        this.containerId        = containerId;
-        this.containerInterface = containerInterface;
-        this.firebaseLogger     = firebaseLogger;
-        this.firebaseLastUpdate = firebaseLastUpdate;
-        this.loggerStrategy     = loggerStrategy;
-        this.tty                = false;
+    constructor({
+        containerId,
+        containerInterface,
+        firebaseLogger,
+        firebaseLastUpdate,
+        firebaseMetricsLogs,
+        logSizeLimit,
+        isWorkflowLogSizeExceeded, // eslint-disable-line
+        loggerStrategy
+    }) {
+        super();
+        this.containerId               = containerId;
+        this.containerInterface        = containerInterface;
+        this.firebaseLogger            = firebaseLogger;
+        this.firebaseLastUpdate        = firebaseLastUpdate;
+        this.firebaseMetricsLogs       = firebaseMetricsLogs;
+        this.loggerStrategy            = loggerStrategy;
+        this.tty                       = false;
+        this.logSizeLimit              = logSizeLimit;
+        this.logSize                   = 0;
+        this.isWorkflowLogSizeExceeded = isWorkflowLogSizeExceeded;
+        this.stepFinished              = false;
     }
 
     start() {
@@ -34,18 +50,24 @@ class ContainerLogger {
                 // See documentation of the docker api here: https://docs.docker.com/engine/reference/api/docker_remote_api_v1.24/#/attach-to-a-container
                 if (this.tty) {
                     this._handleTtyStream(stdout, false);
-                    this._handleTtyStream(stderr, true);
+                    if (stderr) {
+                        this._handleTtyStream(stderr, true);
+                    }
                 } else {
                     this._handleNonTtyStream(stdout, false);
                 }
 
                 stdout.on('end', () => {
+                    this.stepFinished = true;
                     logger.info(`stdout end event was fired for container: ${this.containerId}`);
                 });
 
-                stderr.on('end', () => {
-                    logger.info(`stderr end event was fired for container: ${this.containerId}`);
-                });
+                if (stderr) {
+                    stderr.on('end', () => {
+                        this.stepFinished = true;
+                        logger.info(`stderr end event was fired for container: ${this.containerId}`);
+                    });
+                }
             }, (err) => {
                 return Q.reject(new CFError({
                     cause: err,
@@ -105,13 +127,32 @@ class ContainerLogger {
         logger.info(`Listening on stream 'readable' event for container: ${this.containerId}`);
     }
 
+    _stepLogSizeExceeded() {
+        return this.logSize > this.logSizeLimit;
+    }
+
     _logMessageToFirebase(message, isError) {
+        if (this.logSizeLimit && (this._stepLogSizeExceeded() || this.isWorkflowLogSizeExceeded()) && !isError) {
+            if (!this.logExceededLimitsNotified) {
+                this.logExceededLimitsNotified = true;
+                message = `\x1B[01;93mLog size exceeded for ${this._stepLogSizeExceeded() ? 'this step' : 'the workflow'}.\nThe step will continue to execute until it finished but new logs will not be stored.\x1B[0m\r\n`;
+            } else {
+                return;
+            }
+        }
+
         if (isError) {
             message = `\x1B[31m${message}\x1B[0m`;
         }
-        
+
         this.firebaseLogger.push(message);
         this.firebaseLastUpdate.set(new Date().getTime());
+
+        if (this.logSizeLimit) {
+            this.logSize += Buffer.byteLength(message);
+            this.firebaseMetricsLogs.child('total').set(this.logSize);
+        }
+        this.emit('message.logged');
     }
 
 }

lib/index.js

@@ -11,8 +11,14 @@ cflogs.init(loggerOptions);
 
 const Logger = require('./logger');
 
-
-const logger = new Logger(process.env.LOGGER_ID, process.env.FIREBASE_AUTH_URL, process.env.FIREBASE_SECRET, process.env.LISTEN_ON_EXISTING);
+const logger = new Logger({
+    loggerId: process.env.LOGGER_ID,
+    firebaseAuthUrl: process.env.FIREBASE_AUTH_URL,
+    firebaseSecret: process.env.FIREBASE_SECRET,
+    findExistingContainers: process.env.LISTEN_ON_EXISTING,
+    firebaseMetricsLogsUrl: process.env.FIREBASE_METRICS_LOGS_URL,
+    logSizeLimit: process.env.LOG_SIZE_LIMIT ? (parseInt(process.env.LOG_SIZE_LIMIT) * 1000000) : undefined,
+});
 
 logger.validate();
 logger.start();

lib/logger.js

@@ -14,12 +14,23 @@ const ContainerLogger         = require('./ContainerLogger');
 
 class Logger {
 
-    constructor(loggerId, firebaseAuthUrl, firebaseSecret, findExistingContainers) {
+    constructor({
+        loggerId,
+        firebaseAuthUrl,
+        firebaseSecret,
+        findExistingContainers,
+        firebaseMetricsLogsUrl,
+        logSizeLimit
+    }) {
         this.state                  = { status: 'init' };
         this.firebaseAuthUrl        = firebaseAuthUrl;
         this.firebaseSecret         = firebaseSecret;
         this.loggerId               = loggerId;
         this.findExistingContainers = findExistingContainers === 'true';
+        this.firebaseMetricsLogsUrl = firebaseMetricsLogsUrl;
+        this.logSizeLimit           = logSizeLimit;
+        this.containerLoggers       = [];
+        this.logSize                = 0;
 
         let dockerSockPath;
         if (fs.existsSync('/var/run/codefresh/docker.sock')) {
@@ -75,6 +86,7 @@ class Logger {
                 return;
             }
             logger.info(`Authenticated to firebase url: ${this.firebaseAuthUrl}`);
+            this.firebaseMetricsLogs = new Firebase(this.firebaseMetricsLogsUrl);
 
             this._listenForNewContainers();
 
@@ -118,6 +130,19 @@ class Logger {
         });
     }
 
+
+    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
+        return this.logSizeLimit && this._getTotalLogSize() > this.logSizeLimit;
+    }
+
+    _getTotalLogSize() {
+        return _.reduce(this.containerLoggers, (sum, containerLogger) => {
+            return sum + containerLogger.logSize;
+        }, 0);
+    }
+
     /**
      * receives a container and decides if to start listening on it
      * @param loggerId
@@ -134,6 +159,12 @@ class Logger {
         const receivedFirebaseLastUpdateUrl = _.get(container,
             'Labels',
             _.get(container, 'Actor.Attributes'))['io.codefresh.logger.firebase.lastUpdateUrl'];
+        const receivedFirebaseMetricsLogsUrl = _.get(container,
+            'Labels',
+            _.get(container, 'Actor.Attributes'))['io.codefresh.logger.firebase.metricsLogs'];
+        const receivedLogSizeLimit = _.get(container,
+            'Labels',
+            _.get(container, 'Actor.Attributes'))['io.codefresh.logger.logSizeLimit'];
         const loggerStrategy                = _.get(container, 'Labels', _.get(container, 'Actor.Attributes'))['io.codefresh.logger.strategy'];
 
         if (!containerId) {
@@ -167,6 +198,11 @@ class Logger {
             return;
         }
 
+        if (!receivedFirebaseMetricsLogsUrl) {
+            logger.error(`Container: ${containerId} does contain a loggerFirebaseMetricsLogsUrl label`);
+            return;
+        }
+
         if (!loggerStrategy) {
             logger.error(`Container: ${containerId} does contain a loggerStrategy label`);
             return;
@@ -210,8 +246,33 @@ class Logger {
             return;
         }
 
+        let firebaseMetricsLogs;
+        try {
+            firebaseMetricsLogs = new Firebase(receivedFirebaseMetricsLogsUrl);
+        } catch (err) {
+            const error = new CFError({
+                cause: err,
+                message: `Failed to create a new firebase metricsLogs ref`
+            });
+            logger.error(error.toString());
+            return;
+        }
+
+        const logSizeLimit = receivedLogSizeLimit ? (parseInt(receivedLogSizeLimit) * 1000000) : undefined;
+
         const containerInterface = this.docker.getContainer(containerId);
-        const containerLogger    = new ContainerLogger(containerId, containerInterface, firebaseLogger, firebaseLastUpdate, loggerStrategy);
+        const containerLogger    = new ContainerLogger({
+            containerId,
+            containerInterface,
+            firebaseLogger,
+            firebaseLastUpdate,
+            firebaseMetricsLogs,
+            logSizeLimit,
+            isWorkflowLogSizeExceeded: this.logLimitExceeded.bind(this),
+            loggerStrategy
+        });
+        this.containerLoggers.push(containerLogger);
+        containerLogger.on('message.logged', this._updateTotalLogSize.bind(this));
 
         containerLogger.start()
             .done(() => {
@@ -227,6 +288,11 @@ class Logger {
             });
     }
 
+    _updateTotalLogSize() {
+        this.logSize = this._getTotalLogSize();
+        this.firebaseMetricsLogs.child('total').set(this.logSize);
+    }
+
     /**
      * Will check if a container was already handled (no matter what the handling status is)
      * @param containerId

package.json

@@ -8,8 +8,8 @@
   "dependencies": {
     "cf-errors": "^0.1.11",
     "cf-logs": "git+https://github.com/codefresh-io/cf-logs.git#ceba4f309e52a077747a0c6bf9c3ad02e762dc4b",
-    "dockerode": "^2.3.0",
     "docker-events": "0.0.2",
+    "dockerode": "^2.3.0",
     "firebase": "git+https://github.com/codefresh-io/firebase.git#80b2ed883ff281cd67b53bd0f6a0bbd6f330fed5",
     "forever": "^0.15.3",
     "lodash": "^4.15.0",
@@ -22,6 +22,7 @@
     "gulp-env": "^0.2.0",
     "gulp-istanbul": "^0.10.4",
     "gulp-jshint": "^1.11.0",
+    "gulp-mocha": "^6.0.0",
     "gulp-mocha-co": "^0.4.1-co.3",
     "gulp-rimraf": "^0.1.1",
     "isparta": "^4.0.0",