Merge pull request #6899 from nocodb/nc-feat/worker-cluster

fix: worker

packages/nocodb/src/Noco.ts

@@ -107,9 +107,8 @@ export default class Noco {
         throw new Error('NC_REDIS_URL is required');
       }
       process.env.NC_DISABLE_TELE = 'true';
+    }
 
-      nestApp.init();
-    } else {
     nestApp.useWebSocketAdapter(new IoAdapter(httpServer));
 
     this._httpServer = nestApp.getHttpAdapter().getInstance();
@@ -136,7 +135,6 @@ export default class Noco {
 
     return nestApp.getHttpAdapter().getInstance();
   }
-  }
 
   public static get httpServer(): http.Server {
     return this._httpServer;

packages/nocodb/src/interface/Jobs.ts

@@ -26,3 +26,14 @@ export enum JobEvents {
   STATUS = 'job.status',
   LOG = 'job.log',
 }
+
+export enum InstanceTypes {
+  PRIMARY = 'primary',
+  WORKER = 'worker',
+}
+
+export enum WorkerCommands {
+  RESUME_LOCAL = 'resumeLocal',
+  PAUSE_LOCAL = 'pauseLocal',
+  RESET = 'reset',
+}

packages/nocodb/src/modules/jobs/fallback/jobs.service.ts

@@ -48,4 +48,12 @@ export class JobsService {
 
     return job;
   }
+
+  async resumeQueue() {
+    await this.fallbackQueueService.queue.start();
+  }
+
+  async pauseQueue() {
+    await this.fallbackQueueService.queue.pause();
+  }
 }

packages/nocodb/src/modules/jobs/redis/jobs-redis.service.ts

@@ -1,5 +1,6 @@
 import { Injectable } from '@nestjs/common';
 import Redis from 'ioredis';
+import { InstanceTypes } from '~/interface/Jobs';
 
 @Injectable()
 export class JobsRedisService {
@@ -7,9 +8,28 @@ export class JobsRedisService {
   private redisSubscriber: Redis;
   private unsubscribeCallbacks: { [key: string]: () => void } = {};
 
+  public primaryCallbacks: { [key: string]: () => void } = {};
+  public workerCallbacks: { [key: string]: () => void } = {};
+
   constructor() {
     this.redisClient = new Redis(process.env.NC_REDIS_JOB_URL);
     this.redisSubscriber = new Redis(process.env.NC_REDIS_JOB_URL);
+
+    if (process.env.NC_WORKER_CONTAINER === 'true') {
+      this.redisSubscriber.subscribe(InstanceTypes.WORKER);
+    } else {
+      this.redisSubscriber.subscribe(InstanceTypes.PRIMARY);
+    }
+
+    const onMessage = (channel, message) => {
+      if (channel === InstanceTypes.WORKER) {
+        this.workerCallbacks[message] && this.workerCallbacks[message]();
+      } else if (channel === InstanceTypes.PRIMARY) {
+        this.primaryCallbacks[message] && this.primaryCallbacks[message]();
+      }
+    };
+
+    this.redisSubscriber.on('message', onMessage);
   }
 
   publish(channel: string, message: string | any) {
@@ -47,4 +67,20 @@ export class JobsRedisService {
       delete this.unsubscribeCallbacks[channel];
     }
   }
+
+  workerCount(): Promise<number> {
+    return new Promise((resolve, reject) => {
+      this.redisClient.publish(
+        InstanceTypes.WORKER,
+        'count',
+        (error, numberOfSubscribers) => {
+          if (error) {
+            reject(0);
+          } else {
+            resolve(numberOfSubscribers);
+          }
+        },
+      );
+    });
+  }
 }

packages/nocodb/src/modules/jobs/redis/jobs.service.ts

@@ -1,32 +1,52 @@
 import { InjectQueue } from '@nestjs/bull';
-import { Injectable } from '@nestjs/common';
+import { Injectable, Logger } from '@nestjs/common';
 import { Queue } from 'bull';
 import type { OnModuleInit } from '@nestjs/common';
-import { JOBS_QUEUE, JobStatus } from '~/interface/Jobs';
+import { JOBS_QUEUE, JobStatus, WorkerCommands } from '~/interface/Jobs';
+import { JobsRedisService } from '~/modules/jobs/redis/jobs-redis.service';
 
 @Injectable()
 export class JobsService implements OnModuleInit {
-  constructor(@InjectQueue(JOBS_QUEUE) protected readonly jobsQueue: Queue) {}
+  protected logger = new Logger(JobsService.name);
+
+  constructor(
+    @InjectQueue(JOBS_QUEUE) public readonly jobsQueue: Queue,
+    protected readonly jobsRedisService: JobsRedisService,
+  ) {}
 
   // pause primary instance queue
   async onModuleInit() {
     if (process.env.NC_WORKER_CONTAINER !== 'true') {
-      // await this.jobsQueue.pause(true);
+      await this.jobsQueue.pause(true);
+    } else {
+      this.jobsRedisService.workerCallbacks[WorkerCommands.RESUME_LOCAL] =
+        async () => {
+          this.logger.log('Resuming local queue');
+          await this.jobsQueue.resume(true);
+        };
+      this.jobsRedisService.workerCallbacks[WorkerCommands.PAUSE_LOCAL] =
+        async () => {
+          this.logger.log('Pausing local queue');
+          await this.jobsQueue.pause(true);
+        };
     }
   }
 
   async add(name: string, data: any) {
+    // if NC_WORKER_CONTAINER is false, then skip dynamic queue pause/resume
+    if (process.env.NC_WORKER_CONTAINER !== 'false') {
       // resume primary instance queue if there is no worker
-    /* const workerCount = (await this.jobsQueue.getWorkers()).length;
+      const workerCount = await this.jobsRedisService.workerCount();
       const localWorkerPaused = await this.jobsQueue.isPaused(true);
 
       // if there is no worker and primary instance queue is paused, resume it
       // if there is any worker and primary instance queue is not paused, pause it
-    if (workerCount === 1 && localWorkerPaused) {
+      if (workerCount === 0 && localWorkerPaused) {
         await this.jobsQueue.resume(true);
-    } else if (workerCount > 1 && !localWorkerPaused) {
+      } else if (workerCount > 0 && !localWorkerPaused) {
         await this.jobsQueue.pause(true);
-    } */
+      }
+    }
 
     const job = await this.jobsQueue.add(name, data);
 
@@ -72,4 +92,14 @@ export class JobsService implements OnModuleInit {
 
     return job;
   }
+
+  async resumeQueue() {
+    this.logger.log('Resuming global queue');
+    await this.jobsQueue.resume();
+  }
+
+  async pauseQueue() {
+    this.logger.log('Pausing global queue');
+    await this.jobsQueue.pause();
+  }
 }

packages/nocodb/src/modules/metas/metas.module.ts

@@ -184,6 +184,7 @@ export const metaModuleMetadata = {
     MetaDiffsService,
     BasesService,
     SourcesService,
+    UtilsService,
   ],
 };
 

packages/nocodb/src/run/cloud.ts

@@ -18,12 +18,8 @@ server.use(
 server.set('view engine', 'ejs');
 
 (async () => {
-  if (process.env.NC_WORKER_CONTAINER === 'true') {
-    await Noco.init({}, null, null);
-  } else {
   const httpServer = server.listen(process.env.PORT || 8080, async () => {
     console.log(`App started successfully.\nVisit -> ${Noco.dashboardUrl}`);
     server.use(await Noco.init({}, httpServer, server));
   });
-  }
 })().catch((e) => console.log(e));

packages/nocodb/src/run/docker.ts

@@ -28,11 +28,7 @@ process.env[`DEBUG`] = 'xc*';
 // })().catch((e) => console.log(e));
 
 (async () => {
-  if (process.env.NC_WORKER_CONTAINER === 'true') {
-    await Noco.init({}, null, null);
-  } else {
   const httpServer = server.listen(process.env.PORT || 8080, async () => {
     server.use(await Noco.init({}, httpServer, server));
   });
-  }
 })().catch((e) => console.log(e));

packages/nocodb/src/run/dockerEntry.ts

@@ -13,11 +13,7 @@ server.use(cors());
 server.set('view engine', 'ejs');
 
 (async () => {
-  if (process.env.NC_WORKER_CONTAINER === 'true') {
-    await Noco.init({}, null, null);
-  } else {
   const httpServer = server.listen(process.env.PORT || 8080, async () => {
     server.use(await Noco.init({}, httpServer, server));
   });
-  }
 })().catch((e) => console.log(e));

packages/nocodb/src/run/dockerRunMysql.ts

@@ -31,11 +31,7 @@ process.env[`NC_DB`] = `mysql2://localhost:3306?u=root&p=password&d=${metaDb}`;
 // process.env[`DEBUG`] = 'xc*';
 
 (async () => {
-  if (process.env.NC_WORKER_CONTAINER === 'true') {
-    await Noco.init({}, null, null);
-  } else {
   const httpServer = server.listen(process.env.PORT || 8080, async () => {
     server.use(await Noco.init({}, httpServer, server));
   });
-  }
 })().catch((e) => console.log(e));

packages/nocodb/src/run/dockerRunPG.ts

@@ -30,11 +30,7 @@ process.env[`NC_DB`] = `pg://localhost:5432?u=postgres&p=password&d=${metaDb}`;
 // process.env[`DEBUG`] = 'xc*';
 
 (async () => {
-  if (process.env.NC_WORKER_CONTAINER === 'true') {
-    await Noco.init({}, null, null);
-  } else {
   const httpServer = server.listen(process.env.PORT || 8080, async () => {
     server.use(await Noco.init({}, httpServer, server));
   });
-  }
 })().catch((e) => console.log(e));

packages/nocodb/src/run/dockerRunPG_CyQuick.ts

@@ -24,11 +24,7 @@ process.env[
 //process.env[`DEBUG`] = 'xc*';
 
 (async () => {
-  if (process.env.NC_WORKER_CONTAINER === 'true') {
-    await Noco.init({}, null, null);
-  } else {
   const httpServer = server.listen(process.env.PORT || 8080, async () => {
     server.use(await Noco.init({}, httpServer, server));
   });
-  }
 })().catch((e) => console.log(e));

packages/nocodb/src/run/local.ts

@@ -17,12 +17,8 @@ server.use(
 server.set('view engine', 'ejs');
 
 (async () => {
-  if (process.env.NC_WORKER_CONTAINER === 'true') {
-    await Noco.init({}, null, null);
-  } else {
   const httpServer = server.listen(process.env.PORT || 8080, async () => {
     console.log(`App started successfully.\nVisit -> ${Noco.dashboardUrl}`);
     server.use(await Noco.init({}, httpServer, server));
   });
-  }
 })().catch((e) => console.log(e));

packages/nocodb/src/run/testDocker.ts

@@ -26,13 +26,20 @@ process.env[`DEBUG`] = 'xc*';
 process.env[`NC_ALLOW_LOCAL_HOOKS`] = 'true';
 
 (async () => {
+  if (process.env.NC_WORKER_CONTAINER === 'true') {
+    const httpServer = server.listen(process.env.PORT || 8080, async () => {
+      server.use(await Noco.init({}, httpServer, server));
+    });
+  } else {
     const httpServer = server.listen(process.env.PORT || 8080, async () => {
       server.use(await Noco.init({}, httpServer, server));
 
       let admin_response;
       if (!(await User.getByEmail('user@nocodb.com'))) {
         admin_response = await axios.post(
-        `http://localhost:${process.env.PORT || 8080}/api/v1/auth/user/signup`,
+          `http://localhost:${
+            process.env.PORT || 8080
+          }/api/v1/auth/user/signup`,
           {
             email: 'user@nocodb.com',
             password: 'Password123.',
@@ -79,4 +86,5 @@ process.env[`NC_ALLOW_LOCAL_HOOKS`] = 'true';
         }
       }
     });
+  }
 })().catch((e) => console.log(e));