feat(core): Add log streaming events for instance reports (backport to release-candidate/2.39.x) (#38483)

Co-authored-by: Dmitrii <dmitrii.kulikov@n8n.io>

packages/cli/src/eventbus/event-message-classes/__tests__/event-names.test.ts

@@ -15,6 +15,7 @@ const KNOWN_EVENT_GROUPS = [
 	'n8n.runner.',
 	'n8n.queue.',
 	'n8n.execution.',
+	'n8n.instanceReporting.',
 ];
 
 describe('log streaming event names', () => {

packages/cli/src/eventbus/event-message-classes/index.ts

@@ -66,6 +66,10 @@ export const eventNamesWorkflow = [
 	'n8n.workflow.cancelled',
 ] as const;
 export const eventNamesGeneric = ['n8n.worker.started', 'n8n.worker.stopped'] as const;
+export const eventNamesInstanceReporting = [
+	'n8n.instanceReporting.success',
+	'n8n.instanceReporting.failed',
+] as const;
 export const eventNamesNode = ['n8n.node.started', 'n8n.node.finished'] as const;
 export const eventNamesExecution = [
 	'n8n.execution.throttled',
@@ -183,6 +187,7 @@ export type EventNamesAuditType = (typeof eventNamesAudit)[number];
 export type EventNamesNodeType = (typeof eventNamesNode)[number];
 export type EventNamesExecutionType = (typeof eventNamesExecution)[number];
 export type EventNamesGenericType = (typeof eventNamesGeneric)[number];
+export type EventNamesInstanceReportingType = (typeof eventNamesInstanceReporting)[number];
 
 export type EventNamesTypes =
 	| EventNamesAuditType
@@ -190,6 +195,7 @@ export type EventNamesTypes =
 	| EventNamesNodeType
 	| EventNamesExecutionType
 	| EventNamesGenericType
+	| EventNamesInstanceReportingType
 	| EventNamesAiNodesType
 	| EventNamesRunnerType
 	| EventNamesQueueType
@@ -201,6 +207,7 @@ export const eventNamesAll = [
 	...eventNamesWorkflow,
 	...eventNamesNode,
 	...eventNamesGeneric,
+	...eventNamesInstanceReporting,
 	...eventNamesAiNodes,
 	...eventNamesRunner,
 	...eventNamesQueue,

packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts

@@ -3358,4 +3358,22 @@ describe('LogStreamingEventRelay', () => {
 			});
 		});
 	});
+
+	describe('instance reporting events', () => {
+		it('should log on instance-report-delivered event', () => {
+			eventService.emit('instance-report-delivered');
+
+			expect(eventBus.send).toHaveBeenCalledWith(
+				expect.objectContaining({ eventName: 'n8n.instanceReporting.success' }),
+			);
+		});
+
+		it('should log on instance-report-failed event', () => {
+			eventService.emit('instance-report-failed');
+
+			expect(eventBus.send).toHaveBeenCalledWith(
+				expect.objectContaining({ eventName: 'n8n.instanceReporting.failed' }),
+			);
+		});
+	});
 });

packages/cli/src/events/maps/relay.event-map.ts

@@ -93,6 +93,12 @@ export type RelayEventMap = {
 		credentialId?: string;
 	};
 
+	// Delivery outcome of an instance usage report. No payload: the event name is
+	// the whole signal a log-streaming consumer needs.
+	'instance-report-delivered': {};
+
+	'instance-report-failed': {};
+
 	// #endregion
 
 	// #region Workflow

packages/cli/src/events/relays/log-streaming.event-relay.ts

@@ -3,6 +3,7 @@ import { Service } from '@n8n/di';
 import { InstanceSettings } from 'n8n-core';
 import type { IWorkflowBase, JsonValue } from 'n8n-workflow';
 
+import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-message-generic';
 import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
 import { EventService } from '@/events/event.service';
 import type { RelayEventMap, UserLike } from '@/events/maps/relay.event-map';
@@ -187,6 +188,8 @@ export class LogStreamingEventRelay extends EventRelay {
 			'mcp-oauth-completed': (event) => this.mcpOauthCompleted(event),
 			'mcp-tool-called': (event) => this.mcpToolCalled(event),
 			'mcp-access-updated': (event) => this.mcpAccessUpdated(event),
+			'instance-report-delivered': () => this.instanceReportDelivered(),
+			'instance-report-failed': () => this.instanceReportFailed(),
 		});
 	}
 
@@ -222,6 +225,16 @@ export class LogStreamingEventRelay extends EventRelay {
 		});
 	}
 
+	private instanceReportDelivered() {
+		void this.eventBus.send(
+			new EventMessageGeneric({ eventName: 'n8n.instanceReporting.success' }),
+		);
+	}
+
+	private instanceReportFailed() {
+		void this.eventBus.send(new EventMessageGeneric({ eventName: 'n8n.instanceReporting.failed' }));
+	}
+
 	@Redactable()
 	private packageImportFailed({ user, ...rest }: RelayEventMap['n8n-package-import-failed']) {
 		void this.eventBus.sendAuditEvent({

packages/cli/src/modules/instance-reporting/__tests__/instance-reporting-scheduler.service.test.ts

@@ -6,6 +6,8 @@ import { UserError } from 'n8n-workflow';
 import type { Mocked } from 'vitest';
 import { mock } from 'vitest-mock-extended';
 
+import type { EventService } from '@/events/event.service';
+
 import type { InstanceMonitoringReportRepository } from '../database/repositories/instance-monitoring-report.repository';
 import { InstanceReportingScheduler } from '../instance-reporting-scheduler.service';
 import type { InstanceReportingSettingsService } from '../instance-reporting-settings.service';
@@ -27,12 +29,19 @@ interface Harness {
 	settingsService: Mocked<InstanceReportingSettingsService>;
 	instanceSettings: Mocked<InstanceSettings>;
 	modulesConfig: Mocked<ModulesConfig>;
+	eventService: Mocked<EventService>;
+	/** Fire the deferred `server-started` listeners (only needed when autoStart is false). */
+	startServer: () => void;
 }
 
 function makeHarness({
 	baseUrl = 'https://example.com',
 	isLeader = true,
 	disabledModules = [] as ModuleName[],
+	// The scheduler arms its first tick on `server-started`. By default the harness
+	// fires that listener the moment it is registered, so `init()` starts reporting
+	// as the tests expect. Pass false to hold it and fire it later via `startServer`.
+	autoStart = true,
 } = {}): Harness {
 	const config = new InstanceReportingConfig();
 	config.instanceReportingBaseUrl = baseUrl;
@@ -54,6 +63,23 @@ function makeHarness({
 	});
 	const modulesConfig = mock<ModulesConfig>({ disabledModules });
 
+	const serverStartedListeners: Array<() => void> = [];
+	const eventService = mock<EventService>();
+	eventService.once.mockImplementation((event, listener) => {
+		if (event === 'server-started') {
+			if (autoStart) (listener as () => void)();
+			else serverStartedListeners.push(listener as () => void);
+		}
+		return eventService;
+	});
+	eventService.off.mockImplementation((event, listener) => {
+		if (event === 'server-started') {
+			const index = serverStartedListeners.indexOf(listener as () => void);
+			if (index !== -1) serverStartedListeners.splice(index, 1);
+		}
+		return eventService;
+	});
+
 	const scheduler = new InstanceReportingScheduler(
 		config,
 		reportingService,
@@ -61,6 +87,7 @@ function makeHarness({
 		settingsService,
 		instanceSettings,
 		modulesConfig,
+		eventService,
 		mockLogger(),
 	);
 
@@ -71,6 +98,8 @@ function makeHarness({
 		settingsService,
 		instanceSettings,
 		modulesConfig,
+		eventService,
+		startServer: () => serverStartedListeners.forEach((listener) => listener()),
 	};
 }
 
@@ -136,6 +165,22 @@ describe('InstanceReportingScheduler', () => {
 
 			expect(reportingService.sendReport).toHaveBeenCalledTimes(3);
 		});
+
+		test('holds the first tick until the server has started', async () => {
+			// A boot catch-up report must not run before the event bus is up, or its
+			// success/failure event is emitted with nowhere to land.
+			vi.setSystemTime(new Date(AFTER_SLOT));
+			const { scheduler, reportingService, startServer } = makeHarness({ autoStart: false });
+
+			await scheduler.init();
+			await settle();
+			expect(reportingService.sendReport).not.toHaveBeenCalled();
+
+			startServer();
+			await settle();
+
+			expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
+		});
 	});
 
 	describe('catch-up', () => {
@@ -194,6 +239,21 @@ describe('InstanceReportingScheduler', () => {
 			expect(reportingService.sendReport).toHaveBeenCalledTimes(1);
 		});
 
+		test('drops the deferred first tick when it steps down before the server starts', async () => {
+			// Leader at boot, but it loses leadership before `server-started`. The
+			// pending listener must be removed, so it neither leaks nor reports, and a
+			// later takeover cannot start it a second time.
+			vi.setSystemTime(new Date(AFTER_SLOT));
+			const { scheduler, reportingService, startServer } = makeHarness({ autoStart: false });
+
+			await scheduler.init();
+			scheduler.stop();
+			startServer();
+			await settle();
+
+			expect(reportingService.sendReport).not.toHaveBeenCalled();
+		});
+
 		test('stops reporting when this main steps down', async () => {
 			const { scheduler, reportingService } = makeHarness();
 

packages/cli/src/modules/instance-reporting/__tests__/instance-reporting.service.test.ts

@@ -10,6 +10,7 @@ import type { IHttpRequestOptions } from 'n8n-workflow';
 import type { Mocked } from 'vitest';
 import { mock } from 'vitest-mock-extended';
 
+import type { EventService } from '@/events/event.service';
 import type { InsightsService } from '@/modules/insights/insights.service';
 import type { OwnershipService } from '@/services/ownership.service';
 
@@ -83,6 +84,7 @@ interface Harness {
 	reportRepository: Mocked<InstanceMonitoringReportRepository>;
 	insightsService: Mocked<InsightsService>;
 	http: HttpRequestClient;
+	eventService: Mocked<EventService>;
 	clientOptions: HttpRequestClientOptions | undefined;
 }
 
@@ -116,6 +118,8 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness {
 		}),
 	});
 
+	const eventService = mock<EventService>();
+
 	const service = new InstanceReportingService(
 		config,
 		reportRepository,
@@ -124,10 +128,11 @@ function makeHarness(config: InstanceReportingConfig = makeConfig()): Harness {
 		ownershipService,
 		licenseMetricsRepository,
 		mockLogger(),
+		eventService,
 		outboundHttp,
 	);
 
-	return { service, reportRepository, insightsService, http, clientOptions };
+	return { service, reportRepository, insightsService, http, eventService, clientOptions };
 }
 
 describe('InstanceReportingService', () => {
@@ -317,6 +322,25 @@ describe('InstanceReportingService', () => {
 			expect(reportRepository.markDelivered).not.toHaveBeenCalled();
 		});
 
+		test('emits a delivered event once the report is delivered', async () => {
+			const { service, eventService } = makeHarness();
+
+			await service.sendReport();
+
+			expect(eventService.emit).toHaveBeenCalledWith('instance-report-delivered');
+			expect(eventService.emit).not.toHaveBeenCalledWith('instance-report-failed');
+		});
+
+		test('emits a failed event when a delivery attempt fails', async () => {
+			const { service, eventService, http } = makeHarness();
+			vi.mocked(http.request).mockRejectedValue(new Error('Network error'));
+
+			await expect(service.sendReport()).rejects.toThrow('Network error');
+
+			expect(eventService.emit).toHaveBeenCalledWith('instance-report-failed');
+			expect(eventService.emit).not.toHaveBeenCalledWith('instance-report-delivered');
+		});
+
 		test('marks a report the receiver already holds as delivered, without retrying', async () => {
 			const { service, reportRepository, http } = makeHarness();
 			vi.mocked(http.request).mockResolvedValue({ statusCode: 409, body: '', headers: {} });

packages/cli/src/modules/instance-reporting/instance-reporting-scheduler.service.ts

@@ -11,6 +11,8 @@ import { InstanceReportingSettingsService } from './instance-reporting-settings.
 import { InstanceReportingConfig } from './instance-reporting.config';
 import { InstanceReportingService, RETRY_DELAY_MS } from './instance-reporting.service';
 
+import { EventService } from '@/events/event.service';
+
 const MINUTES_PER_DAY = 24 * 60;
 
 /**
@@ -38,6 +40,8 @@ export class InstanceReportingScheduler {
 
 	private isShuttingDown = false;
 
+	private readonly onServerStarted = () => this.start();
+
 	constructor(
 		private readonly config: InstanceReportingConfig,
 		private readonly reportingService: InstanceReportingService,
@@ -45,6 +49,7 @@ export class InstanceReportingScheduler {
 		private readonly settingsService: InstanceReportingSettingsService,
 		private readonly instanceSettings: InstanceSettings,
 		private readonly modulesConfig: ModulesConfig,
+		private readonly eventService: EventService,
 		private readonly logger: Logger,
 	) {
 		this.logger = this.logger.scoped('instance-reporting');
@@ -74,7 +79,11 @@ export class InstanceReportingScheduler {
 			return;
 		}
 
-		if (this.instanceSettings.isLeader) this.start();
+		// Defer the first tick until the server has finished starting. A boot
+		// catch-up report can send right away, but log streaming module was not properly initialized
+		if (this.instanceSettings.isLeader) {
+			this.eventService.once('server-started', this.onServerStarted);
+		}
 	}
 
 	get isEnabled(): boolean {
@@ -101,6 +110,8 @@ export class InstanceReportingScheduler {
 
 	@OnLeaderStepdown()
 	stop(): void {
+		this.eventService.off('server-started', this.onServerStarted);
+
 		if (this.timeout === undefined) return;
 
 		clearTimeout(this.timeout);

packages/cli/src/modules/instance-reporting/instance-reporting.service.ts

@@ -8,6 +8,7 @@ import { InstanceSettings } from 'n8n-core';
 import { OperationalError } from 'n8n-workflow';
 
 import { N8N_VERSION } from '@/constants';
+import { EventService } from '@/events/event.service';
 import { InsightsService } from '@/modules/insights/insights.service';
 import { OwnershipService } from '@/services/ownership.service';
 
@@ -55,6 +56,7 @@ export class InstanceReportingService {
 		private readonly ownershipService: OwnershipService,
 		private readonly licenseMetricsRepository: LicenseMetricsRepository,
 		private readonly logger: Logger,
+		private readonly eventService: EventService,
 		outboundHttp: OutboundHttp,
 	) {
 		this.logger = this.logger.scoped('instance-reporting');
@@ -153,8 +155,10 @@ export class InstanceReportingService {
 					`Instance report was rejected with status ${response.statusCode}`,
 				);
 			}
+			this.eventService.emit('instance-report-delivered');
 		} catch (error) {
 			const message = error instanceof Error ? error.message : String(error);
+			this.eventService.emit('instance-report-failed');
 			await this.reportRepository.recordFailure(report.id, message, new Date());
 
 			// `recordFailure` incremented the count, so the in-memory row is one behind.

packages/frontend/@n8n/i18n/src/locales/en.json

@@ -3840,6 +3840,8 @@
 	"settings.log-streaming.eventGroup.n8n.runner.info": "Will send an event when a Code node execution is requested from a task runner, and when a response is received from the runner with the result.",
 	"settings.log-streaming.eventGroup.n8n.queue": "Queue events",
 	"settings.log-streaming.eventGroup.n8n.queue.info": "Will send an event when a queue-related event occurs, e.g. enqueuing, dequeueing, completion, failure, or stalling.",
+	"settings.log-streaming.eventGroup.n8n.instanceReporting": "Instance Reporting",
+	"settings.log-streaming.eventGroup.n8n.instanceReporting.info": "Will send an event each time an instance usage report to the central monitoring service succeeds or fails.",
 	"settings.log-streaming.eventName.n8n.audit.mcp.oauth.completed": "MCP OAuth completed",
 	"settings.log-streaming.eventName.n8n.audit.mcp.tool.called": "MCP tool called",
 	"settings.log-streaming.eventName.n8n.audit.mcp.access.updated": "MCP access updated",

packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.test.ts

@@ -98,4 +98,28 @@ describe('LogStreamingStore', () => {
 			expect(nodeEventGroup!.indeterminate).toBe(false);
 		});
 	});
+
+	describe('event group boundaries', () => {
+		it('should not put a longer group under one that only shares leading characters', () => {
+			logStreamingStore.addEventName('n8n.foo.created');
+			logStreamingStore.addEventName('n8n.foobar.created');
+
+			logStreamingStore.addDestination({
+				id: 'boundaryDestination',
+				label: 'Boundary Destination',
+				enabled: true,
+				subscribedEvents: [],
+				anonymizeAuditMessages: false,
+			});
+
+			const groups = logStreamingStore.items.boundaryDestination.eventGroups;
+			const fooGroup = groups.find((group) => group.name === 'n8n.foo');
+			const foobarGroup = groups.find((group) => group.name === 'n8n.foobar');
+
+			// `n8n.foobar` starts with `n8n.foo`, but the two are distinct groups on a
+			// segment boundary: neither must swallow the other's events.
+			expect(fooGroup!.children.map((c) => c.name)).toEqual(['n8n.foo.created']);
+			expect(foobarGroup!.children.map((c) => c.name)).toEqual(['n8n.foobar.created']);
+		});
+	});
 });

packages/frontend/editor-ui/src/features/integrations/logStreaming.ee/logStreaming.store.ts

@@ -78,7 +78,7 @@ const eventGroupsFromStringList = (
 			selected: selectionList.has(group),
 			indeterminate: false,
 		};
-		const eventsOfGroup = eventNameArray.filter((e) => e.startsWith(group));
+		const eventsOfGroup = eventNameArray.filter((e) => e === group || e.startsWith(`${group}.`));
 		for (const event of eventsOfGroup) {
 			if (!collection.selected && selectionList.has(event)) {
 				collection.indeterminate = true;