autoscaler interval log

core/task-executor/lib/reconcile/reconciler.js

@@ -682,6 +682,15 @@ const reconcile = async ({ algorithmTemplates, algorithmRequests, workers, jobs,
                 required: 0
             };
         }
+        const _created = reconcileResult[algorithmName].created;
+        const _skipped = reconcileResult[algorithmName].skipped;
+        const { paused, resumed, required } = reconcileResult[algorithmName];
+        const total = _created + _skipped + paused + resumed + required;
+        if (total !== 0) {
+            log.info(`CYCLE: task-executor: algo: ${algorithmName} created: ${_created}, 
+                skipped: ${_skipped}, paused: ${paused}, 
+                resumed: ${resumed}, required: ${required}.`);
+        }
         reconcileResult[algorithmName].active = ws.count;
     });
     return reconcileResult;

core/worker/config/main/config.base.js

@@ -32,17 +32,10 @@ config.streaming = {
             minTimeNonStatsReport: formatters.parseInt(process.env.AUTO_SCALER_NON_STATS_REPORT, 10000),
         },
         scaleUp: {
-            replicasExtra: formatters.parseInt(process.env.AUTO_SCALER_EXTRA_REPLICAS, 0.35),
             maxScaleUpReplicasPerNode: formatters.parseInt(process.env.AUTO_SCALER_MAX_REPLICAS, 1000),
             maxScaleUpReplicasPerTick: formatters.parseInt(process.env.AUTO_SCALER_MAX_REPLICAS_PER_SCALE, 10),
             replicasOnFirstScale: formatters.parseInt(process.env.AUTO_SCALER_REPLICAS_FIRST_SCALE, 1),
-            minTimeToCleanUpQueue: formatters.parseInt(process.env.AUTO_SCALER_MIN_TIME_CLEAN_QUEUE, 30),
-        },
-        scaleDown: {
-            tolerance: formatters.parseInt(process.env.AUTO_SCALER_SCALE_DOWN_TOLERANCE, 0.4),
-            minTimeIdleBeforeReplicaDown: formatters.parseInt(process.env.AUTO_SCALER_MIN_TIME_WAIT_REPLICA_DOWN, 60000),
-            minQueueSizeBeforeScaleDown: formatters.parseInt(process.env.AUTO_SCALER_MIN_QUEUE_SIZE_BEFORE_SCALE_DOWN, 0),
-            minTimeQueueEmptyBeforeScaleDown: formatters.parseInt(process.env.AUTO_SCALER_MIN_TIME_QUEUE_EMPTY, 60000),
+            minTimeToCleanUpQueue: formatters.parseInt(process.env.AUTO_SCALER_MIN_TIME_CLEAN_QUEUE, 30), // seconds
         },
         scaleIntervention: {
             throttleMs: formatters.parseInt(process.env.SCALE_INTERVENTION_LOG_THROTTLE_TIME, 200)

core/worker/lib/streaming/core/metrics.js

@@ -1,11 +1,16 @@
 const { mean } = require('@hkube/stats');
+const Logger = require('@hkube/logger');
 
 const _calcRate = (list) => {
     let first = list[0];
     if (list.length === 1) {
         first = { time: first.time - 2000, count: 0 };
     }
     const last = list[list.length - 1];
+
+    const log = Logger.GetLogFromContainer();
+    log.info(`STATISTICS: first value: ${first.count}, last value: ${last.count}, time diff: ${last.time - first.time} ms`);
+
     const timeDiff = (last.time - first.time) / 1000;
     const countDiff = last.count - first.count;
     let rate = 0;

core/worker/lib/streaming/core/scaler.js

@@ -40,6 +40,7 @@ class Scaler {
         this._status = SCALE_STATUS.IDLE;
         this._startInterval();
         this._minStatelessCount = minStatelessCount;
+        this._isQueueEmpty = true;
     }
 
     stop() {
@@ -58,38 +59,37 @@ class Scaler {
             return;
         }
 
-        let pendingUp = false;
         this._status = SCALE_STATUS.IDLE;
         const unScheduledAlgorithm = await this._getUnScheduledAlgorithm();
 
         if (unScheduledAlgorithm) {
             this._status = `${SCALE_STATUS.UNABLE_SCALE} ${unScheduledAlgorithm.message}`;
-            pendingUp = true;
         }
         else {
             const queue = await this._getQueue();
             if (queue) {
                 this._status = SCALE_STATUS.PENDING_QUEUE;
-                pendingUp = true;
             }
         }
 
         const currentSize = this._getCurrentSize();
-        const shouldScaleUp = pendingUp ? false : this._shouldScaleUp(currentSize);
-        const shouldScaleDown = this._shouldScaleDown(currentSize);
+        const shouldScaleUp = this._shouldScaleUp(currentSize);
+        const shouldScaleDown = this._isQueueEmpty && this._shouldScaleDown(currentSize);
 
         if (shouldScaleUp) {
-            const required = this._required - currentSize;
+            const required = this._required - this._desired;
             const replicas = Math.min(required, this._maxScaleUpReplicasPerTick);
+            log.info(`CYCLE: worker shouldScaleUp required: ${required}, replicas: ${replicas}, desired: ${this._desired}, currentSize: ${currentSize}`);
             const scaleTo = replicas + currentSize;
-            this._desired = this._required;
+            this._desired += replicas;
             this._lastScaleUpTime = Date.now();
             this._status = SCALE_STATUS.SCALING_UP;
             this._scaleUp({ replicas, currentSize, scaleTo });
         }
         if (shouldScaleDown) {
             const replicas = currentSize - this._required;
             const scaleTo = this._required;
+            log.info(`CYCLE: worker shouldScaleDown scaleTo: ${scaleTo}, replicas: ${replicas}, desired: ${this._desired}, currentSize: ${currentSize}`);
             this._desired = this._required;
             this._lastScaleDownTime = Date.now();
             this._status = SCALE_STATUS.SCALING_DOWN;
@@ -109,14 +109,17 @@ class Scaler {
         return this._status;
     }
 
-    updateRequired(required) {
-        this._scale = true;
-        this._required = Math.min(required, this._maxScaleUpReplicasPerNode);
+    updateRequired(required, isQueueEmpty) {
+        this._isQueueEmpty = isQueueEmpty;
+        if (required !== this._required) {
+            this._scale = true;
+            this._required = Math.min(required, this._maxScaleUpReplicasPerNode);
+        }
     }
 
     _shouldScaleUp(currentSize) {
         let shouldScaleUp = false;
-        if (currentSize < this._required
+        if (currentSize < this._required && this._desired < this._required
             && (!this._lastScaleDownTime || Date.now() - this._lastScaleDownTime > this._minTimeBetweenScales)) {
             if (this._desired <= currentSize) {
                 shouldScaleUp = true;
@@ -140,11 +143,7 @@ class Scaler {
 
     _shouldScaleDown(currentSize) {
         let shouldScaleDown = false;
-        let limitScaleDown = false;
-        if ((this.minStatelessCount > 0)) {
-            limitScaleDown = (this._minStatelessCount >= this._required);
-        }
-        if (currentSize > this._required && !limitScaleDown
+        if (currentSize > this._required
             && (!this._lastScaleUpTime || Date.now() - this._lastScaleUpTime > this._minTimeBetweenScales)) {
             if (this._desired >= currentSize) {
                 shouldScaleDown = true;

core/worker/lib/streaming/core/statistics.js

@@ -58,11 +58,11 @@ class Statistics {
 
     _createStatData({ maxSize }) {
         return {
-            requests: new FixedWindow(maxSize),
+            requests: new FixedWindow(maxSize), // Brings one value every 2 seconds, meaning for a window_size of 10 we will consider reports of last 20 seconds.
             responses: new FixedWindow(maxSize),
-            durations: new FixedWindow(maxSize),
-            grossDurations: new FixedWindow(maxSize),
-            queueDurations: new FixedWindow(maxSize),
+            durations: new FixedWindow(maxSize * 10), // Brings window_size values every 2 seconds, so for a window_size multiplied by 10 we will consider values that occured in the last 20 seconds.
+            grossDurations: new FixedWindow(maxSize * 10),
+            queueDurations: new FixedWindow(maxSize * 10),
         };
     }
 }