refactor(core): Move the telemetry pulse packet to a system task (no-changelog) (#39304)

packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts

@@ -6,14 +6,20 @@ import { ActivityPruningTask } from '@/services/pruning/activity-pruning.task';
 import { ExecutionPruningSoftDeleteTask } from '@/services/pruning/execution-pruning-soft-delete.task';
 import { WorkflowHistoryCompactionOptimizeTask } from '@/services/pruning/workflow-history-compaction-optimize.task';
 import { WorkflowHistoryCompactionTrimTask } from '@/services/pruning/workflow-history-compaction-trim.task';
+import { TelemetryPulseTask } from '@/telemetry/telemetry-pulse.task';
 import { WorkflowPublicationOutboxCleanupTask } from '@/workflows/publication/workflow-publication-outbox-cleanup.task';
 
 import { mainSystemTasks } from '../main-system-tasks';
 
-const configWith = ({ pruneData = true, useWorkflowPublicationService = true } = {}) =>
+const configWith = ({
+	pruneData = true,
+	useWorkflowPublicationService = true,
+	diagnosticsEnabled = true,
+} = {}) =>
 	mock<GlobalConfig>({
 		executions: { pruneData },
 		workflows: { useWorkflowPublicationService },
+		diagnostics: { enabled: diagnosticsEnabled },
 	});
 
 it('should return every main task when all features are on', async () => {
@@ -25,10 +31,17 @@ it('should return every main task when all features are on', async () => {
 		WorkflowHistoryCompactionOptimizeTask,
 		WorkflowHistoryCompactionTrimTask,
 		ExecutionPruningSoftDeleteTask,
+		TelemetryPulseTask,
 		WorkflowPublicationOutboxCleanupTask,
 	]);
 });
 
+it('should leave out the telemetry pulse when diagnostics are off', async () => {
+	const tasks = await mainSystemTasks(configWith({ diagnosticsEnabled: false }));
+
+	expect(tasks).not.toContain(TelemetryPulseTask);
+});
+
 it('should leave out execution pruning soft delete when pruning is off', async () => {
 	const tasks = await mainSystemTasks(configWith({ pruneData: false }));
 

packages/cli/src/scheduling/system-tasks/main-system-tasks.ts

@@ -30,6 +30,11 @@ export async function mainSystemTasks(globalConfig: GlobalConfig): Promise<Syste
 		tasks.push(ExecutionPruningSoftDeleteTask);
 	}
 
+	if (globalConfig.diagnostics.enabled) {
+		const { TelemetryPulseTask } = await import('@/telemetry/telemetry-pulse.task.js');
+		tasks.push(TelemetryPulseTask);
+	}
+
 	if (globalConfig.workflows.useWorkflowPublicationService) {
 		const { WorkflowPublicationOutboxCleanupTask } = await import(
 			'@/workflows/publication/workflow-publication-outbox-cleanup.task.js'

packages/cli/src/telemetry/__tests__/telemetry-pulse.task.test.ts

@@ -0,0 +1,28 @@
+import { mock } from 'vitest-mock-extended';
+
+import type { Telemetry } from '@/telemetry';
+
+import { TelemetryPulseTask } from '../telemetry-pulse.task';
+
+describe('TelemetryPulseTask', () => {
+	let telemetry = mock<Telemetry>();
+	let task = new TelemetryPulseTask(telemetry);
+
+	beforeEach(() => {
+		telemetry = mock<Telemetry>();
+		task = new TelemetryPulseTask(telemetry);
+	});
+
+	it('should send one packet for the whole cluster every 6 hours', () => {
+		expect(task.name).toBe('telemetry-pulse');
+		expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 21600 });
+		expect(task.effects).toBe('non-idempotent');
+		expect(task.placement).toEqual({ scope: 'cluster', durable: false });
+	});
+
+	it('should send one pulse packet', async () => {
+		await task.run();
+
+		expect(telemetry.sendPulsePacket).toHaveBeenCalledTimes(1);
+	});
+});

packages/cli/src/telemetry/__tests__/telemetry.test.ts

@@ -2,6 +2,8 @@ import type { Logger } from '@n8n/backend-common';
 import type { OutboundHttp } from '@n8n/backend-network';
 import { mockInstance } from '@n8n/backend-test-utils';
 import type { GlobalConfig } from '@n8n/config';
+import { ProjectRelationRepository, ProjectRepository, UserRepository } from '@n8n/db';
+import type { WorkflowRepository } from '@n8n/db';
 import { defineTelemetryEvents, TELEMETRY_EVENT } from '@n8n/telemetry';
 import type RudderStack from '@rudderstack/rudder-sdk-node';
 import { InstanceSettings } from 'n8n-core';
@@ -9,6 +11,8 @@ import type { MockInstance } from 'vitest';
 import { mock } from 'vitest-mock-extended';
 import { z } from 'zod/v4';
 
+import type { License } from '@/license';
+import { SourceControlPreferencesService } from '@/modules/source-control.ee/source-control-preferences.service.ee';
 import { PostHogClient } from '@/posthog';
 import { Telemetry } from '@/telemetry';
 
@@ -26,6 +30,8 @@ vi.mock('@rudderstack/rudder-sdk-node', () => ({
 describe('Telemetry', () => {
 	let spyTrack: MockInstance;
 
+	let spyStartPulse: MockInstance;
+
 	const mockRudderStack = mock<RudderStack>();
 
 	let telemetry: Telemetry;
@@ -52,7 +58,7 @@ describe('Telemetry', () => {
 	beforeEach(async () => {
 		spyTrack = vi.spyOn(Telemetry.prototype, 'track').mockName('track');
 		// @ts-expect-error Spying on private method
-		vi.spyOn(Telemetry.prototype, 'startPulse').mockImplementation(function () {});
+		spyStartPulse = vi.spyOn(Telemetry.prototype, 'startPulse').mockImplementation(function () {});
 
 		const postHog = new PostHogClient(instanceSettings, mock());
 		await postHog.init();
@@ -1193,8 +1199,129 @@ describe('Telemetry', () => {
 			expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('failed schema validation'));
 		});
 	});
+
+	describe('sendPulsePacket', () => {
+		const license = mock<License>({
+			getPlanName: () => 'enterprise',
+			getTriggerLimit: () => 400,
+		});
+		const workflowRepository = mock<WorkflowRepository>({
+			getActiveTriggerCount: async () => 7,
+		});
+
+		let pulseTelemetry: Telemetry;
+
+		beforeEach(() => {
+			mockPulsePacketSources();
+
+			pulseTelemetry = new Telemetry(
+				mock(),
+				new PostHogClient(instanceSettings, mock()),
+				license,
+				instanceSettings,
+				workflowRepository,
+				globalConfig,
+				mock(),
+				mock(),
+			);
+		});
+
+		afterEach(async () => {
+			await pulseTelemetry.stopTracking();
+		});
+
+		test('should send one packet of instance-wide counters', async () => {
+			// @ts-expect-error Assigning to private property
+			pulseTelemetry.rudderStack = mockRudderStack;
+
+			await pulseTelemetry.sendPulsePacket();
+
+			expect(spyTrack).toHaveBeenCalledWith('pulse', {
+				plan_name_current: 'enterprise',
+				quota: 400,
+				usage: 7,
+				role_count: { owner: 1 },
+				source_control_set_up: 'main',
+				branchName: 'main',
+				read_only_instance: true,
+				team_projects: 2,
+				project_role_count: {
+					'project:admin': 4,
+					'project:chatUser': 0,
+					'project:editor': 0,
+					'project:personalOwner': 0,
+					'project:viewer': 0,
+				},
+			});
+		});
+
+		test('should send nothing when RudderStack is not initialized', async () => {
+			await pulseTelemetry.sendPulsePacket();
+
+			expect(spyTrack).not.toHaveBeenCalled();
+		});
+	});
+
+	describe('pulse interval', () => {
+		test('should flush the buffers without sending a pulse packet', async () => {
+			mockPulsePacketSources();
+			spyStartPulse.mockRestore();
+
+			const intervalTelemetry = new Telemetry(
+				mock(),
+				new PostHogClient(instanceSettings, mock()),
+				mock(),
+				instanceSettings,
+				mock(),
+				globalConfig,
+				mock(),
+				mock(),
+			);
+			// @ts-expect-error Assigning to private property
+			intervalTelemetry.rudderStack = mockRudderStack;
+			// @ts-expect-error Calling a private method
+			intervalTelemetry.startPulse();
+
+			intervalTelemetry.trackApiInvocation({
+				user_id: 'user1',
+				path: '/workflows',
+				method: 'GET',
+				api_version: 'v1',
+			});
+
+			await vi.advanceTimersByTimeAsync(6 * 60 * 60 * 1000);
+
+			expect(spyTrack).toHaveBeenCalledWith(
+				'Public API usage',
+				expect.objectContaining({ user_id: 'user1' }),
+			);
+			expect(spyTrack).not.toHaveBeenCalledWith('pulse', expect.anything());
+			expect(intervalTelemetry.getApiInvocationsBuffer()).toEqual({});
+
+			await intervalTelemetry.stopTracking();
+		});
+	});
 });
 
+/** Registers the services the pulse packet reads its counters from. */
+const mockPulsePacketSources = () => {
+	mockInstance(SourceControlPreferencesService, {
+		getPreferences: () => mock({ branchName: 'main', branchReadOnly: true }),
+		isSourceControlSetup: () => 'main',
+	});
+	mockInstance(UserRepository, { countUsersByRole: async () => ({ owner: 1 }) });
+	mockInstance(ProjectRepository, { getProjectCounts: async () => ({ personal: 3, team: 2 }) });
+	mockInstance(ProjectRelationRepository, {
+		countUsersByRole: async () => ({
+			'project:admin': 4,
+			'project:chatUser': 0,
+			'project:editor': 0,
+			'project:personalOwner': 0,
+			'project:viewer': 0,
+		}),
+	});
+};
+
 const fakeTestSystemTime = (dateTime: string | Date): Date => {
 	const dt = new Date(dateTime);
 	vi.setSystemTime(dt);

packages/cli/src/telemetry/index.ts

@@ -216,14 +216,18 @@ export class Telemetry {
 
 	private startPulse() {
 		this.pulseIntervalReference = setInterval(
-			async () => {
-				void this.pulse();
+			() => {
+				this.flushBuffers();
 			},
 			6 * 60 * 60 * 1000,
 		); // every 6 hours
 	}
 
-	private async pulse() {
+	/**
+	 * Sends the events buffered in this process and empties the buffers. Does
+	 * nothing while diagnostics are off, because nothing buffers then.
+	 */
+	private flushBuffers(): void {
 		if (!this.rudderStack) {
 			return;
 		}
@@ -231,21 +235,18 @@ export class Telemetry {
 		this.flushWorkflowExecutionCounts();
 		this.flushAgentExecutionCounts();
 		this.flushAgentSessionMetrics();
+		this.flushApiInvocations();
+	}
 
-		// Flush API invocation counts
-		for (const userId of Object.keys(this.apiInvocationsBuffer)) {
-			const entry = this.apiInvocationsBuffer[userId];
-			if (entry.total_calls > 0) {
-				this.track('Public API usage', {
-					user_id: userId,
-					total_calls: entry.total_calls,
-					first: entry.first,
-					endpoints: JSON.stringify(entry.endpoints),
-					user_agents: JSON.stringify(entry.user_agents),
-				});
-			}
+	/**
+	 * Sends one `pulse` packet of license and usage counters. The counters
+	 * describe the whole instance, so a second sender reports the same numbers
+	 * again. Does nothing while diagnostics are off.
+	 */
+	async sendPulsePacket(): Promise<void> {
+		if (!this.rudderStack) {
+			return;
 		}
-		this.apiInvocationsBuffer = {};
 
 		const sourceControlPreferences = Container.get(
 			SourceControlPreferencesService,
@@ -267,6 +268,22 @@ export class Telemetry {
 		this.track('pulse', pulsePacket);
 	}
 
+	private flushApiInvocations() {
+		for (const userId of Object.keys(this.apiInvocationsBuffer)) {
+			const entry = this.apiInvocationsBuffer[userId];
+			if (entry.total_calls > 0) {
+				this.track('Public API usage', {
+					user_id: userId,
+					total_calls: entry.total_calls,
+					first: entry.first,
+					endpoints: JSON.stringify(entry.endpoints),
+					user_agents: JSON.stringify(entry.user_agents),
+				});
+			}
+		}
+		this.apiInvocationsBuffer = {};
+	}
+
 	private flushWorkflowExecutionCounts() {
 		const workflowIdsToReport = Object.keys(this.executionCountsBuffer).filter((workflowId) => {
 			const data = this.executionCountsBuffer[workflowId];

packages/cli/src/telemetry/telemetry-pulse.task.ts

@@ -0,0 +1,33 @@
+import { Time } from '@n8n/constants';
+import { SystemTask } from '@n8n/decorators';
+import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators';
+
+import { Telemetry } from '@/telemetry';
+
+/**
+ * Sends the `pulse` packet of license and usage counters.
+ */
+@SystemTask()
+export class TelemetryPulseTask implements SystemTask {
+	readonly name = 'telemetry-pulse';
+
+	readonly schedule: SystemTaskSchedule = {
+		kind: 'interval',
+		intervalSeconds: 6 * Time.hours.toSeconds,
+	};
+
+	/** The packet carries instance-wide counters, so a repeat over-reports them. */
+	readonly effects: SystemTaskEffects = 'non-idempotent';
+
+	/**
+	 * Each leader takeover restarts the six-hour wait. Delayed or missing packets are
+	 * acceptable until this task moves to the durable scheduler.
+	 */
+	readonly placement: SystemTaskPlacement = { scope: 'cluster', durable: false };
+
+	constructor(private readonly telemetry: Telemetry) {}
+
+	async run(): Promise<void> {
+		await this.telemetry.sendPulsePacket();
+	}
+}