feat(core): Add unique id to instances (#6863)

* add serverId and workerId

* adjust id format

* fix logger init / test

* update implementation

packages/cli/package.json

@@ -25,6 +25,7 @@
     "buildAndDev": "pnpm run build && pnpm run dev",
     "dev": "concurrently -k -n \"TypeScript,Node\" -c \"yellow.bold,cyan.bold\" \"npm run watch\" \"nodemon\"",
     "dev:worker": "concurrently -k -n \"TypeScript,Node\" -c \"yellow.bold,cyan.bold\" \"npm run watch\" \"nodemon worker\"",
+    "dev:webhook": "concurrently -k -n \"TypeScript,Node\" -c \"yellow.bold,cyan.bold\" \"npm run watch\" \"nodemon webhook\"",
     "format": "prettier --write . --ignore-path ../../.prettierignore",
     "lint": "eslint . --quiet --report-unused-disable-directives",
     "lintfix": "eslint . --fix --report-unused-disable-directives",

packages/cli/src/AbstractServer.ts

@@ -7,6 +7,7 @@ import config from '@/config';
 import { N8N_VERSION, inDevelopment, inTest } from '@/constants';
 import { ActiveWorkflowRunner } from '@/ActiveWorkflowRunner';
 import * as Db from '@/Db';
+import { N8nInstanceType } from '@/Interfaces';
 import type { IExternalHooksClass } from '@/Interfaces';
 import { ExternalHooks } from '@/ExternalHooks';
 import { send, sendErrorResponse, ServiceUnavailableError } from '@/ResponseHelper';
@@ -24,6 +25,7 @@ import {
 	EVENT_BUS_REDIS_CHANNEL,
 	WORKER_RESPONSE_REDIS_CHANNEL,
 } from './services/redis/RedisServiceHelper';
+import { generateHostInstanceId } from './databases/utils/generators';
 
 export abstract class AbstractServer {
 	protected server: Server;
@@ -56,7 +58,9 @@ export abstract class AbstractServer {
 
 	protected testWebhooksEnabled = false;
 
-	constructor() {
+	readonly uniqueInstanceId: string;
+
+	constructor(instanceType: N8nInstanceType = 'main') {
 		this.app = express();
 		this.app.disable('x-powered-by');
 
@@ -70,6 +74,8 @@ export abstract class AbstractServer {
 		this.endpointWebhook = config.getEnv('endpoints.webhook');
 		this.endpointWebhookTest = config.getEnv('endpoints.webhookTest');
 		this.endpointWebhookWaiting = config.getEnv('endpoints.webhookWaiting');
+
+		this.uniqueInstanceId = generateHostInstanceId(instanceType);
 	}
 
 	async configure(): Promise<void> {

packages/cli/src/Interfaces.ts

@@ -778,3 +778,5 @@ export interface N8nApp {
 }
 
 export type UserSettings = Pick<User, 'id' | 'settings'>;
+
+export type N8nInstanceType = 'main' | 'webhook' | 'worker';

packages/cli/src/Server.ts

@@ -192,7 +192,7 @@ export class Server extends AbstractServer {
 	push: Push;
 
 	constructor() {
-		super();
+		super('main');
 
 		this.app.engine('handlebars', expressHandlebars({ defaultLayout: false }));
 		this.app.set('view engine', 'handlebars');
@@ -339,6 +339,7 @@ export class Server extends AbstractServer {
 		this.push = Container.get(Push);
 
 		await super.start();
+		LoggerProxy.debug(`Server ID: ${this.uniqueInstanceId}`);
 
 		const cpus = os.cpus();
 		const binaryDataConfig = config.getEnv('binaryDataManager');

packages/cli/src/WebhookServer.ts

@@ -1,3 +1,7 @@
 import { AbstractServer } from '@/AbstractServer';
 
-export class WebhookServer extends AbstractServer {}
+export class WebhookServer extends AbstractServer {
+	constructor() {
+		super('webhook');
+	}
+}

packages/cli/src/commands/webhook.ts

@@ -85,6 +85,7 @@ export class Webhook extends BaseCommand {
 	async run() {
 		await Container.get(Queue).init();
 		await this.server.start();
+		this.logger.debug(`Webhook listener ID: ${this.server.uniqueInstanceId}`);
 		this.logger.info('Webhook listener waiting for requests.');
 
 		// Make sure that the process does not close

packages/cli/src/commands/worker.ts

@@ -23,6 +23,9 @@ import { N8N_VERSION } from '@/constants';
 import { BaseCommand } from './BaseCommand';
 import { ExecutionRepository } from '@db/repositories';
 import { OwnershipService } from '@/services/ownership.service';
+import { generateHostInstanceId } from '@/databases/utils/generators';
+// eslint-disable-next-line import/no-extraneous-dependencies
+import { IConfig } from '@oclif/config';
 
 export class Worker extends BaseCommand {
 	static description = '\nStarts a n8n worker';
@@ -43,6 +46,13 @@ export class Worker extends BaseCommand {
 
 	static jobQueue: JobQueue;
 
+	readonly uniqueInstanceId: string;
+
+	constructor(argv: string[], cmdConfig: IConfig) {
+		super(argv, cmdConfig);
+		this.uniqueInstanceId = generateHostInstanceId('worker');
+	}
+
 	/**
 	 * Stop n8n in a graceful way.
 	 * Make for example sure that all the webhooks from third party services
@@ -227,6 +237,7 @@ export class Worker extends BaseCommand {
 	async init() {
 		await this.initCrashJournal();
 		await super.init();
+		this.logger.debug(`Worker ID: ${this.uniqueInstanceId}`);
 		this.logger.debug('Starting n8n worker...');
 
 		await this.initLicense();

packages/cli/src/databases/utils/generators.ts

@@ -1,6 +1,12 @@
 import { customAlphabet } from 'nanoid';
+import type { N8nInstanceType } from '@/Interfaces';
+
 const nanoid = customAlphabet('0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz', 16);
 
 export function generateNanoId() {
 	return nanoid();
 }
+
+export function generateHostInstanceId(instanceType: N8nInstanceType) {
+	return `${instanceType}-${nanoid()}`;
+}