admin/n8n-enterprise-unlocked
1
0
Fork 0
mirror of https://github.com/Abdulazizzn/n8n-enterprise-unlocked.git synced 2025-12-17 18:12:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): Delete duplicate code across all commands (#5452)

Browse Source
This commit is contained in:
कारतोफ्फेलस्क्रिप्ट™
2023-02-10 14:59:20 +01:00
committed by GitHub
parent 8494c97821
commit 5194513850
26 changed files with 979 additions and 1345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

384
packages/cli/src/commands/worker.ts
View File

@@ -1,54 +1,28 @@
/* eslint-disable @typescript-eslint/no-unsafe-argument */
/* eslint-disable @typescript-eslint/no-shadow */
/* eslint-disable @typescript-eslint/unbound-method */
/* eslint-disable @typescript-eslint/no-unsafe-member-access */
/* eslint-disable @typescript-eslint/restrict-template-expressions */
import express from 'express';
import http from 'http';
import type PCancelable from 'p-cancelable';
import { Command, flags } from '@oclif/command';
import { BinaryDataManager, UserSettings, WorkflowExecute } from 'n8n-core';
import { flags } from '@oclif/command';
import { WorkflowExecute } from 'n8n-core';
import type { IExecuteResponsePromiseData, INodeTypes, IRun } from 'n8n-workflow';
import { Workflow, LoggerProxy, ErrorReporterProxy as ErrorReporter, sleep } from 'n8n-workflow';
import { Workflow, NodeOperationError, LoggerProxy, sleep } from 'n8n-workflow';
import { CredentialsOverwrites } from '@/CredentialsOverwrites';
import { CredentialTypes } from '@/CredentialTypes';
import * as Db from '@/Db';
import { ExternalHooks } from '@/ExternalHooks';
import { NodeTypes } from '@/NodeTypes';
import * as ResponseHelper from '@/ResponseHelper';
import * as WebhookHelpers from '@/WebhookHelpers';
import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData';
import { InternalHooksManager } from '@/InternalHooksManager';
import { LoadNodesAndCredentials } from '@/LoadNodesAndCredentials';
import { getLogger } from '@/Logger';
import { PermissionChecker } from '@/UserManagement/PermissionChecker';
import config from '@/config';
import * as Queue from '@/Queue';
import * as CrashJournal from '@/CrashJournal';
import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
import { generateFailedExecutionFromError } from '@/WorkflowHelpers';
import { N8N_VERSION } from '@/constants';
import { initErrorHandling } from '@/ErrorReporting';
import { BaseCommand } from './BaseCommand';
const exitWithCrash = async (message: string, error: unknown) => {
ErrorReporter.error(new Error(message, { cause: error }), { level: 'fatal' });
await sleep(2000);
process.exit(1);
};
const exitSuccessFully = async () => {
try {
await CrashJournal.cleanup();
} finally {
process.exit();
}
};
export class Worker extends Command {
export class Worker extends BaseCommand {
static description = '\nStarts a n8n worker';
static examples = ['$ n8n worker --concurrency=5'];
@@ -72,7 +46,7 @@ export class Worker extends Command {
* Make for example sure that all the webhooks from third party services
* get removed.
*/
static async stopProcess() {
async stopProcess() {
LoggerProxy.info('Stopping n8n...');
// Stop accepting new jobs
@@ -80,8 +54,7 @@ export class Worker extends Command {
Worker.jobQueue.pause(true);
try {
const externalHooks = ExternalHooks();
await externalHooks.run('n8n.stop', []);
await this.externalHooks.run('n8n.stop', []);
const maxStopTime = config.getEnv('queue.bull.gracefulShutdownTimeout') * 1000;
@@ -90,7 +63,7 @@ export class Worker extends Command {
setTimeout(async () => {
// In case that something goes wrong with shutdown we
// kill after max. 30 seconds no matter what
await exitSuccessFully();
await this.exitSuccessFully();
}, maxStopTime);
// Wait for active workflow executions to finish
@@ -108,10 +81,10 @@ export class Worker extends Command {
await sleep(500);
}
} catch (error) {
await exitWithCrash('There was an error shutting down n8n.', error);
await this.exitWithCrash('There was an error shutting down n8n.', error);
}
await exitSuccessFully();
await this.exitSuccessFully();
}
async runJob(job: Queue.Job, nodeTypes: INodeTypes): Promise<Queue.JobResponse> {
@@ -128,31 +101,27 @@ export class Worker extends Command {
);
}
const currentExecutionDb = ResponseHelper.unflattenExecutionData(executionDb);
const workflowId = currentExecutionDb.workflowData.id!;
LoggerProxy.info(
`Start job: ${job.id} (Workflow ID: ${currentExecutionDb.workflowData.id} | Execution: ${executionId})`,
`Start job: ${job.id} (Workflow ID: ${workflowId} | Execution: ${executionId})`,
);
const workflowOwner = await getWorkflowOwner(currentExecutionDb.workflowData.id!.toString());
const workflowOwner = await getWorkflowOwner(workflowId);
let { staticData } = currentExecutionDb.workflowData;
if (loadStaticData) {
const workflowData = await Db.collections.Workflow.findOne({
select: ['id', 'staticData'],
where: {
id: currentExecutionDb.workflowData.id,
id: workflowId,
},
});
if (workflowData === null) {
LoggerProxy.error(
'Worker execution failed because workflow could not be found in database.',
{
workflowId: currentExecutionDb.workflowData.id,
executionId,
},
);
throw new Error(
`The workflow with the ID "${currentExecutionDb.workflowData.id}" could not be found`,
{ workflowId, executionId },
);
throw new Error(`The workflow with the ID "${workflowId}" could not be found`);
}
staticData = workflowData.staticData;
}
@@ -173,7 +142,7 @@ export class Worker extends Command {
}
const workflow = new Workflow({
id: currentExecutionDb.workflowData.id as string,
id: workflowId,
name: currentExecutionDb.workflowData.name,
nodes: currentExecutionDb.workflowData.nodes,
connections: currentExecutionDb.workflowData.connections,
@@ -198,15 +167,15 @@ export class Worker extends Command {
try {
await PermissionChecker.check(workflow, workflowOwner.id);
} catch (error) {
const failedExecution = generateFailedExecutionFromError(
currentExecutionDb.mode,
error,
error.node,
);
await additionalData.hooks.executeHookFunctions('workflowExecuteAfter', [failedExecution]);
return {
success: true,
};
if (error instanceof NodeOperationError) {
const failedExecution = generateFailedExecutionFromError(
currentExecutionDb.mode,
error,
error.node,
);
await additionalData.hooks.executeHookFunctions('workflowExecuteAfter', [failedExecution]);
}
return { success: true };
}
additionalData.hooks.hookFunctions.sendResponse = [
@@ -249,184 +218,149 @@ export class Worker extends Command {
};
}
async run() {
const logger = getLogger();
LoggerProxy.init(logger);
// eslint-disable-next-line no-console
console.info('Starting n8n worker...');
async init() {
// Make sure that n8n shuts down gracefully if possible
process.once('SIGTERM', Worker.stopProcess);
process.once('SIGINT', Worker.stopProcess);
process.once('SIGTERM', this.stopProcess);
process.once('SIGINT', this.stopProcess);
await initErrorHandling();
await CrashJournal.init();
await this.initCrashJournal();
await super.init();
this.logger.debug('Starting n8n worker...');
// Wrap that the process does not close but we can still use async
await (async () => {
try {
const { flags } = this.parse(Worker);
await this.initBinaryManager();
await this.initExternalHooks();
}
// Start directly with the init of the database to improve startup time
const startDbInitPromise = Db.init().catch(async (error: Error) =>
exitWithCrash('There was an error initializing DB', error),
);
async run() {
// eslint-disable-next-line @typescript-eslint/no-shadow
const { flags } = this.parse(Worker);
// Make sure the settings exist
await UserSettings.prepareUserSettings();
// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
const redisConnectionTimeoutLimit = config.getEnv('queue.bull.redis.timeoutThreshold');
// Load all node and credential types
const loadNodesAndCredentials = LoadNodesAndCredentials();
await loadNodesAndCredentials.init();
const queue = await Queue.getInstance();
Worker.jobQueue = queue.getBullObjectInstance();
// eslint-disable-next-line @typescript-eslint/no-floating-promises
Worker.jobQueue.process(flags.concurrency, async (job) => this.runJob(job, this.nodeTypes));
// Add the found types to an instance other parts of the application can use
const nodeTypes = NodeTypes(loadNodesAndCredentials);
const credentialTypes = CredentialTypes(loadNodesAndCredentials);
this.logger.info('\nn8n worker is now ready');
this.logger.info(` * Version: ${N8N_VERSION}`);
this.logger.info(` * Concurrency: ${flags.concurrency}`);
this.logger.info('');
// Load the credentials overwrites if any exist
CredentialsOverwrites(credentialTypes);
Worker.jobQueue.on('global:progress', (jobId: Queue.JobId, progress) => {
// Progress of a job got updated which does get used
// to communicate that a job got canceled.
// Load all external hooks
const externalHooks = ExternalHooks();
await externalHooks.init();
// Wait till the database is ready
await startDbInitPromise;
// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
const redisConnectionTimeoutLimit = config.getEnv('queue.bull.redis.timeoutThreshold');
const queue = await Queue.getInstance();
Worker.jobQueue = queue.getBullObjectInstance();
// eslint-disable-next-line @typescript-eslint/no-floating-promises
Worker.jobQueue.process(flags.concurrency, async (job) => this.runJob(job, nodeTypes));
const instanceId = await UserSettings.getInstanceId();
await InternalHooksManager.init(instanceId, nodeTypes);
const binaryDataConfig = config.getEnv('binaryDataManager');
await BinaryDataManager.init(binaryDataConfig);
console.info('\nn8n worker is now ready');
console.info(` * Version: ${N8N_VERSION}`);
console.info(` * Concurrency: ${flags.concurrency}`);
console.info('');
Worker.jobQueue.on('global:progress', (jobId, progress) => {
// Progress of a job got updated which does get used
// to communicate that a job got canceled.
if (progress === -1) {
// Job has to get canceled
if (Worker.runningJobs[jobId] !== undefined) {
// Job is processed by current worker so cancel
Worker.runningJobs[jobId].cancel();
delete Worker.runningJobs[jobId];
}
}
});
let lastTimer = 0;
let cumulativeTimeout = 0;
Worker.jobQueue.on('error', (error: Error) => {
if (error.toString().includes('ECONNREFUSED')) {
const now = Date.now();
if (now - lastTimer > 30000) {
// Means we had no timeout at all or last timeout was temporary and we recovered
lastTimer = now;
cumulativeTimeout = 0;
} else {
cumulativeTimeout += now - lastTimer;
lastTimer = now;
if (cumulativeTimeout > redisConnectionTimeoutLimit) {
logger.error(
`Unable to connect to Redis after ${redisConnectionTimeoutLimit}. Exiting process.`,
);
process.exit(1);
}
}
logger.warn('Redis unavailable - trying to reconnect...');
} else if (error.toString().includes('Error initializing Lua scripts')) {
// This is a non-recoverable error
// Happens when worker starts and Redis is unavailable
// Even if Redis comes back online, worker will be zombie
logger.error('Error initializing worker.');
process.exit(2);
} else {
logger.error('Error from queue: ', error);
throw error;
}
});
if (config.getEnv('queue.health.active')) {
const port = config.getEnv('queue.health.port');
const app = express();
app.disable('x-powered-by');
const server = http.createServer(app);
app.get(
'/healthz',
// eslint-disable-next-line consistent-return
async (req: express.Request, res: express.Response) => {
LoggerProxy.debug('Health check started!');
const connection = Db.getConnection();
try {
if (!connection.isInitialized) {
// Connection is not active
throw new Error('No active database connection!');
}
// DB ping
await connection.query('SELECT 1');
} catch (e) {
LoggerProxy.error('No Database connection!', e);
const error = new ResponseHelper.ServiceUnavailableError('No Database connection!');
return ResponseHelper.sendErrorResponse(res, error);
}
// Just to be complete, generally will the worker stop automatically
// if it loses the connection to redis
try {
// Redis ping
await Worker.jobQueue.client.ping();
} catch (e) {
LoggerProxy.error('No Redis connection!', e);
const error = new ResponseHelper.ServiceUnavailableError('No Redis connection!');
return ResponseHelper.sendErrorResponse(res, error);
}
// Everything fine
const responseData = {
status: 'ok',
};
LoggerProxy.debug('Health check completed successfully!');
ResponseHelper.sendSuccessResponse(res, responseData, true, 200);
},
);
server.listen(port, () => {
console.info(`\nn8n worker health check via, port ${port}`);
});
server.on('error', (error: Error & { code: string }) => {
if (error.code === 'EADDRINUSE') {
console.log(
`n8n's port ${port} is already in use. Do you have the n8n main process running on that port?`,
);
process.exit(1);
}
});
if (progress === -1) {
// Job has to get canceled
if (Worker.runningJobs[jobId] !== undefined) {
// Job is processed by current worker so cancel
Worker.runningJobs[jobId].cancel();
delete Worker.runningJobs[jobId];
}
} catch (error) {
await exitWithCrash('Worker process cannot continue.', error);
}
})();
});
let lastTimer = 0;
let cumulativeTimeout = 0;
Worker.jobQueue.on('error', (error: Error) => {
if (error.toString().includes('ECONNREFUSED')) {
const now = Date.now();
if (now - lastTimer > 30000) {
// Means we had no timeout at all or last timeout was temporary and we recovered
lastTimer = now;
cumulativeTimeout = 0;
} else {
cumulativeTimeout += now - lastTimer;
lastTimer = now;
if (cumulativeTimeout > redisConnectionTimeoutLimit) {
this.logger.error(
`Unable to connect to Redis after ${redisConnectionTimeoutLimit}. Exiting process.`,
);
process.exit(1);
}
}
this.logger.warn('Redis unavailable - trying to reconnect...');
} else if (error.toString().includes('Error initializing Lua scripts')) {
// This is a non-recoverable error
// Happens when worker starts and Redis is unavailable
// Even if Redis comes back online, worker will be zombie
this.logger.error('Error initializing worker.');
process.exit(2);
} else {
this.logger.error('Error from queue: ', error);
throw error;
}
});
if (config.getEnv('queue.health.active')) {
const port = config.getEnv('queue.health.port');
const app = express();
app.disable('x-powered-by');
const server = http.createServer(app);
app.get(
'/healthz',
// eslint-disable-next-line consistent-return
async (req: express.Request, res: express.Response) => {
LoggerProxy.debug('Health check started!');
const connection = Db.getConnection();
try {
if (!connection.isInitialized) {
// Connection is not active
throw new Error('No active database connection!');
}
// DB ping
await connection.query('SELECT 1');
} catch (e) {
LoggerProxy.error('No Database connection!', e as Error);
const error = new ResponseHelper.ServiceUnavailableError('No Database connection!');
return ResponseHelper.sendErrorResponse(res, error);
}
// Just to be complete, generally will the worker stop automatically
// if it loses the connection to redis
try {
// Redis ping
await Worker.jobQueue.client.ping();
} catch (e) {
LoggerProxy.error('No Redis connection!', e as Error);
const error = new ResponseHelper.ServiceUnavailableError('No Redis connection!');
return ResponseHelper.sendErrorResponse(res, error);
}
// Everything fine
const responseData = {
status: 'ok',
};
LoggerProxy.debug('Health check completed successfully!');
ResponseHelper.sendSuccessResponse(res, responseData, true, 200);
},
);
server.listen(port, () => {
this.logger.info(`\nn8n worker health check via, port ${port}`);
});
server.on('error', (error: Error & { code: string }) => {
if (error.code === 'EADDRINUSE') {
this.logger.error(
`n8n's port ${port} is already in use. Do you have the n8n main process running on that port?`,
);
process.exit(1);
}
});
}
}
async catch(error: Error) {
await this.exitWithCrash('Worker exiting due to an error.', error);
}
}