Support date and origin in stored logs

taskLogger/StepLogger.js

@@ -25,6 +25,10 @@ class StepLogger extends EventEmitter {
         this.name = name;
 
         this.fatal = false;
+
+        const { origin, logWriteStrategyOptions } = opts;
+        this.origin = origin || 'unknown';
+        this.logWriteStrategyOptions = logWriteStrategyOptions || { logWriteStrategy: 'singleLogWrite' };
     }
 
     start(eventReporting) {
@@ -63,25 +67,38 @@ class StepLogger extends EventEmitter {
     }
 
     write(message) {
-        this._reportLog(message);
+        this._reportLog(this.createLogContext(message));
         this.updateLastUpdate();
     }
 
+    writeStream() {
+        return this.streamLog();
+        // TODO: this.updateLastUpdate(); ? should we pass it into stream log?
+    }
+
     debug(message) {
-        this._reportLog(`${message}\r\n`);
+        this._reportLog(this.createLogContext(`${message}\r\n`));
         this.updateLastUpdate();
     }
 
     warn(message) {
-        this._reportLog(`\x1B[01;93m${message}\x1B[0m\r\n`);
+        this._reportLog(this.createLogContext(`\x1B[01;93m${message}\x1B[0m\r\n`));
         this.updateLastUpdate();
     }
 
     info(message) {
-        this._reportLog(`${message}\r\n`);
+        this._reportLog(this.createLogContext(`${message}\r\n`));
         this.updateLastUpdate();
     }
 
+    createLogContext(message) {
+        return {
+            message,
+            origin: this.origin,
+            timestamp: new Date().toISOString()
+        };
+    }
+
     finish(err, skip, finishTime) {
         if (this.status === STATUS.PENDING && !skip) { // do not close a pending step that should not be skipped
             return;

taskLogger/TaskLogger.js

@@ -31,6 +31,10 @@ class TaskLogger extends EventEmitter {
         this.fatal    = false;
         this.finished = false;
         this.steps    = {};
+
+        this.rateLimitOptions = opts.rateLimitOptions;
+
+        this.origin = opts.origin || 'unknown';
     }
 
     create(name, resetStatus, runCreationLogic) {
@@ -75,7 +79,8 @@ class TaskLogger extends EventEmitter {
         const step = new StepClass({
             accountId: this.accountId,
             jobId: this.jobId,
-            name
+            name,
+            origin: this.origin
         }, {
             ...opts
         }, this);

taskLogger/firebase/FirebaseWritableStream.js

@@ -0,0 +1,106 @@
+const { Writable } = require('stream');
+const _ = require('lodash');
+
+const FIREBASE_MESSAGE_SIZE_LIMIT = 10 * 1024 * 1024; // 10 MB Maximum size of a string
+
+
+class FirebaseWritableStream extends Writable {
+    constructor(_firebaseClient, options) {
+        super(options);
+        this._firebaseClient = _firebaseClient;
+        this._timeUnitLimitMs = options.timeUnitLimitMs;
+        this._debounceDelay = options.debounceDelay;
+        this._messageSizeLimitPerTimeUnit = options.messageSizeLimitPerTimeUnit;
+        this._batchSize = options.batchSize;
+    }
+
+    _write(chunk, encoding, next) {
+        clearTimeout(this._debounceTimeout);
+        const newLogKey = `${this._firebaseClient.child('logs').push().key()}`;
+        const currentMessageSize = Buffer.byteLength(chunk);
+
+        this._previousTimestamp = this._previousTimestamp || new Date().getTime();
+
+        const now = new Date().getTime();
+        const msDelta = now - this._previousTimestamp;
+        console.log(`${new Date().toISOString()} streamLog.updateBatch: ms passed ${msDelta} / ${this._timeUnitLimitMs}`);
+        // minute as passed, reset current byte size
+        if (msDelta > this._timeUnitLimitMs) {
+            console.log(`${new Date().toISOString()} streamLog.updateBatch: ${this._timeUnitLimitMs} 
+                    has passed resetting current log byte size to 0`);
+            this._currentLogByteSize = 0;
+            this._previousTimestamp = now;
+        }
+
+        // current logs size during timeUnit exceeds limit (1MB/Second)
+        if (currentMessageSize + this._currentLogByteSize > this._messageSizeLimitPerTimeUnit) {
+            console.log(`${new Date().toISOString()} streamLog.updateBatch: current log size + message
+                     [${currentMessageSize + this._currentLogByteSize} / ${this._messageSizeLimitPerTimeUnit}] exceeded, flushing...`);
+
+            this._firebaseClient.update(this._logsBatch, (err) => {
+                if (err) {
+                    next(err);
+                    return;
+                }
+                this._logBatchByteSize = 0;
+                this._logsBatch = Object.create(null);
+
+                const waitMs = (this._timeUnitLimitMs - msDelta) + 5;
+                // lets wait till time unit limit will pass in order to continue
+                this._debounceTimeout = setTimeout(this._write.bind(this, chunk, encoding, next), waitMs);
+                console.log(`${new Date().toISOString()} streamLog.updateBatch: successfully flushed to firebase,
+                         waiting ${waitMs} ms for logs byte size reset`);
+            });
+            return;
+        }
+
+        this._logBatchByteSize += currentMessageSize;
+        this._currentLogByteSize += this._logBatchByteSize;
+        this._logsBatch[`${newLogKey}`] = chunk.toString();
+        console.log(`${new Date().toISOString()} streamLog.updateBatch: updated logs batch with new key
+                 '${newLogKey}', current logs byte size ${this._currentLogByteSize / 1024} KB`);
+
+        if (_.size(this._logsBatch) < this._batchSize) {
+            console.log(`${new Date().toISOString()} streamLog.updateBatch: batch capacity is still
+                     available [${_.keys(this._logsBatch).length}/${this._batchSize}],  resetting debounce flush and continue`);
+            this._setBatchFlushTimeout(this._debounceDelay);
+            next();
+            return;
+        }
+
+        console.log(`${new Date().toISOString()} streamLog.updateBatch: logs batch size has been met [${this._batchSize}] flushing...`);
+        this._firebaseClient.update(this._logsBatch, (err) => {
+            if (err) {
+                next(err);
+                return;
+            }
+            console.log(`${new Date().toISOString()} streamLog.updateBatch: flushed successfully,
+                     resetting logs batch and debounce flush`);
+            this._logsBatch = Object.create(null);
+            this._setBatchFlushTimeout(this._debounceDelay);
+            next();
+        });
+    }
+
+    _setBatchFlushTimeout(flushInterval) {
+        console.log(new Date().toISOString(), 'streamLog._setBatchFlushTimeout: setting flush timout', flushInterval);
+        this._debounceTimeout = setTimeout(() => {
+            if (_.isEmpty(this._logsBatch)) {
+                console.log(`${new Date().toISOString()} streamLog._setBatchFlushTimeout: logs batch is empty, no update is required`);
+                return;
+            }
+            console.log(`${new Date().toISOString()} streamLog._setBatchFlushTimeout: timeout 
+                        triggered, [${this._logBatchByteSize / 1024} KB /${FIREBASE_MESSAGE_SIZE_LIMIT / 1024} KB], flushing...`);
+            this._firebaseClient.update(this._logsBatch, (err) => {
+                if (err) {
+                    console.error(`${new Date().toISOString()} streamLog._setBatchFlushTimeout: flushed successfully`);
+                } else {
+                    console.log(`${new Date().toISOString()} streamLog._setBatchFlushTimeout: flushed successfully`);
+                }
+                this._logsBatch = Object.create(null);
+            });
+        }, flushInterval);
+    }
+}
+
+module.exports = FirebaseWritableStream;

taskLogger/firebase/StepLogger.js

@@ -5,8 +5,10 @@ const CFError            = require('cf-errors');
 const { STATUS }         = require('../enums');
 const BaseStepLogger     = require('../StepLogger');
 const { wrapWithRetry }  = require('../helpers');
+const FirebaseWritableStream = require('./FirebaseWritableStream');
 
 class FirebaseStepLogger extends BaseStepLogger {
+
     constructor(step, opts) {
         super(step, opts);
 
@@ -21,6 +23,12 @@ class FirebaseStepLogger extends BaseStepLogger {
 
         this.stepUrl = `${this.baseUrl}/steps/${this.name}`;
         this.stepRef = new Firebase(this.stepUrl);
+
+        this.writableStream = new FirebaseWritableStream(this.stepRef, opts.rateLimitOptions);
+
+       /* this._reportLog = this._logWriteStrategyFactory();
+        this._logsBatch = null;
+        this._logBatchFlushTimeout = null; */
     }
 
     async restore() {
@@ -95,6 +103,27 @@ class FirebaseStepLogger extends BaseStepLogger {
         this.stepRef.child('logs').set({});
     }
 
+    streamLog() {
+        return this.writableStream;
+    }
+
+    _setBatchFlushTimeout(flushInterval) {
+        this._logBatchFlushTimeout = setTimeout(() => {
+            this.stepRef.update(this._logsBatch, err => console.log(err));
+            this._logsBatch = {};
+        }, flushInterval);
+    }
+
+    _logWriteStrategyFactory(logWriteStrategy) {
+        switch (logWriteStrategy) {
+            case 'batchLogWrite':
+                this._logsBatch = {};
+                return this.batchLogWrite.bind(this);
+            case 'singleLogWrite':
+            default: return this.singleLogWrite.bind(this);
+        }
+    }
+
     async delete() {
         return this.stepRef.remove();
     }

taskLogger/firebase/TaskLogger.js

@@ -179,7 +179,8 @@ class FirebaseTaskLogger extends BaseTaskLogger {
                     const step = new StepLogger({
                         accountId: this.accountId,
                         jobId: this.jobId,
-                        name: key
+                        name: key,
+                        rateLimitOptions: this.rateLimitOptions
                     }, {
                         ...this.opts
                     });

taskLogger/test/StepLogger.unit.spec.js

@@ -161,7 +161,7 @@ describe('Base StepLogger tests', () => {
                 const stepLogger = getStepLoggerInstance();
                 const message = 'message';
                 stepLogger.write(message);
-                expect(stepLogger._reportLog).to.have.been.calledWith(message);
+                expect(stepLogger._reportLog).to.have.been.calledWithMatch({ message, origin: 'unknwon' });
                 expect(stepLogger.updateLastUpdate).to.have.been.calledWith();
             });
         });