diff --git a/packages/aws-cdk/lib/cli/telemetry/session.ts b/packages/aws-cdk/lib/cli/telemetry/session.ts index 0ab769a34..b488f17a6 100644 --- a/packages/aws-cdk/lib/cli/telemetry/session.ts +++ b/packages/aws-cdk/lib/cli/telemetry/session.ts @@ -64,6 +64,7 @@ export class TelemetrySession { private sigintListener?: () => void; private _nextEventCounters?: Record; private count = 0; + private lastStateUpdate: Promise = Promise.resolve(); private loadTime?: number; constructor(private readonly props: TelemetrySessionProps) { @@ -241,6 +242,12 @@ export class TelemetrySession { public async emit(event: TelemetryEvent): Promise { this.count += 1; + const { event: { command }, environment, project } = this.sessionInfo; + const identifiers = { + ...this.sessionInfo.identifiers, + eventId: `${this.sessionInfo.identifiers.sessionId}:${this.count}`, + timestamp: new Date().toISOString(), + }; const counters = { ...this._nextEventCounters, @@ -254,17 +261,13 @@ export class TelemetrySession { return this.client.emit({ event: { - command: this.sessionInfo.event.command, + command, state: getState(event.error), eventType: event.eventType, }, - identifiers: { - ...this.sessionInfo.identifiers, - eventId: `${this.sessionInfo.identifiers.sessionId}:${this.count}`, - timestamp: new Date().toISOString(), - }, - environment: this.sessionInfo.environment, - project: this.sessionInfo.project, + identifiers, + environment, + project, duration: { total: event.duration, }, @@ -284,7 +287,8 @@ export class TelemetrySession { * failed DEPLOY events in a sequence. */ private async trackDeployStatistics(isSuccessful: boolean, counters: Record) { - await withTelemetryState((state) => { + // Concurrent deployments read and write the same state file, so update it one at a time. + const update = this.lastStateUpdate.then(() => withTelemetryState((state) => { const recentFailures = state.sequentialDeploymentFailures ?? 0; if (isSuccessful) { @@ -294,7 +298,14 @@ export class TelemetrySession { } counters.sequentialDeploymentFailures = state.sequentialDeploymentFailures; - }); + })); + this.lastStateUpdate = update.catch(() => undefined); + + try { + await update; + } catch (e: any) { + await this.ioHost.defaults.trace(`Failed to update telemetry state: ${e.message}`); + } } private get sessionInfo(): SessionSchema { diff --git a/packages/aws-cdk/test/cli/telemetry/session.test.ts b/packages/aws-cdk/test/cli/telemetry/session.test.ts index 556c9ceab..c8ca7436e 100644 --- a/packages/aws-cdk/test/cli/telemetry/session.test.ts +++ b/packages/aws-cdk/test/cli/telemetry/session.test.ts @@ -7,6 +7,7 @@ import { CliIoHost } from '../../../lib/cli/io-host'; import { type TelemetrySchema } from '../../../lib/cli/telemetry/schema'; import { TelemetrySession, isValidWrapperUserAgent } from '../../../lib/cli/telemetry/session'; import { IoHostTelemetrySink } from '../../../lib/cli/telemetry/sink/io-host-sink'; +import * as telemetryState from '../../../lib/cli/telemetry/telemetry-state'; import { withEnv } from '../../_helpers/with-env'; let ioHost: CliIoHost; @@ -34,6 +35,10 @@ describe('TelemetrySession', () => { clientFlushSpy = jest.spyOn(client, 'flush'); }); + afterEach(() => { + jest.restoreAllMocks(); + }); + test('can emit data to the client', async () => { // WHEN await session.emit({ @@ -114,6 +119,63 @@ describe('TelemetrySession', () => { })); }); + test('concurrent DEPLOY events get distinct event ids', async () => { + // GIVEN + jest.spyOn(telemetryState, 'withTelemetryState').mockResolvedValue(undefined); + + // WHEN: two stacks finish deploying at the same time, as with --concurrency + const first = session.emit({ eventType: 'DEPLOY', duration: 1234 }); + const second = session.emit({ eventType: 'DEPLOY', duration: 1234 }); + await first; + await second; + + // THEN + const sequenceNumbers = clientEmitSpy.mock.calls.map(([e]) => e.identifiers.eventId.split(':').pop()); + expect(sequenceNumbers.sort()).toEqual(['1', '2']); + }); + + test('concurrent failed DEPLOY events count every sequential failure', async () => { + // GIVEN: an in-memory state file that yields between read and write, so updates can overlap + let state: telemetryState.TelemetryState = {}; + jest.spyOn(telemetryState, 'withTelemetryState').mockImplementation(async (block) => { + const copy = { ...state }; + await Promise.resolve(); + const ret = block(copy); + state = copy; + return ret; + }); + + // WHEN + const error = { name: ToolkitError.name }; + const first = session.emit({ eventType: 'DEPLOY', duration: 1234, error }); + const second = session.emit({ eventType: 'DEPLOY', duration: 1234, error }); + const third = session.emit({ eventType: 'DEPLOY', duration: 1234, error }); + await first; + await second; + await third; + + // THEN + const failures = clientEmitSpy.mock.calls.map(([e]) => e.counters?.sequentialDeploymentFailures); + expect(failures).toEqual([1, 2, 3]); + expect(state.sequentialDeploymentFailures).toEqual(3); + }); + + test('DEPLOY event is still sent when the state file cannot be updated, and later ones are counted', async () => { + // GIVEN + jest.spyOn(telemetryState, 'withTelemetryState') + .mockRejectedValueOnce(new Error('EACCES')) + .mockImplementation(async (block) => block({})); + + // WHEN + await session.emit({ eventType: 'DEPLOY', duration: 1234 }); + await session.emit({ eventType: 'DEPLOY', duration: 1234, error: { name: ToolkitError.name } }); + + // THEN + expect(clientEmitSpy).toHaveBeenCalledTimes(2); + expect(clientEmitSpy.mock.calls[0][0].counters?.sequentialDeploymentFailures).toBeUndefined(); + expect(clientEmitSpy.mock.calls[1][0].counters?.sequentialDeploymentFailures).toEqual(1); + }); + test('calling end more than once results in no-op', async () => { // GIVEN const spanEndSpy = jest.spyOn(session.commandSpan!, 'end');