refactor(core): Move instanceType to InstanceSettings (no-changelog) (#10640)
Some checks are pending
Test Master / install-and-build (push) Waiting to run
Test Master / Unit tests (18.x) (push) Blocked by required conditions
Test Master / Unit tests (20.x) (push) Blocked by required conditions
Test Master / Unit tests (22.4) (push) Blocked by required conditions
Test Master / Lint (push) Blocked by required conditions
Test Master / Notify Slack on failure (push) Blocked by required conditions
Benchmark Docker Image CI / build (push) Waiting to run

This commit is contained in:
कारतोफ्फेलस्क्रिप्ट™ 2024-09-16 13:37:14 +02:00 committed by GitHub
parent 50beefb658
commit 25c8a328a8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 85 additions and 89 deletions

View file

@ -1,13 +1,11 @@
import { LicenseManager } from '@n8n_io/license-sdk';
import { mock } from 'jest-mock-extended';
import { InstanceSettings } from 'n8n-core';
import type { InstanceSettings } from 'n8n-core';
import config from '@/config';
import { N8N_VERSION } from '@/constants';
import { License } from '@/license';
import { Logger } from '@/logger';
import { OrchestrationService } from '@/services/orchestration.service';
import { mockInstance } from '@test/mocking';
import type { Logger } from '@/logger';
jest.mock('@n8n_io/license-sdk');
@ -27,9 +25,11 @@ describe('License', () => {
});
let license: License;
const logger = mockInstance(Logger);
const instanceSettings = mockInstance(InstanceSettings, { instanceId: MOCK_INSTANCE_ID });
mockInstance(OrchestrationService);
const logger = mock<Logger>();
const instanceSettings = mock<InstanceSettings>({
instanceId: MOCK_INSTANCE_ID,
instanceType: 'main',
});
beforeEach(async () => {
license = new License(logger, instanceSettings, mock(), mock(), mock());
@ -56,8 +56,14 @@ describe('License', () => {
});
test('initializes license manager for worker', async () => {
license = new License(logger, instanceSettings, mock(), mock(), mock());
await license.init('worker');
license = new License(
logger,
mock<InstanceSettings>({ instanceType: 'worker' }),
mock(),
mock(),
mock(),
);
await license.init();
expect(LicenseManager).toHaveBeenCalledWith({
autoRenewEnabled: false,
autoRenewOffset: MOCK_RENEW_OFFSET,
@ -265,7 +271,7 @@ describe('License', () => {
await license.reinit();
expect(initSpy).toHaveBeenCalledWith('main', true);
expect(initSpy).toHaveBeenCalledWith(true);
expect(LicenseManager.prototype.reset).toHaveBeenCalled();
expect(LicenseManager.prototype.initialize).toHaveBeenCalled();

View file

@ -5,6 +5,7 @@ import { engine as expressHandlebars } from 'express-handlebars';
import { readFile } from 'fs/promises';
import type { Server } from 'http';
import isbot from 'isbot';
import type { InstanceType } from 'n8n-core';
import { Container, Service } from 'typedi';
import config from '@/config';
@ -12,7 +13,6 @@ import { N8N_VERSION, TEMPLATES_DIR, inDevelopment, inTest } from '@/constants';
import * as Db from '@/db';
import { OnShutdown } from '@/decorators/on-shutdown';
import { ExternalHooks } from '@/external-hooks';
import { N8nInstanceType } from '@/interfaces';
import { Logger } from '@/logger';
import { rawBodyReader, bodyParser, corsMiddleware } from '@/middlewares';
import { send, sendErrorResponse } from '@/response-helper';
@ -61,7 +61,7 @@ export abstract class AbstractServer {
readonly uniqueInstanceId: string;
constructor(instanceType: N8nInstanceType = 'main') {
constructor(instanceType: Exclude<InstanceType, 'worker'>) {
this.app = express();
this.app.disable('x-powered-by');

View file

@ -17,7 +17,6 @@ import { TelemetryEventRelay } from '@/events/telemetry-event-relay';
import { initExpressionEvaluator } from '@/expression-evaluator';
import { ExternalHooks } from '@/external-hooks';
import { ExternalSecretsManager } from '@/external-secrets/external-secrets-manager.ee';
import type { N8nInstanceType } from '@/interfaces';
import { License } from '@/license';
import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
import { Logger } from '@/logger';
@ -33,9 +32,7 @@ export abstract class BaseCommand extends Command {
protected nodeTypes: NodeTypes;
protected instanceSettings: InstanceSettings;
private instanceType: N8nInstanceType = 'main';
protected instanceSettings: InstanceSettings = Container.get(InstanceSettings);
queueModeId: string;
@ -62,9 +59,6 @@ export abstract class BaseCommand extends Command {
process.once('SIGTERM', this.onTerminationSignal('SIGTERM'));
process.once('SIGINT', this.onTerminationSignal('SIGINT'));
// Make sure the settings exist
this.instanceSettings = Container.get(InstanceSettings);
this.nodeTypes = Container.get(NodeTypes);
await Container.get(LoadNodesAndCredentials).init();
@ -128,17 +122,13 @@ export abstract class BaseCommand extends Command {
await Container.get(TelemetryEventRelay).init();
}
protected setInstanceType(instanceType: N8nInstanceType) {
this.instanceType = instanceType;
config.set('generic.instanceType', instanceType);
}
protected setInstanceQueueModeId() {
if (config.get('redis.queueModeId')) {
this.queueModeId = config.get('redis.queueModeId');
return;
}
this.queueModeId = generateHostInstanceId(this.instanceType);
// eslint-disable-next-line @typescript-eslint/no-unnecessary-type-assertion
this.queueModeId = generateHostInstanceId(this.instanceSettings.instanceType!);
config.set('redis.queueModeId', this.queueModeId);
}
@ -278,7 +268,7 @@ export abstract class BaseCommand extends Command {
async initLicense(): Promise<void> {
this.license = Container.get(License);
await this.license.init(this.instanceType ?? 'main');
await this.license.init();
const activationKey = config.getEnv('license.activationKey');

View file

@ -69,7 +69,6 @@ export class Start extends BaseCommand {
constructor(argv: string[], cmdConfig: Config) {
super(argv, cmdConfig);
this.setInstanceType('main');
this.setInstanceQueueModeId();
}

View file

@ -25,7 +25,6 @@ export class Webhook extends BaseCommand {
constructor(argv: string[], cmdConfig: Config) {
super(argv, cmdConfig);
this.setInstanceType('webhook');
if (this.queueModeId) {
this.logger.debug(`Webhook Instance queue mode id: ${this.queueModeId}`);
}

View file

@ -78,7 +78,6 @@ export class Worker extends BaseCommand {
);
}
this.setInstanceType('worker');
this.setInstanceQueueModeId();
}

View file

@ -175,12 +175,6 @@ export const schema = {
env: 'GENERIC_TIMEZONE',
},
instanceType: {
doc: 'Type of n8n instance',
format: ['main', 'webhook', 'worker'] as const,
default: 'main',
},
releaseChannel: {
doc: 'N8N release channel',
format: ['stable', 'beta', 'nightly', 'dev'] as const,

View file

@ -1,14 +1,13 @@
import type { InstanceType } from 'n8n-core';
import { ALPHABET } from 'n8n-workflow';
import { customAlphabet } from 'nanoid';
import type { N8nInstanceType } from '@/interfaces';
const nanoid = customAlphabet(ALPHABET, 16);
export function generateNanoId() {
return nanoid();
}
export function generateHostInstanceId(instanceType: N8nInstanceType) {
export function generateHostInstanceId(instanceType: InstanceType) {
return `${instanceType}-${nanoid()}`;
}

View file

@ -422,5 +422,3 @@ export abstract class SecretsProvider {
abstract hasSecret(name: string): boolean;
abstract getSecretNames(): string[];
}
export type N8nInstanceType = 'main' | 'webhook' | 'worker';

View file

@ -17,7 +17,7 @@ import {
SETTINGS_LICENSE_CERT_KEY,
UNLIMITED_LICENSE_QUOTA,
} from './constants';
import type { BooleanLicenseFeature, N8nInstanceType, NumericLicenseFeature } from './interfaces';
import type { BooleanLicenseFeature, NumericLicenseFeature } from './interfaces';
import type { RedisServicePubSubPublisher } from './services/redis/redis-service-pub-sub-publisher';
import { RedisService } from './services/redis.service';
@ -46,8 +46,8 @@ export class License {
/**
* Whether this instance should renew the license - on init and periodically.
*/
private renewalEnabled(instanceType: N8nInstanceType) {
if (instanceType !== 'main') return false;
private renewalEnabled() {
if (this.instanceSettings.instanceType !== 'main') return false;
const autoRenewEnabled = config.getEnv('license.autoRenewEnabled');
@ -63,7 +63,7 @@ export class License {
return autoRenewEnabled;
}
async init(instanceType: N8nInstanceType = 'main', forceRecreate = false) {
async init(forceRecreate = false) {
if (this.manager && !forceRecreate) {
this.logger.warn('License manager already initialized or shutting down');
return;
@ -73,6 +73,7 @@ export class License {
return;
}
const { instanceType } = this.instanceSettings;
const isMainInstance = instanceType === 'main';
const server = config.getEnv('license.serverUrl');
const offlineMode = !isMainInstance;
@ -90,7 +91,7 @@ export class License {
? async () => await this.licenseMetricsService.collectPassthroughData()
: async () => ({});
const renewalEnabled = this.renewalEnabled(instanceType);
const renewalEnabled = this.renewalEnabled();
try {
this.manager = new LicenseManager({
@ -399,7 +400,7 @@ export class License {
async reinit() {
this.manager?.reset();
await this.init('main', true);
await this.init(true);
this.logger.debug('License reinitialized');
}
}

View file

@ -5,7 +5,6 @@ import { InstanceSettings } from 'n8n-core';
import { ApplicationError } from 'n8n-workflow';
import Container from 'typedi';
import config from '@/config';
import type { OrchestrationService } from '@/services/orchestration.service';
import { mockInstance } from '@test/mocking';
@ -70,7 +69,8 @@ describe('ScalingService', () => {
beforeEach(() => {
jest.clearAllMocks();
config.set('generic.instanceType', 'main');
// @ts-expect-error readonly property
instanceSettings.instanceType = 'main';
instanceSettings.markAsLeader();
scalingService = new ScalingService(
@ -128,8 +128,8 @@ describe('ScalingService', () => {
describe('if worker', () => {
it('should set up queue + listeners', async () => {
// @ts-expect-error Private field
scalingService.instanceType = 'worker';
// @ts-expect-error readonly property
instanceSettings.instanceType = 'worker';
await scalingService.setupQueue();
@ -141,8 +141,8 @@ describe('ScalingService', () => {
describe('webhook', () => {
it('should set up a queue + listeners', async () => {
// @ts-expect-error Private field
scalingService.instanceType = 'webhook';
// @ts-expect-error readonly property
instanceSettings.instanceType = 'webhook';
await scalingService.setupQueue();
@ -155,8 +155,8 @@ describe('ScalingService', () => {
describe('setupWorker', () => {
it('should set up a worker with concurrency', async () => {
// @ts-expect-error Private field
scalingService.instanceType = 'worker';
// @ts-expect-error readonly property
instanceSettings.instanceType = 'worker';
await scalingService.setupQueue();
const concurrency = 5;
@ -172,8 +172,8 @@ describe('ScalingService', () => {
});
it('should throw if called before queue is ready', async () => {
// @ts-expect-error Private field
scalingService.instanceType = 'worker';
// @ts-expect-error readonly property
instanceSettings.instanceType = 'worker';
expect(() => scalingService.setupWorker(5)).toThrow();
});

View file

@ -31,8 +31,6 @@ import type {
export class ScalingService {
private queue: JobQueue;
private readonly instanceType = config.getEnv('generic.instanceType');
constructor(
private readonly logger: Logger,
private readonly activeExecutions: ActiveExecutions,
@ -211,9 +209,10 @@ export class ScalingService {
throw error;
});
if (this.instanceType === 'main' || this.instanceType === 'webhook') {
const { instanceType } = this.instanceSettings;
if (instanceType === 'main' || instanceType === 'webhook') {
this.registerMainOrWebhookListeners();
} else if (this.instanceType === 'worker') {
} else if (instanceType === 'worker') {
this.registerWorkerListeners();
}
}
@ -295,7 +294,7 @@ export class ScalingService {
}
private assertWorker() {
if (this.instanceType === 'worker') return;
if (this.instanceSettings.instanceType === 'worker') return;
throw new ApplicationError('This method must be called on a `worker` instance');
}
@ -311,7 +310,7 @@ export class ScalingService {
get isQueueMetricsEnabled() {
return (
this.globalConfig.endpoints.metrics.includeQueueMetrics &&
this.instanceType === 'main' &&
this.instanceSettings.instanceType === 'main' &&
!this.orchestrationService.isMultiMainSetupEnabled
);
}

View file

@ -34,7 +34,6 @@ let queueModeId: string;
function setDefaultConfig() {
config.set('executions.mode', 'queue');
config.set('generic.instanceType', 'main');
}
const workerRestartEventBusResponse: RedisServiceWorkerResponseObject = {
@ -73,6 +72,9 @@ describe('Orchestration Service', () => {
});
setDefaultConfig();
queueModeId = config.get('redis.queueModeId');
// @ts-expect-error readonly property
instanceSettings.instanceType = 'main';
});
beforeEach(() => {

View file

@ -14,7 +14,7 @@ import { RedisService } from './redis.service';
export class OrchestrationService {
constructor(
private readonly logger: Logger,
private readonly instanceSettings: InstanceSettings,
protected readonly instanceSettings: InstanceSettings,
private readonly redisService: RedisService,
readonly multiMainSetup: MultiMainSetup,
) {}
@ -31,7 +31,7 @@ export class OrchestrationService {
return (
config.getEnv('executions.mode') === 'queue' &&
config.getEnv('multiMainSetup.enabled') &&
config.getEnv('generic.instanceType') === 'main' &&
this.instanceSettings.instanceType === 'main' &&
this.isMultiMainSetupLicensed
);
}

View file

@ -1,3 +1,4 @@
import { InstanceSettings } from 'n8n-core';
import { Container } from 'typedi';
import { ActiveWorkflowManager } from '@/active-workflow-manager';
@ -17,7 +18,7 @@ import { debounceMessageReceiver, messageToRedisServiceCommandObject } from '../
// eslint-disable-next-line complexity
export async function handleCommandMessageMain(messageString: string) {
const queueModeId = config.getEnv('redis.queueModeId');
const isMainInstance = config.getEnv('generic.instanceType') === 'main';
const isMainInstance = Container.get(InstanceSettings).instanceType === 'main';
const message = messageToRedisServiceCommandObject(messageString);
const logger = Container.get(Logger);

View file

@ -1,3 +1,4 @@
import { InstanceSettings } from 'n8n-core';
import Container from 'typedi';
import { Logger } from 'winston';
@ -11,7 +12,7 @@ import { messageToRedisServiceCommandObject, debounceMessageReceiver } from '../
export async function handleCommandMessageWebhook(messageString: string) {
const queueModeId = config.getEnv('redis.queueModeId');
const isMainInstance = config.getEnv('generic.instanceType') === 'main';
const isMainInstance = Container.get(InstanceSettings).instanceType === 'main';
const message = messageToRedisServiceCommandObject(messageString);
const logger = Container.get(Logger);

View file

@ -10,7 +10,7 @@ export class OrchestrationWebhookService extends OrchestrationService {
return (
this.isInitialized &&
config.get('executions.mode') === 'queue' &&
config.get('generic.instanceType') === 'webhook'
this.instanceSettings.instanceType === 'webhook'
);
}
}

View file

@ -10,7 +10,7 @@ export class OrchestrationWorkerService extends OrchestrationService {
return (
this.isInitialized &&
config.get('executions.mode') === 'queue' &&
config.get('generic.instanceType') === 'worker'
this.instanceSettings.instanceType === 'worker'
);
}
}

View file

@ -48,19 +48,12 @@ export class PruningService {
}
private isPruningEnabled() {
if (
!config.getEnv('executions.pruneData') ||
inTest ||
config.get('generic.instanceType') !== 'main'
) {
const { instanceType, isFollower } = this.instanceSettings;
if (!config.getEnv('executions.pruneData') || inTest || instanceType !== 'main') {
return false;
}
if (
config.getEnv('multiMainSetup.enabled') &&
config.getEnv('generic.instanceType') === 'main' &&
this.instanceSettings.isFollower
) {
if (config.getEnv('multiMainSetup.enabled') && instanceType === 'main' && isFollower) {
return false;
}

View file

@ -3,7 +3,7 @@
/* eslint-disable @typescript-eslint/no-shadow */
/* eslint-disable @typescript-eslint/no-unsafe-assignment */
import { GlobalConfig } from '@n8n/config';
import { WorkflowExecute } from 'n8n-core';
import { InstanceSettings, WorkflowExecute } from 'n8n-core';
import type {
ExecutionError,
IDeferredPromise,
@ -54,6 +54,7 @@ export class WorkflowRunner {
private readonly nodeTypes: NodeTypes,
private readonly permissionChecker: PermissionChecker,
private readonly eventService: EventService,
private readonly instanceSettings: InstanceSettings,
) {}
/** The process did error */
@ -150,7 +151,7 @@ export class WorkflowRunner {
// since these calls are now done by the worker directly
if (
this.executionsMode !== 'queue' ||
config.getEnv('generic.instanceType') === 'worker' ||
this.instanceSettings.instanceType === 'worker' ||
data.executionMode === 'manual'
) {
const postExecutePromise = this.activeExecutions.getPostExecutePromise(executionId);

View file

@ -1,3 +1,5 @@
process.argv[2] = 'worker';
import { BinaryDataService } from 'n8n-core';
import { Worker } from '@/commands/worker';
@ -27,6 +29,7 @@ const logStreamingEventRelay = mockInstance(LogStreamingEventRelay);
const orchestrationHandlerWorkerService = mockInstance(OrchestrationHandlerWorkerService);
const scalingService = mockInstance(ScalingService);
const orchestrationWorkerService = mockInstance(OrchestrationWorkerService);
const command = setupTestCommand(Worker);
test('worker initializes all its components', async () => {

View file

@ -7,12 +7,12 @@ import {
} from 'n8n-workflow';
import { agent as testAgent } from 'supertest';
import { AbstractServer } from '@/abstract-server';
import type { WorkflowEntity } from '@/databases/entities/workflow-entity';
import { ExternalHooks } from '@/external-hooks';
import { NodeTypes } from '@/node-types';
import { Push } from '@/push';
import { Telemetry } from '@/telemetry';
import { WebhookServer } from '@/webhooks/webhook-server';
import { createUser } from './shared/db/users';
import { createWorkflow } from './shared/db/workflows';
@ -49,7 +49,7 @@ describe('Webhook API', () => {
await initActiveWorkflowManager();
const server = new (class extends AbstractServer {})();
const server = new WebhookServer();
await server.start();
agent = testAgent(server.app);
});
@ -152,7 +152,7 @@ describe('Webhook API', () => {
await initActiveWorkflowManager();
const server = new (class extends AbstractServer {})();
const server = new WebhookServer();
await server.start();
agent = testAgent(server.app);
});

View file

@ -4,12 +4,12 @@ import { agent as testAgent } from 'supertest';
import type SuperAgentTest from 'supertest/lib/agent';
import Container from 'typedi';
import { AbstractServer } from '@/abstract-server';
import { ExternalHooks } from '@/external-hooks';
import { WaitingForms } from '@/waiting-forms';
import { LiveWebhooks } from '@/webhooks/live-webhooks';
import { TestWebhooks } from '@/webhooks/test-webhooks';
import { WaitingWebhooks } from '@/webhooks/waiting-webhooks';
import { WebhookServer } from '@/webhooks/webhook-server';
import type { IWebhookResponseCallbackData } from '@/webhooks/webhook.types';
import { mockInstance } from '@test/mocking';
@ -26,9 +26,9 @@ describe('WebhookServer', () => {
mockInstance(WaitingForms);
beforeAll(async () => {
const server = new (class extends AbstractServer {
testWebhooksEnabled = true;
})();
const server = new WebhookServer();
// @ts-expect-error: testWebhooksEnabled is private
server.testWebhooksEnabled = true;
await server.start();
agent = testAgent(server.app);
});

View file

@ -16,6 +16,8 @@ type Settings = ReadOnlySettings & WritableSettings;
type InstanceRole = 'unset' | 'leader' | 'follower';
export type InstanceType = 'main' | 'webhook' | 'worker';
const inTest = process.env.NODE_ENV === 'test';
@Service()
@ -40,6 +42,15 @@ export class InstanceSettings {
readonly instanceId = this.generateInstanceId();
readonly instanceType: InstanceType;
constructor() {
const command = process.argv[2];
this.instanceType = ['webhook', 'worker'].includes(command)
? (command as InstanceType)
: 'main';
}
/**
* A main is:
* - `unset` during bootup,

View file

@ -10,7 +10,7 @@ export * from './Constants';
export * from './Credentials';
export * from './DirectoryLoader';
export * from './Interfaces';
export { InstanceSettings } from './InstanceSettings';
export { InstanceSettings, InstanceType } from './InstanceSettings';
export * from './NodeExecuteFunctions';
export * from './WorkflowExecute';
export { NodeExecuteFunctions };