mirror of
https://github.com/n8n-io/n8n.git
synced 2026-09-21 12:51:16 +08:00
feat(core): Add cluster check reconciliation cycle (no-changelog) (#28936)
This commit is contained in:
Andreas Fitzek
@@ -42,9 +42,32 @@ const ClusterVersionMismatchSchema = z.object({
|
||||
// Version mismatch type (for REST API response)
|
||||
export type ClusterVersionMismatch = z.infer<typeof ClusterVersionMismatchSchema>;
|
||||
|
||||
const ClusterCheckWarningSchema = z.object({
|
||||
check: z.string(),
|
||||
code: z.string(),
|
||||
message: z.string(),
|
||||
severity: z.enum(['info', 'warning', 'error']).optional(),
|
||||
context: z.record(z.string(), z.unknown()).optional(),
|
||||
});
|
||||
|
||||
const ClusterCheckResultSchema = z.object({
|
||||
check: z.string(),
|
||||
executedAt: z.number(),
|
||||
warnings: z.array(ClusterCheckWarningSchema),
|
||||
status: z.enum(['succeeded', 'failed']),
|
||||
});
|
||||
|
||||
export type ClusterCheckResult = z.infer<typeof ClusterCheckResultSchema>;
|
||||
|
||||
// Single warning raised by a cluster check (for REST API response)
|
||||
export type ClusterCheckWarning = z.infer<typeof ClusterCheckWarningSchema>;
|
||||
|
||||
const ClusterCheckSummarySchema = z.record(z.string(), ClusterCheckResultSchema);
|
||||
export type ClusterCheckSummary = z.infer<typeof ClusterCheckSummarySchema>;
|
||||
|
||||
const ClusterInfoResponseSchema = z.object({
|
||||
instances: z.array(instanceRegistrationSchema),
|
||||
versionMismatch: ClusterVersionMismatchSchema.nullable(),
|
||||
checks: ClusterCheckSummarySchema,
|
||||
});
|
||||
|
||||
// REST API response type
|
||||
|
||||
@@ -0,0 +1,340 @@
|
||||
import type { InstanceRegistration } from '@n8n/api-types';
|
||||
import type { Logger } from '@n8n/backend-common';
|
||||
import type {
|
||||
ClusterCheckContext,
|
||||
ClusterCheckMetadata,
|
||||
ClusterCheckResult,
|
||||
IClusterCheck,
|
||||
} from '@n8n/decorators';
|
||||
import { Container } from '@n8n/di';
|
||||
import { mock } from 'jest-mock-extended';
|
||||
import type { InstanceSettings } from 'n8n-core';
|
||||
|
||||
import type { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
|
||||
import type { Push } from '@/push';
|
||||
|
||||
import { CheckService, computeDiff } from '../checks/check.service';
|
||||
import type { InstanceRegistryService } from '../instance-registry.service';
|
||||
import { REGISTRY_CONSTANTS } from '../instance-registry.types';
|
||||
|
||||
const makeLogger = () => {
|
||||
const logger = mock<Logger>();
|
||||
logger.scoped.mockReturnValue(logger);
|
||||
return logger;
|
||||
};
|
||||
|
||||
const makeInstance = (override: Partial<InstanceRegistration> = {}): InstanceRegistration => ({
|
||||
schemaVersion: 1 as const,
|
||||
instanceKey: 'key',
|
||||
hostId: 'host',
|
||||
instanceType: 'main',
|
||||
instanceRole: 'follower',
|
||||
version: '1.0.0',
|
||||
registeredAt: 0,
|
||||
lastSeen: 0,
|
||||
...override,
|
||||
});
|
||||
|
||||
const namedClass = (name: string) => {
|
||||
class Anon {}
|
||||
Object.defineProperty(Anon, 'name', { value: name });
|
||||
return Anon as unknown as new () => IClusterCheck;
|
||||
};
|
||||
|
||||
describe('CheckService', () => {
|
||||
let logger: ReturnType<typeof makeLogger>;
|
||||
let instanceSettings: InstanceSettings;
|
||||
let registryService: jest.Mocked<InstanceRegistryService>;
|
||||
let clusterCheckMetadata: jest.Mocked<ClusterCheckMetadata>;
|
||||
let messageEventBus: jest.Mocked<MessageEventBus>;
|
||||
let push: jest.Mocked<Push>;
|
||||
let containerGet: jest.SpyInstance;
|
||||
let service: CheckService | undefined;
|
||||
|
||||
const buildService = () =>
|
||||
new CheckService(
|
||||
logger,
|
||||
instanceSettings,
|
||||
registryService,
|
||||
clusterCheckMetadata,
|
||||
messageEventBus,
|
||||
push,
|
||||
);
|
||||
|
||||
beforeEach(() => {
|
||||
jest.useFakeTimers();
|
||||
logger = makeLogger();
|
||||
instanceSettings = mock<InstanceSettings>({ isLeader: false });
|
||||
registryService = mock<InstanceRegistryService>();
|
||||
clusterCheckMetadata = mock<ClusterCheckMetadata>();
|
||||
messageEventBus = mock<MessageEventBus>();
|
||||
push = mock<Push>();
|
||||
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([]);
|
||||
registryService.getAllInstances.mockResolvedValue([]);
|
||||
registryService.getLastKnownState.mockResolvedValue(new Map());
|
||||
registryService.saveLastKnownState.mockResolvedValue();
|
||||
messageEventBus.sendAuditEvent.mockResolvedValue(undefined);
|
||||
|
||||
containerGet = jest.spyOn(Container, 'get');
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
service?.shutdown();
|
||||
service = undefined;
|
||||
containerGet.mockRestore();
|
||||
jest.useRealTimers();
|
||||
});
|
||||
|
||||
it('discovers checks via metadata and DI, skipping failures', () => {
|
||||
const WorkingCheck = namedClass('WorkingCheck');
|
||||
const FailingCheck = namedClass('FailingCheck');
|
||||
const workingInstance: IClusterCheck = {
|
||||
checkDescription: { name: 'cluster.working' },
|
||||
async run() {
|
||||
return {};
|
||||
},
|
||||
};
|
||||
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([FailingCheck, WorkingCheck]);
|
||||
containerGet.mockImplementation((cls: unknown) => {
|
||||
if (cls === WorkingCheck) return workingInstance;
|
||||
throw new Error('not registered');
|
||||
});
|
||||
|
||||
service = buildService();
|
||||
service.init();
|
||||
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
'Failed to instantiate cluster check "FailingCheck"',
|
||||
expect.objectContaining({ error: expect.any(Error) }),
|
||||
);
|
||||
expect(logger.info).toHaveBeenCalledWith('Discovered 1 cluster checks', {
|
||||
names: ['cluster.working'],
|
||||
});
|
||||
});
|
||||
|
||||
it('runs reconcile immediately on takeover, again every 180s, and stops on stepdown', async () => {
|
||||
const TickCheck = namedClass('TickCheck');
|
||||
const runMock = jest
|
||||
.fn<Promise<ClusterCheckResult>, [ClusterCheckContext]>()
|
||||
.mockResolvedValue({});
|
||||
const tickInstance: IClusterCheck = {
|
||||
checkDescription: { name: 'cluster.tick' },
|
||||
run: runMock,
|
||||
};
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([TickCheck]);
|
||||
containerGet.mockReturnValue(tickInstance);
|
||||
|
||||
service = buildService();
|
||||
service.init();
|
||||
expect(runMock).not.toHaveBeenCalled();
|
||||
|
||||
service.startReconciliation();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(runMock).toHaveBeenCalledTimes(1);
|
||||
|
||||
await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS);
|
||||
expect(runMock).toHaveBeenCalledTimes(2);
|
||||
|
||||
service.stopReconciliation();
|
||||
await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS * 2);
|
||||
expect(runMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('does not reconcile when not leader, nor after shutdown', async () => {
|
||||
const NoOp = namedClass('NoOp');
|
||||
const runMock = jest
|
||||
.fn<Promise<ClusterCheckResult>, [ClusterCheckContext]>()
|
||||
.mockResolvedValue({});
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([NoOp]);
|
||||
containerGet.mockReturnValue({
|
||||
checkDescription: { name: 'cluster.noop' },
|
||||
run: runMock,
|
||||
});
|
||||
|
||||
Object.assign(instanceSettings, { isLeader: false });
|
||||
service = buildService();
|
||||
service.init();
|
||||
await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS * 2);
|
||||
expect(runMock).not.toHaveBeenCalled();
|
||||
|
||||
service.shutdown();
|
||||
service.startReconciliation();
|
||||
await jest.advanceTimersByTimeAsync(REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS * 2);
|
||||
expect(runMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('reconcile forwards warnings/audit/push from runChecks and saves current state', async () => {
|
||||
const WorkingCheck = namedClass('WorkingCheck');
|
||||
const workingRun = jest
|
||||
.fn<Promise<ClusterCheckResult>, [ClusterCheckContext]>()
|
||||
.mockResolvedValue({
|
||||
warnings: [{ code: 'cluster.w', message: 'warn msg', severity: 'warning' }],
|
||||
auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: { a: 1 } }],
|
||||
pushNotifications: [{ type: 'cluster-foo', data: { b: 2 } }],
|
||||
});
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([WorkingCheck]);
|
||||
containerGet.mockReturnValue({
|
||||
checkDescription: { name: 'cluster.work' },
|
||||
run: workingRun,
|
||||
});
|
||||
|
||||
const inst = makeInstance({ instanceKey: 'k1' });
|
||||
registryService.getAllInstances.mockResolvedValue([inst]);
|
||||
|
||||
service = buildService();
|
||||
service.init();
|
||||
service.startReconciliation();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
expect(logger.warn).toHaveBeenCalledWith(
|
||||
'Cluster check warning',
|
||||
expect.objectContaining({ check: 'cluster.work', code: 'cluster.w' }),
|
||||
);
|
||||
expect(messageEventBus.sendAuditEvent).toHaveBeenCalledWith({
|
||||
eventName: 'n8n.audit.cluster.foo',
|
||||
payload: { a: 1 },
|
||||
});
|
||||
expect(push.broadcast).toHaveBeenCalledWith({ type: 'cluster-foo', data: { b: 2 } });
|
||||
expect(registryService.saveLastKnownState).toHaveBeenCalledWith(new Map([['k1', inst]]));
|
||||
});
|
||||
|
||||
describe('runChecks', () => {
|
||||
it('returns results for succeeded checks, failed markers for thrown checks, and no side effects', async () => {
|
||||
const WorkingCheck = namedClass('WorkingCheck');
|
||||
const FailingCheck = namedClass('FailingCheck');
|
||||
const workingInstance: IClusterCheck = {
|
||||
checkDescription: { name: 'cluster.work', displayName: 'Work Check' },
|
||||
run: jest.fn<Promise<ClusterCheckResult>, [ClusterCheckContext]>().mockResolvedValue({
|
||||
warnings: [{ code: 'cluster.w', message: 'warn msg', severity: 'warning' }],
|
||||
auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: { a: 1 } }],
|
||||
pushNotifications: [{ type: 'cluster-foo', data: { b: 2 } }],
|
||||
}),
|
||||
};
|
||||
const failingInstance: IClusterCheck = {
|
||||
checkDescription: { name: 'cluster.fail', displayName: 'Fail Check' },
|
||||
run: jest.fn().mockRejectedValue(new Error('boom')),
|
||||
};
|
||||
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([WorkingCheck, FailingCheck]);
|
||||
containerGet.mockImplementation((cls: unknown) => {
|
||||
if (cls === WorkingCheck) return workingInstance;
|
||||
if (cls === FailingCheck) return failingInstance;
|
||||
throw new Error('unexpected class');
|
||||
});
|
||||
|
||||
const inst = makeInstance({ instanceKey: 'k1' });
|
||||
registryService.getAllInstances.mockResolvedValue([inst]);
|
||||
|
||||
service = buildService();
|
||||
service.init();
|
||||
|
||||
const { currentState, results } = await service.runChecks();
|
||||
|
||||
expect(currentState).toEqual(new Map([['k1', inst]]));
|
||||
expect(results).toEqual([
|
||||
{
|
||||
checkName: 'cluster.work',
|
||||
checkDisplayName: 'Work Check',
|
||||
result: {
|
||||
warnings: [{ code: 'cluster.w', message: 'warn msg', severity: 'warning' }],
|
||||
auditEvents: [{ eventName: 'n8n.audit.cluster.foo', payload: { a: 1 } }],
|
||||
pushNotifications: [{ type: 'cluster-foo', data: { b: 2 } }],
|
||||
},
|
||||
},
|
||||
{
|
||||
checkName: 'cluster.fail',
|
||||
checkDisplayName: 'Fail Check',
|
||||
failed: true,
|
||||
},
|
||||
]);
|
||||
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
'Cluster check failed',
|
||||
expect.objectContaining({ checkName: 'cluster.fail', error: expect.any(Error) }),
|
||||
);
|
||||
expect(messageEventBus.sendAuditEvent).not.toHaveBeenCalled();
|
||||
expect(push.broadcast).not.toHaveBeenCalled();
|
||||
expect(registryService.saveLastKnownState).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('passes diff context to checks; short-circuits without I/O when no checks are registered', async () => {
|
||||
const DiffCheck = namedClass('DiffCheck');
|
||||
const runMock = jest
|
||||
.fn<Promise<ClusterCheckResult>, [ClusterCheckContext]>()
|
||||
.mockResolvedValue({});
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([DiffCheck]);
|
||||
containerGet.mockReturnValue({
|
||||
checkDescription: { name: 'cluster.diff' },
|
||||
run: runMock,
|
||||
});
|
||||
|
||||
const prev = makeInstance({ instanceKey: 'old' });
|
||||
const curr = makeInstance({ instanceKey: 'new' });
|
||||
registryService.getLastKnownState.mockResolvedValue(new Map([['old', prev]]));
|
||||
registryService.getAllInstances.mockResolvedValue([curr]);
|
||||
|
||||
service = buildService();
|
||||
service.init();
|
||||
await service.runChecks();
|
||||
|
||||
const context = runMock.mock.calls[0][0];
|
||||
expect(context.currentState).toEqual(new Map([['new', curr]]));
|
||||
expect(context.previousState).toEqual(new Map([['old', prev]]));
|
||||
expect(context.diff.added.map((x) => x.instanceKey)).toEqual(['new']);
|
||||
expect(context.diff.removed.map((x) => x.instanceKey)).toEqual(['old']);
|
||||
|
||||
service.shutdown();
|
||||
service = undefined;
|
||||
registryService.getAllInstances.mockClear();
|
||||
registryService.getLastKnownState.mockClear();
|
||||
clusterCheckMetadata.getClasses.mockReturnValue([]);
|
||||
|
||||
service = buildService();
|
||||
service.init();
|
||||
const empty = await service.runChecks();
|
||||
|
||||
expect(empty).toEqual({ currentState: new Map(), results: [] });
|
||||
expect(registryService.getAllInstances).not.toHaveBeenCalled();
|
||||
expect(registryService.getLastKnownState).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('computeDiff', () => {
|
||||
const i = (key: string, extra: Partial<InstanceRegistration> = {}): InstanceRegistration =>
|
||||
makeInstance({ instanceKey: key, ...extra });
|
||||
|
||||
it.each([
|
||||
{
|
||||
name: 'reports added when a key appears only in current',
|
||||
previous: new Map<string, InstanceRegistration>(),
|
||||
current: new Map([['a', i('a')]]),
|
||||
expected: { added: ['a'], removed: [], changed: [] },
|
||||
},
|
||||
{
|
||||
name: 'reports removed when a key is missing from current',
|
||||
previous: new Map([['a', i('a')]]),
|
||||
current: new Map<string, InstanceRegistration>(),
|
||||
expected: { added: [], removed: ['a'], changed: [] },
|
||||
},
|
||||
{
|
||||
name: 'reports changed on meaningful diff; lastSeen-only drift is ignored',
|
||||
previous: new Map([
|
||||
['a', i('a', { version: '1.0.0', lastSeen: 0 })],
|
||||
['b', i('b', { lastSeen: 0 })],
|
||||
]),
|
||||
current: new Map([
|
||||
['a', i('a', { version: '1.1.0', lastSeen: 1 })],
|
||||
['b', i('b', { lastSeen: 9_999 })],
|
||||
]),
|
||||
expected: { added: [], removed: [], changed: ['a'] },
|
||||
},
|
||||
])('$name', ({ previous, current, expected }) => {
|
||||
const diff = computeDiff(previous, current);
|
||||
expect(diff.added.map((x) => x.instanceKey)).toEqual(expected.added);
|
||||
expect(diff.removed.map((x) => x.instanceKey)).toEqual(expected.removed);
|
||||
expect(diff.changed.map((x) => x.current.instanceKey)).toEqual(expected.changed);
|
||||
});
|
||||
});
|
||||
+66
-6
@@ -1,6 +1,7 @@
|
||||
import type { InstanceRegistration } from '@n8n/api-types';
|
||||
import { mock } from 'jest-mock-extended';
|
||||
|
||||
import type { CheckService } from '../checks/check.service';
|
||||
import { InstanceRegistryController } from '../instance-registry.controller';
|
||||
import type { InstanceRegistryService } from '../instance-registry.service';
|
||||
|
||||
@@ -19,14 +20,17 @@ const makeRegistration = (overrides: Partial<InstanceRegistration> = {}): Instan
|
||||
describe('InstanceRegistryController', () => {
|
||||
let controller: InstanceRegistryController;
|
||||
let service: jest.Mocked<InstanceRegistryService>;
|
||||
let checkService: jest.Mocked<CheckService>;
|
||||
|
||||
beforeEach(() => {
|
||||
service = mock<InstanceRegistryService>();
|
||||
controller = new InstanceRegistryController(service);
|
||||
checkService = mock<CheckService>();
|
||||
checkService.runChecks.mockResolvedValue({ currentState: new Map(), results: [] });
|
||||
controller = new InstanceRegistryController(service, checkService);
|
||||
});
|
||||
|
||||
describe('getClusterInfo', () => {
|
||||
it('should return instances from the service', async () => {
|
||||
it('returns instances and an empty check summary when no checks produced results', async () => {
|
||||
const instances = [
|
||||
makeRegistration({ instanceKey: 'key-1', hostId: 'host-1' }),
|
||||
makeRegistration({ instanceKey: 'key-2', hostId: 'host-2', instanceType: 'worker' }),
|
||||
@@ -36,16 +40,72 @@ describe('InstanceRegistryController', () => {
|
||||
const result = await controller.getClusterInfo();
|
||||
|
||||
expect(result.instances).toEqual(instances);
|
||||
expect(result.versionMismatch).toBeNull();
|
||||
expect(result.checks).toEqual({});
|
||||
});
|
||||
|
||||
it('should return empty instances when no instances are registered', async () => {
|
||||
it('builds a per-check summary from runChecks results (succeeded, warned, and execution-failed)', async () => {
|
||||
service.getAllInstances.mockResolvedValue([]);
|
||||
checkService.runChecks.mockResolvedValue({
|
||||
currentState: new Map(),
|
||||
results: [
|
||||
{
|
||||
checkName: 'cluster.versionMismatch',
|
||||
result: {
|
||||
warnings: [
|
||||
{
|
||||
code: 'cluster.versionMismatch',
|
||||
message: 'Detected 2 versions',
|
||||
severity: 'warning',
|
||||
context: { versions: ['1.0.0', '1.1.0'] },
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
{
|
||||
checkName: 'cluster.quiet',
|
||||
result: {},
|
||||
},
|
||||
{
|
||||
checkName: 'cluster.broken',
|
||||
failed: true,
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
const result = await controller.getClusterInfo();
|
||||
|
||||
expect(result.instances).toEqual([]);
|
||||
expect(result.versionMismatch).toBeNull();
|
||||
expect(result.checks['cluster.versionMismatch']).toMatchObject({
|
||||
check: 'cluster.versionMismatch',
|
||||
status: 'failed',
|
||||
warnings: [
|
||||
{
|
||||
check: 'cluster.versionMismatch',
|
||||
code: 'cluster.versionMismatch',
|
||||
message: 'Detected 2 versions',
|
||||
severity: 'warning',
|
||||
context: { versions: ['1.0.0', '1.1.0'] },
|
||||
},
|
||||
],
|
||||
});
|
||||
expect(typeof result.checks['cluster.versionMismatch'].executedAt).toBe('number');
|
||||
|
||||
expect(result.checks['cluster.quiet']).toMatchObject({
|
||||
check: 'cluster.quiet',
|
||||
status: 'succeeded',
|
||||
warnings: [],
|
||||
});
|
||||
|
||||
expect(result.checks['cluster.broken']).toMatchObject({
|
||||
check: 'cluster.broken',
|
||||
status: 'failed',
|
||||
warnings: [
|
||||
{
|
||||
check: 'cluster.broken',
|
||||
code: 'cluster.check-execution-failed',
|
||||
severity: 'warning',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
import type { InstanceRegistration } from '@n8n/api-types';
|
||||
import type { ClusterCheckContext, ClusterStateDiff } from '@n8n/decorators';
|
||||
|
||||
import { VersionMismatchCheck } from '../checks/version-mismatch.check';
|
||||
|
||||
const makeInstance = (override: Partial<InstanceRegistration> = {}): InstanceRegistration => ({
|
||||
schemaVersion: 1 as const,
|
||||
instanceKey: 'key',
|
||||
hostId: 'host',
|
||||
instanceType: 'main',
|
||||
instanceRole: 'follower',
|
||||
version: '1.0.0',
|
||||
registeredAt: 0,
|
||||
lastSeen: 0,
|
||||
...override,
|
||||
});
|
||||
|
||||
const emptyDiff: ClusterStateDiff = {
|
||||
added: [],
|
||||
removed: [],
|
||||
changed: [],
|
||||
};
|
||||
|
||||
const makeContext = (currentState: Map<string, InstanceRegistration>): ClusterCheckContext => ({
|
||||
currentState,
|
||||
previousState: new Map(),
|
||||
diff: emptyDiff,
|
||||
});
|
||||
|
||||
describe('VersionMismatchCheck', () => {
|
||||
const check = new VersionMismatchCheck();
|
||||
|
||||
it('returns no warnings when currentState is empty', async () => {
|
||||
const result = await check.run(makeContext(new Map()));
|
||||
|
||||
expect(result).toEqual({});
|
||||
});
|
||||
|
||||
it('returns no warnings when a single instance is present', async () => {
|
||||
const result = await check.run(
|
||||
makeContext(new Map([['k1', makeInstance({ instanceKey: 'k1', version: '1.0.0' })]])),
|
||||
);
|
||||
|
||||
expect(result).toEqual({});
|
||||
});
|
||||
|
||||
it('returns no warnings when multiple instances share a version', async () => {
|
||||
const result = await check.run(
|
||||
makeContext(
|
||||
new Map([
|
||||
['k1', makeInstance({ instanceKey: 'k1', version: '1.0.0' })],
|
||||
['k2', makeInstance({ instanceKey: 'k2', version: '1.0.0' })],
|
||||
['k3', makeInstance({ instanceKey: 'k3', version: '1.0.0' })],
|
||||
]),
|
||||
),
|
||||
);
|
||||
|
||||
expect(result).toEqual({});
|
||||
});
|
||||
|
||||
it('emits a version-mismatch warning when distinct versions are present', async () => {
|
||||
const result = await check.run(
|
||||
makeContext(
|
||||
new Map([
|
||||
['k1', makeInstance({ instanceKey: 'k1', version: '1.0.0' })],
|
||||
['k2', makeInstance({ instanceKey: 'k2', version: '1.1.0' })],
|
||||
['k3', makeInstance({ instanceKey: 'k3', version: '1.0.0' })],
|
||||
]),
|
||||
),
|
||||
);
|
||||
|
||||
expect(result.warnings).toHaveLength(1);
|
||||
const [warning] = result.warnings!;
|
||||
expect(warning.code).toBe('cluster.version-mismatch');
|
||||
expect(warning.severity).toBe('error');
|
||||
expect(warning.context).toEqual({ versions: ['1.0.0', '1.1.0'] });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,310 @@
|
||||
import type { InstanceRegistration, PushMessage } from '@n8n/api-types';
|
||||
import { Logger } from '@n8n/backend-common';
|
||||
import type {
|
||||
ClusterCheckAuditEvent,
|
||||
ClusterCheckContext,
|
||||
ClusterCheckPushNotification,
|
||||
ClusterCheckResult,
|
||||
ClusterCheckWarning,
|
||||
ClusterStateDiff,
|
||||
IClusterCheck,
|
||||
} from '@n8n/decorators';
|
||||
import {
|
||||
ClusterCheckMetadata,
|
||||
OnLeaderStepdown,
|
||||
OnLeaderTakeover,
|
||||
OnShutdown,
|
||||
} from '@n8n/decorators';
|
||||
import { Container, Service } from '@n8n/di';
|
||||
import { InstanceSettings } from 'n8n-core';
|
||||
|
||||
import type { EventNamesAuditType } from '@/eventbus/event-message-classes';
|
||||
import type { EventPayloadAudit } from '@/eventbus/event-message-classes/event-message-audit';
|
||||
import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
|
||||
import { Push } from '@/push';
|
||||
|
||||
import { InstanceRegistryService } from '../instance-registry.service';
|
||||
import { REGISTRY_CONSTANTS } from '../instance-registry.types';
|
||||
|
||||
/**
|
||||
* Leader-only service that reconciles cluster state and runs health checks
|
||||
* on a fixed interval. Discovers checks via `ClusterCheckMetadata`, fans them
|
||||
* out with error isolation, and forwards each result into logs, the audit
|
||||
* event bus, and the push channel.
|
||||
*/
|
||||
@Service()
|
||||
export class CheckService {
|
||||
private reconcileController?: AbortController;
|
||||
private reconcileTimer: NodeJS.Timeout | undefined;
|
||||
|
||||
private isShuttingDown = false;
|
||||
|
||||
private readonly checks: IClusterCheck[] = [];
|
||||
|
||||
private readonly logger: Logger;
|
||||
|
||||
constructor(
|
||||
logger: Logger,
|
||||
private readonly instanceSettings: InstanceSettings,
|
||||
private readonly instanceRegistryService: InstanceRegistryService,
|
||||
private readonly clusterCheckMetadata: ClusterCheckMetadata,
|
||||
private readonly messageEventBus: MessageEventBus,
|
||||
private readonly push: Push,
|
||||
) {
|
||||
this.logger = logger.scoped('instance-registry');
|
||||
}
|
||||
|
||||
init() {
|
||||
this.discoverChecks();
|
||||
if (this.instanceSettings.isLeader) this.startReconciliation();
|
||||
}
|
||||
|
||||
@OnLeaderTakeover()
|
||||
startReconciliation() {
|
||||
if (this.isShuttingDown || this.reconcileController) return;
|
||||
this.reconcileController = new AbortController();
|
||||
const { signal } = this.reconcileController;
|
||||
|
||||
void this.runReconcileSafely(signal);
|
||||
this.scheduleNextReconcile(signal);
|
||||
|
||||
this.logger.debug('Cluster check reconciliation scheduled');
|
||||
}
|
||||
|
||||
@OnLeaderStepdown()
|
||||
stopReconciliation() {
|
||||
this.reconcileController?.abort();
|
||||
this.reconcileController = undefined;
|
||||
clearTimeout(this.reconcileTimer);
|
||||
this.reconcileTimer = undefined;
|
||||
}
|
||||
|
||||
@OnShutdown()
|
||||
shutdown() {
|
||||
this.isShuttingDown = true;
|
||||
this.stopReconciliation();
|
||||
}
|
||||
|
||||
private scheduleNextReconcile(signal: AbortSignal) {
|
||||
if (signal.aborted) return;
|
||||
this.reconcileTimer = setTimeout(async () => {
|
||||
await this.runReconcileSafely(signal);
|
||||
this.scheduleNextReconcile(signal);
|
||||
}, REGISTRY_CONSTANTS.RECONCILIATION_INTERVAL_MS);
|
||||
}
|
||||
|
||||
private discoverChecks() {
|
||||
const checkClasses = this.clusterCheckMetadata.getClasses();
|
||||
|
||||
for (const CheckClass of checkClasses) {
|
||||
try {
|
||||
const check = Container.get(CheckClass);
|
||||
this.checks.push(check);
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to instantiate cluster check "${CheckClass.name}"`, {
|
||||
error,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
this.logger.info(`Discovered ${this.checks.length} cluster checks`, {
|
||||
names: this.checks.map((c) => c.checkDescription.name),
|
||||
});
|
||||
}
|
||||
|
||||
private async runReconcileSafely(signal: AbortSignal) {
|
||||
try {
|
||||
await this.reconcile(signal);
|
||||
} catch (error) {
|
||||
this.logger.warn('Reconciliation cycle failed', { error });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Evaluates all registered cluster checks against the current cluster state
|
||||
* and returns their aggregated results. Side-effect free: does not dispatch
|
||||
* warnings/audit events/push notifications and does not persist state.
|
||||
*
|
||||
* Safe to call from any instance (leader or follower), e.g. from the REST
|
||||
* controller serving the cluster overview UI. The leader's scheduled
|
||||
* reconciliation loop is the single writer of `lastKnownState`; this method
|
||||
* only reads it to build the diff context for checks.
|
||||
*/
|
||||
async runChecks(): Promise<{
|
||||
currentState: Map<string, InstanceRegistration>;
|
||||
results: Array<{
|
||||
checkName: string;
|
||||
checkDisplayName?: string;
|
||||
result?: ClusterCheckResult;
|
||||
failed?: true;
|
||||
}>;
|
||||
}> {
|
||||
if (this.checks.length === 0) {
|
||||
return { currentState: new Map(), results: [] };
|
||||
}
|
||||
|
||||
const instances = await this.instanceRegistryService.getAllInstances();
|
||||
const currentState = new Map<string, InstanceRegistration>(
|
||||
instances.map((i) => [i.instanceKey, i]),
|
||||
);
|
||||
|
||||
const previousState = await this.instanceRegistryService.getLastKnownState();
|
||||
const diff = computeDiff(previousState, currentState);
|
||||
|
||||
const context: ClusterCheckContext = { currentState, previousState, diff };
|
||||
|
||||
const settled = await Promise.allSettled(
|
||||
this.checks.map(async (check) => await check.run(context)),
|
||||
);
|
||||
|
||||
const results: Array<{
|
||||
checkName: string;
|
||||
checkDisplayName?: string;
|
||||
result?: ClusterCheckResult;
|
||||
failed?: true;
|
||||
}> = [];
|
||||
for (let i = 0; i < settled.length; i++) {
|
||||
const outcome = settled[i];
|
||||
const check = this.checks[i];
|
||||
const checkResult: {
|
||||
checkName: string;
|
||||
checkDisplayName?: string;
|
||||
result?: ClusterCheckResult;
|
||||
failed?: true;
|
||||
} = {
|
||||
checkName: check.checkDescription.name,
|
||||
checkDisplayName: check.checkDescription.displayName,
|
||||
};
|
||||
if (outcome.status === 'fulfilled') {
|
||||
checkResult.result = outcome.value;
|
||||
} else {
|
||||
this.logger.error('Cluster check failed', {
|
||||
...checkResult,
|
||||
error: outcome.reason,
|
||||
});
|
||||
checkResult.failed = true;
|
||||
}
|
||||
results.push(checkResult);
|
||||
}
|
||||
|
||||
return { currentState, results };
|
||||
}
|
||||
|
||||
private async reconcile(signal: AbortSignal) {
|
||||
if (this.checks.length === 0) return;
|
||||
|
||||
if (signal.aborted) return;
|
||||
|
||||
const { currentState, results } = await this.runChecks();
|
||||
|
||||
if (signal.aborted) return;
|
||||
|
||||
for (const { checkName, result } of results) {
|
||||
this.processResult(checkName, result);
|
||||
}
|
||||
|
||||
try {
|
||||
if (signal.aborted) return;
|
||||
await this.instanceRegistryService.saveLastKnownState(currentState);
|
||||
} catch (error) {
|
||||
this.logger.warn('Failed to persist last known cluster state', { error });
|
||||
}
|
||||
}
|
||||
|
||||
private processResult(checkName: string, result?: ClusterCheckResult) {
|
||||
for (const warning of result?.warnings ?? []) {
|
||||
this.logWarning(checkName, warning);
|
||||
}
|
||||
|
||||
for (const event of result?.auditEvents ?? []) {
|
||||
this.emitAuditEvent(checkName, event);
|
||||
}
|
||||
|
||||
for (const notification of result?.pushNotifications ?? []) {
|
||||
this.broadcastPush(checkName, notification);
|
||||
}
|
||||
}
|
||||
|
||||
private logWarning(checkName: string, warning: ClusterCheckWarning) {
|
||||
const severity = warning.severity ?? 'warning';
|
||||
const method = severity === 'info' ? 'info' : severity === 'error' ? 'error' : 'warn';
|
||||
this.logger[method]('Cluster check warning', {
|
||||
check: checkName,
|
||||
code: warning.code,
|
||||
message: warning.message,
|
||||
context: warning.context,
|
||||
});
|
||||
}
|
||||
|
||||
private emitAuditEvent(checkName: string, event: ClusterCheckAuditEvent) {
|
||||
void this.messageEventBus
|
||||
.sendAuditEvent({
|
||||
eventName: event.eventName as EventNamesAuditType,
|
||||
payload: event.payload as unknown as EventPayloadAudit,
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
this.logger.warn('Failed to emit cluster check audit event', {
|
||||
check: checkName,
|
||||
eventName: event.eventName,
|
||||
error,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private broadcastPush(checkName: string, notification: ClusterCheckPushNotification) {
|
||||
try {
|
||||
this.push.broadcast({
|
||||
type: notification.type,
|
||||
data: notification.data,
|
||||
} as PushMessage);
|
||||
} catch (error) {
|
||||
this.logger.warn('Failed to broadcast cluster check push notification', {
|
||||
check: checkName,
|
||||
type: notification.type,
|
||||
error,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Computes the structured diff between two cluster state snapshots keyed by
|
||||
* `instanceKey`. Exported for direct unit testing.
|
||||
*
|
||||
* Equality for the `changed` bucket ignores `lastSeen` — heartbeats refresh it
|
||||
* every 30s, which would otherwise flag every instance on every cycle.
|
||||
*/
|
||||
export function computeDiff(
|
||||
previousState: ReadonlyMap<string, InstanceRegistration>,
|
||||
currentState: ReadonlyMap<string, InstanceRegistration>,
|
||||
): ClusterStateDiff {
|
||||
const added: InstanceRegistration[] = [];
|
||||
const removed: InstanceRegistration[] = [];
|
||||
const changed: Array<{ previous: InstanceRegistration; current: InstanceRegistration }> = [];
|
||||
|
||||
for (const [key, current] of currentState) {
|
||||
const previous = previousState.get(key);
|
||||
if (!previous) {
|
||||
added.push(current);
|
||||
} else if (!isEquivalent(previous, current)) {
|
||||
changed.push({ previous, current });
|
||||
}
|
||||
}
|
||||
|
||||
for (const [key, previous] of previousState) {
|
||||
if (!currentState.has(key)) removed.push(previous);
|
||||
}
|
||||
|
||||
return { added, removed, changed };
|
||||
}
|
||||
|
||||
function isEquivalent(a: InstanceRegistration, b: InstanceRegistration): boolean {
|
||||
return (
|
||||
a.schemaVersion === b.schemaVersion &&
|
||||
a.instanceKey === b.instanceKey &&
|
||||
a.hostId === b.hostId &&
|
||||
a.instanceType === b.instanceType &&
|
||||
a.instanceRole === b.instanceRole &&
|
||||
a.version === b.version
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
import './version-mismatch.check';
|
||||
@@ -0,0 +1,39 @@
|
||||
import {
|
||||
ClusterCheck,
|
||||
ClusterCheckContext,
|
||||
ClusterCheckResult,
|
||||
IClusterCheck,
|
||||
} from '@n8n/decorators';
|
||||
|
||||
@ClusterCheck()
|
||||
export class VersionMismatchCheck implements IClusterCheck {
|
||||
constructor() {}
|
||||
|
||||
checkDescription = {
|
||||
name: 'version-mismatch',
|
||||
displayName: 'Version mismatch',
|
||||
};
|
||||
|
||||
async run(context: ClusterCheckContext): Promise<ClusterCheckResult> {
|
||||
const allInstanceVersions = [...context.currentState.values()].map((i) => i.version);
|
||||
const versions = [...new Set<string>(allInstanceVersions)];
|
||||
|
||||
if (versions.length <= 1) {
|
||||
// Zero instances or a single version — no mismatch.
|
||||
return {};
|
||||
}
|
||||
|
||||
return {
|
||||
warnings: [
|
||||
{
|
||||
code: 'cluster.version-mismatch',
|
||||
message: `Detected multiple N8N versions in the cluster!: ${versions.join(', ')}`,
|
||||
severity: 'error',
|
||||
context: {
|
||||
versions,
|
||||
},
|
||||
},
|
||||
],
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -1,20 +1,67 @@
|
||||
import type { ClusterInfoResponse } from '@n8n/api-types';
|
||||
import type { ClusterCheckSummary, ClusterInfoResponse } from '@n8n/api-types';
|
||||
import { Get, GlobalScope, RestController } from '@n8n/decorators';
|
||||
|
||||
import { CheckService } from './checks/check.service';
|
||||
import { InstanceRegistryService } from './instance-registry.service';
|
||||
|
||||
@RestController('/instance-registry')
|
||||
export class InstanceRegistryController {
|
||||
constructor(private readonly instanceRegistryService: InstanceRegistryService) {}
|
||||
constructor(
|
||||
private readonly instanceRegistryService: InstanceRegistryService,
|
||||
private readonly checkService: CheckService,
|
||||
) {}
|
||||
|
||||
@Get('/')
|
||||
@GlobalScope('orchestration:read')
|
||||
async getClusterInfo(): Promise<ClusterInfoResponse> {
|
||||
const instances = await this.instanceRegistryService.getAllInstances();
|
||||
const [instances, { results }] = await Promise.all([
|
||||
this.instanceRegistryService.getAllInstances(),
|
||||
this.checkService.runChecks(),
|
||||
]);
|
||||
|
||||
const now = Date.now();
|
||||
|
||||
const checks = results.reduce((acc, cur) => {
|
||||
const { checkName, result, failed } = cur;
|
||||
|
||||
if (!acc[checkName]) {
|
||||
acc[checkName] = {
|
||||
status: 'succeeded',
|
||||
check: cur.checkName,
|
||||
executedAt: now,
|
||||
warnings: [],
|
||||
};
|
||||
}
|
||||
|
||||
const entry = acc[checkName];
|
||||
|
||||
const newWarnings =
|
||||
result?.warnings?.map((w) => ({
|
||||
check: checkName,
|
||||
...w, // Assuming the warning object structure matches
|
||||
})) ?? [];
|
||||
|
||||
if (newWarnings.length > 0) {
|
||||
entry.status = 'failed';
|
||||
entry.warnings.push(...newWarnings);
|
||||
}
|
||||
|
||||
if (failed) {
|
||||
entry.status = 'failed';
|
||||
entry.warnings.push({
|
||||
check: cur.checkName,
|
||||
code: 'cluster.check-execution-failed',
|
||||
message: 'Failed to execute cluster check, please check error logs for details',
|
||||
severity: 'warning',
|
||||
});
|
||||
}
|
||||
|
||||
return acc;
|
||||
}, {} as ClusterCheckSummary);
|
||||
|
||||
return {
|
||||
instances,
|
||||
versionMismatch: null,
|
||||
checks,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -34,6 +34,10 @@ export class InstanceRegistryModule implements ModuleInterface {
|
||||
|
||||
const { StaleMemberCleanupService } = await import('./stale-member-cleanup.service');
|
||||
Container.get(StaleMemberCleanupService).init();
|
||||
|
||||
await import('./checks');
|
||||
const { CheckService } = await import('./checks/check.service');
|
||||
Container.get(CheckService).init();
|
||||
}
|
||||
|
||||
@OnShutdown()
|
||||
|
||||
Reference in New Issue
Block a user