feat: algorithm-queue scale

core/algorithm-operator/bootstrap.js

@@ -8,13 +8,15 @@ const db = require('./lib/helpers/db');
 const etcd = require('./lib/helpers/etcd');
 const kubernetes = require('./lib/helpers/kubernetes');
 const operator = require('./lib/operator');
+const jobsMessageQueue = require('./lib/helpers/jobs-message-queue');
 const { setFromConfig } = require('./lib/helpers/settings');
 
 const modules = [
     db,
     etcd,
     kubernetes,
-    operator
+    operator,
+    jobsMessageQueue
 ];
 
 class Bootstrap {

core/algorithm-operator/config/main/config.base.js

@@ -2,6 +2,7 @@ const packageJson = require(process.cwd() + '/package.json');
 const formatter = require('../../lib/helpers/formatters');
 const config = module.exports = {};
 
+const useSentinel = !!process.env.REDIS_SENTINEL_SERVICE_HOST;
 config.serviceName = packageJson.name;
 config.version = packageJson.version;
 config.intervalMs = process.env.INTERVAL_MS || 10000;
@@ -15,6 +16,17 @@ config.algorithmQueueBalancer = {
     limit: formatter.parseInt(process.env.ALGORITHM_QUEUE_CONCURRENCY_LIMIT, 5)
 };
 
+config.JobsMessageQueue = {
+    enableCheckStalledJobs: false,
+    prefix: 'algorithm-queue',
+};
+
+config.redis = {
+    host: useSentinel ? process.env.REDIS_SENTINEL_SERVICE_HOST : process.env.REDIS_SERVICE_HOST || 'localhost',
+    port: useSentinel ? process.env.REDIS_SENTINEL_SERVICE_PORT : process.env.REDIS_SERVICE_PORT || 6379,
+    sentinel: useSentinel,
+};
+
 config.driversSetting = {
     name: 'pipeline-driver',
     concurrency: formatter.parseInt(process.env.PIPELINE_DRIVERS_CONCURRENCY_LIMIT, 5),

core/algorithm-operator/lib/helpers/jobs-message-queue.js

@@ -0,0 +1,37 @@
+const { Producer } = require('@hkube/producer-consumer');
+
+class JobsMessageQueue {
+    async init(options) {
+        this._producer = new Producer({
+            setting: {
+                redis: options.redis,
+                ...options.JobsMessageQueue
+            }
+        });
+    }
+
+    async getWaitingCount(algorithms) {
+        const map = {};
+        await Promise.all(algorithms.map(async (a) => {
+            const count = await this._getWaitingCount(a.name);
+            if (count >= 0) {
+                map[a.name] = count;
+            }
+        }));
+        return map;
+    }
+
+    async _getWaitingCount(algorithm) {
+        let count = 0;
+        try {
+            const queue = this._producer._createQueue(algorithm);
+            count = await queue.getWaitingCount();
+        }
+        catch {
+            count = null;
+        }
+        return count;
+    }
+}
+
+module.exports = new JobsMessageQueue();

core/algorithm-operator/lib/reconcile/algorithm-queue.js

@@ -8,6 +8,7 @@ const { findVersion } = require('../helpers/images');
 const component = require('../consts/componentNames').ALGORITHM_QUEUE_RECONCILER;
 const QueueActions = require('../consts/queue-actions');
 const { normalizeQueuesDeployments, normalizeQueuesDiscovery, normalizeAlgorithms } = require('./normalize');
+const jobsMessageQueue = require('../helpers/jobs-message-queue');
 const CONTAINERS = require('../consts/containers');
 
 const _createDeployment = async ({ queueId, options }) => {
@@ -117,15 +118,16 @@ const reconcile = async ({ deployments, algorithms, discovery, versions, registr
     const normAlgorithms = normalizeAlgorithms(algorithms);
     const normDeployments = normalizeQueuesDeployments(deployments);
     const availableQueues = _findAvailableQueues({ queueToAlgorithms, limit });
-    const addAlgorithms = normAlgorithms.filter(a => !algorithmsToQueue[a.name]);
     const removeAlgorithms = _findObsoleteAlgorithms({ algorithmsToQueue, normAlgorithms });
+    const waitingCount = await jobsMessageQueue.getWaitingCount(algorithms);
+    const requiredAlgorithms = normAlgorithms.filter(a => !algorithmsToQueue[a.name] && waitingCount[a.name] > 0);
 
     if (!devMode) {
-        await _addDeployments({ limit, algorithms: normAlgorithms.length, deployments: normDeployments.length, versions, registry, clusterOptions, resources, options });
+        await _addDeployments({ limit, algorithms: requiredAlgorithms.length, deployments: normDeployments.length, versions, registry, clusterOptions, resources, options });
         await _updateDeployments({ normDeployments, options: { versions, registry, clusterOptions, resources, options } });
         await _deleteDeployments({ queues: queueToAlgorithms, normDeployments });
     }
-    await _matchAlgorithmsToQueue({ algorithms: addAlgorithms, queues: availableQueues, limit });
+    await _matchAlgorithmsToQueue({ algorithms: requiredAlgorithms, queues: availableQueues, limit });
     await _removeAlgorithmsFromQueue({ algorithms: removeAlgorithms });
     await _removeDuplicatesAlgorithms({ algorithms: duplicateAlgorithms });
 };