chore(cli): funnel (#760)

A funnel is a combination sink class. It will be used later to record
telemetry both to file _and_ send to endpoint. That version will be used
in the situation where telemetry is turned on _and_ `--telemetry-file`
is set.

---
By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache-2.0 license

---------

Signed-off-by: github-actions <[email protected]>
Co-authored-by: github-actions <[email protected]>

Author: kaizencc

packages/aws-cdk/lib/cli/telemetry/sink/funnel.ts

@@ -0,0 +1,36 @@
+import { ToolkitError } from '@aws-cdk/toolkit-lib';
+import type { TelemetrySchema } from '../schema';
+import type { ITelemetrySink } from './sink-interface';
+
+export interface FunnelProps {
+  readonly sinks: ITelemetrySink[];
+}
+
+/**
+ * A funnel is a combination of one or more sinks.
+ * The sink functions are executed in parallel, and a maximum of 5
+ * sinks are supported per funnel.
+ */
+export class Funnel {
+  private readonly sinks: ITelemetrySink[];
+
+  constructor(props: FunnelProps) {
+    if (props.sinks.length > 5) {
+      throw new ToolkitError(`Funnel class supports a maximum of 5 parallel sinks, got ${props.sinks.length} sinks.`);
+    }
+
+    this.sinks = props.sinks;
+  }
+
+  public async emit(event: TelemetrySchema): Promise<void> {
+    // Funnel class enforces a maximum of 5 parallel sinks
+    // eslint-disable-next-line @cdklabs/promiseall-no-unbounded-parallelism
+    await Promise.all(this.sinks.map(sink => sink.emit(event)));
+  }
+
+  public async flush(): Promise<void> {
+    // Funnel class enforces a maximum of 5 parallel sinks
+    // eslint-disable-next-line @cdklabs/promiseall-no-unbounded-parallelism
+    await Promise.all(this.sinks.map(sink => sink.flush()));
+  }
+}

packages/aws-cdk/test/cli/telemetry/sink/endpoint-sink.test.ts

@@ -1,49 +1,14 @@
 import * as https from 'https';
+import { createTestEvent } from './util';
 import { IoHelper } from '../../../../lib/api-private';
 import { CliIoHost } from '../../../../lib/cli/io-host';
-import type { EventType, TelemetrySchema } from '../../../../lib/cli/telemetry/schema';
 import { EndpointTelemetrySink } from '../../../../lib/cli/telemetry/sink/endpoint-sink';
 
 // Mock the https module
 jest.mock('https', () => ({
   request: jest.fn(),
 }));
 
-// Helper function to create a test event
-function createTestEvent(eventType: EventType, properties: Record<string, any> = {}): TelemetrySchema {
-  return {
-    identifiers: {
-      cdkCliVersion: '1.0.0',
-      telemetryVersion: '1.0.0',
-      sessionId: 'test-session',
-      eventId: `test-event-${eventType}`,
-      installationId: 'test-installation',
-      timestamp: new Date().toISOString(),
-    },
-    event: {
-      state: 'SUCCEEDED',
-      eventType,
-      command: {
-        path: ['test'],
-        parameters: {},
-        config: properties,
-      },
-    },
-    environment: {
-      os: {
-        platform: 'test',
-        release: 'test',
-      },
-      ci: false,
-      nodeVersion: process.version,
-    },
-    project: {},
-    duration: {
-      total: 0,
-    },
-  };
-}
-
 describe('EndpointTelemetrySink', () => {
   let ioHost: CliIoHost;
 

packages/aws-cdk/test/cli/telemetry/sink/file-sink.test.ts

@@ -1,9 +1,9 @@
 import * as os from 'os';
 import * as path from 'path';
 import * as fs from 'fs-extra';
+import { createTestEvent } from './util';
 import { IoHelper } from '../../../../lib/api-private';
 import { CliIoHost } from '../../../../lib/cli/io-host';
-import type { TelemetrySchema } from '../../../../lib/cli/telemetry/schema';
 import { FileTelemetrySink } from '../../../../lib/cli/telemetry/sink/file-sink';
 
 describe('FileTelemetrySink', () => {
@@ -32,37 +32,7 @@ describe('FileTelemetrySink', () => {
 
   test('saves data to a file', async () => {
     // GIVEN
-    const testEvent: TelemetrySchema = {
-      identifiers: {
-        cdkCliVersion: '1.0.0',
-        telemetryVersion: '1.0.0',
-        sessionId: 'test-session',
-        eventId: 'test-event',
-        installationId: 'test-installation',
-        timestamp: new Date().toISOString(),
-      },
-      event: {
-        state: 'SUCCEEDED',
-        eventType: 'INVOKE',
-        command: {
-          path: ['test'],
-          parameters: {},
-          config: { context: { foo: true } },
-        },
-      },
-      environment: {
-        os: {
-          platform: 'test',
-          release: 'test',
-        },
-        ci: false,
-        nodeVersion: process.version,
-      },
-      project: {},
-      duration: {
-        total: 0,
-      },
-    };
+    const testEvent = createTestEvent('INVOKE', { context: { foo: true } });
     const client = new FileTelemetrySink({ logFilePath, ioHost });
 
     // WHEN
@@ -76,37 +46,7 @@ describe('FileTelemetrySink', () => {
 
   test('appends data to a file', async () => {
     // GIVEN
-    const testEvent: TelemetrySchema = {
-      identifiers: {
-        cdkCliVersion: '1.0.0',
-        telemetryVersion: '1.0.0',
-        sessionId: 'test-session',
-        eventId: 'test-event',
-        installationId: 'test-installation',
-        timestamp: new Date().toISOString(),
-      },
-      event: {
-        state: 'SUCCEEDED',
-        eventType: 'INVOKE',
-        command: {
-          path: ['test'],
-          parameters: {},
-          config: { context: { foo: true } },
-        },
-      },
-      environment: {
-        os: {
-          platform: 'test',
-          release: 'test',
-        },
-        ci: false,
-        nodeVersion: process.version,
-      },
-      project: {},
-      duration: {
-        total: 0,
-      },
-    };
+    const testEvent = createTestEvent('INVOKE', { context: { foo: true } });
     const client = new FileTelemetrySink({ logFilePath, ioHost });
 
     // WHEN
@@ -130,37 +70,7 @@ describe('FileTelemetrySink', () => {
 
   test('handles errors gracefully and logs to trace without throwing', async () => {
     // GIVEN
-    const testEvent: TelemetrySchema = {
-      identifiers: {
-        cdkCliVersion: '1.0.0',
-        telemetryVersion: '1.0.0',
-        sessionId: 'test-session',
-        eventId: 'test-event',
-        installationId: 'test-installation',
-        timestamp: new Date().toISOString(),
-      },
-      event: {
-        state: 'SUCCEEDED',
-        eventType: 'INVOKE',
-        command: {
-          path: ['test'],
-          parameters: {},
-          config: { context: { foo: true } },
-        },
-      },
-      environment: {
-        os: {
-          platform: 'test',
-          release: 'test',
-        },
-        ci: false,
-        nodeVersion: process.version,
-      },
-      project: {},
-      duration: {
-        total: 0,
-      },
-    };
+    const testEvent = createTestEvent('INVOKE', { context: { foo: true } });
 
     // Create a mock IoHelper with trace spy
     const traceSpy = jest.fn();