diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts index 3135bb5a26d..131423df3dc 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts +++ b/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({ 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 })); diff --git a/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts b/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts index bd3c7c3c177..e54d28713cf 100644 --- a/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts +++ b/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts @@ -30,6 +30,11 @@ export async function mainSystemTasks(globalConfig: GlobalConfig): Promise { + let telemetry = mock(); + let task = new TelemetryPulseTask(telemetry); + + beforeEach(() => { + telemetry = mock(); + 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); + }); +}); diff --git a/packages/cli/src/telemetry/__tests__/telemetry.test.ts b/packages/cli/src/telemetry/__tests__/telemetry.test.ts index 1e56f86c0b5..f88f148c8d7 100644 --- a/packages/cli/src/telemetry/__tests__/telemetry.test.ts +++ b/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(); 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({ + getPlanName: () => 'enterprise', + getTriggerLimit: () => 400, + }); + const workflowRepository = mock({ + 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); diff --git a/packages/cli/src/telemetry/index.ts b/packages/cli/src/telemetry/index.ts index b9e0caff34b..5a6e7475fab 100644 --- a/packages/cli/src/telemetry/index.ts +++ b/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 { + 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]; diff --git a/packages/cli/src/telemetry/telemetry-pulse.task.ts b/packages/cli/src/telemetry/telemetry-pulse.task.ts new file mode 100644 index 00000000000..47224353ed1 --- /dev/null +++ b/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 { + await this.telemetry.sendPulsePacket(); + } +}