use bigger buffer (#110)

* use bigger buffer

* added buffers size

* import new task logger

* improved log size calculation

* fixing stream updates

* fix setLogSize of undefined

* added flush time limit to stream

* reduced buffer to 2MiB and reduced print info interval

* update task logger

* fixed tests

Author: roi-codefresh

Dockerfile

@@ -2,15 +2,15 @@ FROM node:16.18.1-bullseye-slim
 
 WORKDIR /root/cf-runtime
 
-RUN apt-get update && apt upgrade -y
+RUN apt-get update && apt upgrade -y && \
+    apt-get install g++ git  make python3 -y
 
 COPY package.json yarn.lock ./
 
 # install cf-runtime required binaries
-RUN apt-get install g++ git  make python3 -y && \
-    yarn install --frozen-lockfile --production && \
+RUN yarn install --frozen-lockfile --production && \
     yarn cache clean && \
-    apt-get purge g++ git make python3  -y && \
+    apt-get purge g++ git make python3 -y && \
     apt-get autoremove -y && \
     apt-get clean -y && \
     rm -rf /tmp/* && \

lib/ContainerLogger.js

@@ -4,10 +4,12 @@ const promiseRetry = require('promise-retry');
 const logger = require('cf-logs').Logger('codefresh:containerLogger');
 const CFError = require('cf-errors');
 const { Transform } = require('stream');
+const _ = require('lodash');
 const { LoggerStrategy } = require('./enums');
 
 const CONTAINER_START_RETRY_TIMEOUT_SECONDS = 1;
 const CONTAINER_START_RETRY_LIMIT = 10;
+const BUFFER_SIZE = 2 * 1024 * 1024; // 2 MiB
 
 class ContainerLogger extends EventEmitter {
 
@@ -27,6 +29,7 @@ class ContainerLogger extends EventEmitter {
         this.tty = false;
         this.logSizeLimit = logSizeLimit;
         this.logSize = 0;
+        this.bufferUsed = 0.0;
         this.isWorkflowLogSizeExceeded = isWorkflowLogSizeExceeded;
         this.stepFinished = false;
         this.finishedStreams = 0;
@@ -222,12 +225,12 @@ class ContainerLogger extends EventEmitter {
         }
 
         this.stepLogger.write(message);
-
+        const curLogSize = Buffer.byteLength(message);
         if (this.logSizeLimit) {
-            this.logSize += Buffer.byteLength(message);
+            this.logSize += curLogSize;
             this.stepLogger.setLogSize(this.logSize);
         }
-        this.emit('message.logged');
+        this.emit('message.logged', curLogSize);
     }
 
     _errorTransformerStream() {
@@ -240,13 +243,14 @@ class ContainerLogger extends EventEmitter {
     }
 
     _logSizeLimitStream() {
+        const self = this;
         return new Transform({
-            transform: (data, encoding, done) => {
-                if (this.logSizeLimit && (this._stepLogSizeExceeded() || this.isWorkflowLogSizeExceeded())) {
-                    if (!this.logExceededLimitsNotified) {
-                        this.logExceededLimitsNotified = true;
-                        const message = `\x1B[01;93mLog size exceeded for ${this._stepLogSizeExceeded()
-                            ? 'this step'
+            transform(data, encoding, done) {
+                if (self.logSizeLimit && (self._stepLogSizeExceeded() || self.isWorkflowLogSizeExceeded())) {
+                    if (!self.logExceededLimitsNotified) {
+                        self.logExceededLimitsNotified = true;
+                        const message = `\x1B[01;93mLog size exceeded for ${self._stepLogSizeExceeded()
+                            ? 'self step'
                             : 'the workflow'}.\nThe step will continue to execute until it finished but new logs will not be stored.\x1B[0m\r\n`;
                         done(null, Buffer.from(message));
                         return;
@@ -256,14 +260,17 @@ class ContainerLogger extends EventEmitter {
                     return;
                 }
 
-                if (this.logSizeLimit) {
-                    this.logSize += Buffer.byteLength(data);
-                    this.stepLogger.setLogSize(this.logSize);
+                const curLogSize = Buffer.byteLength(data);
+                if (self.logSizeLimit) {
+                    self.logSize += curLogSize;
+                    self.stepLogger.setLogSize(self.logSize);
                 }
 
-                this.emit('message.logged');
+                self.emit('message.logged', curLogSize);
+                self.bufferUsed = _.get(this, '_readableState.length') / BUFFER_SIZE;
                 done(null, data);
-            }
+            },
+            highWaterMark: BUFFER_SIZE,
         });
     }
 

lib/enums.js

@@ -10,6 +10,7 @@ const ContainerHandlingStatus = {
     INITIALIZING: 'initializing',
     LISTENING: 'listening',
     WAITING_FOR_START: 'waitingForStart',
+    FINISHED: 'finished',
 };
 
 const BuildFinishedSignalFilename = 'build_finished';

lib/isReady.js

@@ -5,7 +5,11 @@ const { ContainerHandlingStatus } = require('./enums');
 function isContainerReady(state, containerId) {
     console.log(`checking if container ${containerId} is ready`);
     const containerState = _.get(state, `containers[${containerId}]`, {});
-    const isReady = [ContainerHandlingStatus.LISTENING, ContainerHandlingStatus.WAITING_FOR_START].includes(containerState.status);
+    const isReady = [
+        ContainerHandlingStatus.LISTENING,
+        ContainerHandlingStatus.WAITING_FOR_START,
+        ContainerHandlingStatus.FINISHED,
+    ].includes(containerState.status);
     console.log(`container ${containerId} is: ${isReady ? 'READY' : 'NOT READY'}`);
     return isReady;
 }

lib/logger.js

@@ -33,7 +33,7 @@ class Logger {
         this.findExistingContainers = findExistingContainers === 'true';
         this.logSizeLimit = logSizeLimit;
         this.containerLoggers = [];
-        this.logSize = 0;
+        this.totalLogSize = 0;
         this.taskLogger = undefined;
         this.buildFinishedPromise = buildFinishedPromise || Q.resolve();
         this.finishedContainers = 0;
@@ -54,7 +54,7 @@ class Logger {
         });
         this._readState();
         this._handleBuildFinished();
-        this._updateStateInterval = setInterval(this._updateStateFile.bind(this), 1000);
+        this._updateStateInterval = setInterval(this._updateStateFile.bind(this), 3000);
     }
 
     /**
@@ -173,13 +173,7 @@ 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);
+        return this.logSizeLimit && this.totalLogSize > this.logSizeLimit;
     }
 
     /**
@@ -263,7 +257,7 @@ class Logger {
         });
         this.containerLoggers.push(containerLogger);
         containerLogger.on('message.logged', this._updateTotalLogSize.bind(this));
-        containerLogger.once('end', this._handleContainerStreamEnd.bind(this));
+        containerLogger.once('end', this._handleContainerStreamEnd.bind(this, containerId));
 
         containerLogger.start()
             .done(() => {
@@ -287,9 +281,9 @@ class Logger {
         _.set(this, 'state.logsStatus.missingLogs', writeCalls - resolvedCalls - rejectedCalls);
     }
 
-    _updateTotalLogSize() {
-        this.logSize = this._getTotalLogSize();
-        this.taskLogger.setLogSize(this.logSize);
+    _updateTotalLogSize(size) {
+        this.totalLogSize += size;
+        this.taskLogger.setLogSize(this.totalLogSize);
     }
 
     _updateLastLoggingDate() {
@@ -301,9 +295,16 @@ class Logger {
             clearInterval(this._updateStateInterval);
         } else {
             this._writeNewState(true);
-
             if (this.showProgress) {
+                const activeContainerLoggers = this.containerLoggers.filter(
+                    (containerLogger) => this.state.containers[containerLogger.containerId].status === ContainerHandlingStatus.LISTENING
+                );
+                const buffers = activeContainerLoggers.reduce((acc, cur) => {
+                    acc[cur.containerId] = cur.bufferUsed;
+                    return acc;
+                }, {});
                 logger.debug(`logger progress update: ${JSON.stringify(this.state.logsStatus)}`);
+                logger.debug(`buffers: ${JSON.stringify(buffers)}`);
             }
         }
     }
@@ -376,9 +377,10 @@ class Logger {
         });
     }
 
-    _handleContainerStreamEnd() {
+    _handleContainerStreamEnd(containerId) {
         this.finishedContainers++;
         this.finishedContainersEmitter.emit('end');
+        this.state.containers[containerId] = { status: ContainerHandlingStatus.FINISHED };
     }
 
     // do not call before build is finished

package.json

@@ -10,7 +10,7 @@
     "redis": "^3.1.1"
   },
   "dependencies": {
-    "@codefresh-io/task-logger": "^1.9.24",
+    "@codefresh-io/task-logger": "^1.10.0",
     "body-parser": "^1.19.0",
     "cf-errors": "^0.1.16",
     "cf-logs": "^1.1.25",
@@ -52,4 +52,4 @@
     "start": "node lib/index.js",
     "version": "exit 0"
   }
-}
+}

service.yaml

@@ -1 +1 @@
-version: 1.9.0
+version: 1.10.0

test/isReady.unit.spec.js

@@ -60,6 +60,16 @@ describe('isReady script', () => {
             });
             expect(process.exit).to.have.been.calledOnceWith(0);
         });
+        it('Should check exit with 0 code if container is finished status', () => {
+            const state = JSON.stringify({ status: 'ready', containers: { 'container-id': { status: ContainerHandlingStatus.FINISHED } } })
+            process.argv = ['foo', 'bar', 'container-id'];
+            proxyquire('../lib/isReady.js', {
+                'fs': {
+                    readFileSync: () => Buffer.from(state),
+                },
+            });
+            expect(process.exit).to.have.been.calledOnceWith(0);
+        });
         it('Should check exit with 1 code if container is not ready', () => {
             const state = JSON.stringify({ status: 'ready', containers: { 'container-id': { status: ContainerHandlingStatus.INITIALIZING } } })
             process.argv = ['foo', 'bar', 'container-id'];