feat(toolkit-lib): structured log monitoring events

mrgrain · mrgrain · commit 24886204df13 · 2025-03-05T11:11:33.000Z
diff --git a/packages/@aws-cdk/toolkit-lib/lib/toolkit/toolkit.ts b/packages/@aws-cdk/toolkit-lib/lib/toolkit/toolkit.ts
@@ -509,7 +509,7 @@ export class Toolkit extends CloudAssemblySourceBuilder implements AsyncDisposab
       } finally {
         if (options.traceLogs) {
           // deploy calls that originate from watch will come with their own cloudWatchLogMonitor
-          const cloudWatchLogMonitor = options.cloudWatchLogMonitor ?? new CloudWatchLogEventMonitor();
+          const cloudWatchLogMonitor = options.cloudWatchLogMonitor ?? new CloudWatchLogEventMonitor({ action, ioHost });
           const foundLogGroupsResult = await findCloudWatchLogGroups(await this.sdkProvider('deploy'), { ioHost, action }, stack);
           cloudWatchLogMonitor.addLogGroups(
             foundLogGroupsResult.env,
@@ -618,10 +618,10 @@ export class Toolkit extends CloudAssemblySourceBuilder implements AsyncDisposab
     // --------------                --------  'cdk deploy' done  --------------  'cdk deploy' done  --------------
     let latch: 'pre-ready' | 'open' | 'deploying' | 'queued' = 'pre-ready';
 
-    const cloudWatchLogMonitor = options.traceLogs ? new CloudWatchLogEventMonitor() : undefined;
+    const cloudWatchLogMonitor = options.traceLogs ? new CloudWatchLogEventMonitor({ ioHost, action: 'watch' }) : undefined;
     const deployAndWatch = async () => {
       latch = 'deploying';
-      cloudWatchLogMonitor?.deactivate();
+      await cloudWatchLogMonitor?.deactivate();
 
       await this.invokeDeployFromWatch(assembly, options, cloudWatchLogMonitor);
 
@@ -635,7 +635,7 @@ export class Toolkit extends CloudAssemblySourceBuilder implements AsyncDisposab
         await this.invokeDeployFromWatch(assembly, options, cloudWatchLogMonitor);
       }
       latch = 'open';
-      cloudWatchLogMonitor?.activate();
+      await cloudWatchLogMonitor?.activate();
     };
 
     chokidar
diff --git a/packages/aws-cdk/lib/api/logs/find-cloudwatch-logs.ts b/packages/aws-cdk/lib/api/logs/find-cloudwatch-logs.ts
@@ -1,6 +1,6 @@
 import type { CloudFormationStackArtifact, Environment } from '@aws-cdk/cx-api';
 import type { StackResourceSummary } from '@aws-sdk/client-cloudformation';
-import { debug } from '../../logging';
+import { debug } from '../../cli/messages';
 import { IoMessaging } from '../../toolkit/cli-io-host';
 import { formatErrorMessage } from '../../util';
 import type { SDK, SdkProvider } from '../aws-auth';
@@ -47,7 +47,7 @@ export async function findCloudWatchLogGroups(
   try {
     sdk = (await new EnvironmentAccess(sdkProvider, DEFAULT_TOOLKIT_STACK_NAME, msg).accessStackForLookup(stackArtifact)).sdk;
   } catch (e: any) {
-    debug(`Failed to access SDK environment: ${formatErrorMessage(e)}`);
+    await msg.ioHost.notify(debug(msg.action, `Failed to access SDK environment: ${formatErrorMessage(e)}`));
     sdk = (await sdkProvider.forEnvironment(resolvedEnv, Mode.ForReading)).sdk;
   }
 
diff --git a/packages/aws-cdk/lib/api/logs/logs-monitor.ts b/packages/aws-cdk/lib/api/logs/logs-monitor.ts
@@ -1,24 +1,34 @@
 import * as util from 'util';
 import * as cxapi from '@aws-cdk/cx-api';
 import * as chalk from 'chalk';
-import { info, error } from '../../logging';
+import * as uuid from 'uuid';
+import { debug, error, info } from '../../cli/messages';
+import { IoMessaging } from '../../toolkit/cli-io-host';
 import { flatten } from '../../util';
 import type { SDK } from '../aws-auth';
 
 /**
- * After reading events from all CloudWatch log groups
- * how long should we wait to read more events.
- *
- * If there is some error with reading events (i.e. Throttle)
- * then this is also how long we wait until we try again
+ * Payload when stack monitoring is starting or stopping for a given stack deployment.
  */
-const SLEEP = 2_000;
+export interface CloudWatchLogMonitorControlEvent {
+  /**
+   * A unique identifier for a monitor
+   *
+   * Use this value to attribute events received for concurrent log monitoring.
+   */
+  readonly monitor: string;
+
+  /**
+   * The names of monitored log groups
+   */
+  readonly logGroupNames: string[];
+}
 
 /**
  * Represents a CloudWatch Log Event that will be
  * printed to the terminal
  */
-interface CloudWatchLogEvent {
+export interface CloudWatchLogEvent {
   /**
    * The log event message
    */
@@ -54,6 +64,25 @@ interface LogGroupsAccessSettings {
   readonly logGroupsStartTimes: { [logGroupName: string]: number };
 }
 
+export interface CloudWatchLogEventMonitorProps {
+  /**
+   * The IoHost used for messaging
+   */
+  readonly ioHost: IoMessaging['ioHost'];
+
+  /**
+   * The current ToolkitAction
+   */
+  readonly action: IoMessaging['action'];
+
+  /**
+   * The time from which we start reading log messages
+   *
+   * @default - now
+   */
+  readonly startTime?: Date;
+}
+
 export class CloudWatchLogEventMonitor {
   /**
    * Determines which events not to display
@@ -65,18 +94,38 @@ export class CloudWatchLogEventMonitor {
    */
   private readonly envsLogGroupsAccessSettings = new Map<string, LogGroupsAccessSettings>();
 
-  private active = false;
+  /**
+   * After reading events from all CloudWatch log groups
+   * how long should we wait to read more events.
+   *
+   * If there is some error with reading events (i.e. Throttle)
+   * then this is also how long we wait until we try again
+   */
+  private readonly pollingInterval: number = 2_000;
+
+  private monitorId?: string;
+  private readonly ioHost: IoMessaging['ioHost'];
+  private readonly action: IoMessaging['action'];
 
-  constructor(startTime?: Date) {
-    this.startTime = startTime?.getTime() ?? Date.now();
+  constructor(props: CloudWatchLogEventMonitorProps) {
+    this.startTime = props.startTime?.getTime() ?? Date.now();
+    this.ioHost = props.ioHost;
+    this.action = props.action;
   }
 
   /**
    * resume reading/printing events
    */
-  public activate(): void {
-    this.active = true;
-    this.scheduleNextTick(0);
+  public async activate(): Promise<void> {
+    this.monitorId = uuid.v4();
+
+    await this.ioHost.notify(debug(this.action, 'Start monitoring log groups', '', {
+      monitor: this.monitorId,
+      logGroupNames: this.logGroupNames(),
+    } as CloudWatchLogMonitorControlEvent));
+
+    await this.tick();
+    this.scheduleNextTick();
   }
 
   /**
@@ -88,9 +137,16 @@ export class CloudWatchLogEventMonitor {
    * Also resets the start time to be when the new deployment was triggered
    * and clears the list of tracked log groups
    */
-  public deactivate(): void {
-    this.active = false;
+  public async deactivate(): Promise<void> {
+    const oldMonitorId = this.monitorId!;
+    this.monitorId = undefined;
     this.startTime = Date.now();
+
+    await this.ioHost.notify(debug(this.action, 'Stopped monitoring log groups', '', {
+      monitor: oldMonitorId,
+      logGroupNames: this.logGroupNames(),
+    } as CloudWatchLogMonitorControlEvent));
+
     this.envsLogGroupsAccessSettings.clear();
   }
 
@@ -119,27 +175,41 @@ export class CloudWatchLogEventMonitor {
     });
   }
 
-  private scheduleNextTick(sleep: number): void {
-    setTimeout(() => void this.tick(), sleep);
+  private logGroupNames(): string[] {
+    return Array.from(this.envsLogGroupsAccessSettings.values()).flatMap((settings) => Object.keys(settings.logGroupsStartTimes));
+  }
+
+  private scheduleNextTick(): void {
+    if (!this.monitorId) {
+      return;
+    }
+
+    setTimeout(() => void this.tick(), this.pollingInterval);
   }
 
   private async tick(): Promise<void> {
     // excluding from codecoverage because this
     // doesn't always run (depends on timing)
-    /* istanbul ignore next */
-    if (!this.active) {
+    /* c8 ignore next */
+    if (!this.monitorId) {
       return;
     }
+
     try {
       const events = flatten(await this.readNewEvents());
-      events.forEach((event) => {
-        this.print(event);
-      });
+      for (const event of events) {
+        await this.print(event);
+      }
+
+      // We might have been stop()ped while the network call was in progress.
+      if (!this.monitorId) {
+        return;
+      }
     } catch (e) {
-      error('Error occurred while monitoring logs: %s', e);
+      await this.ioHost.notify(error(this.action, 'Error occurred while monitoring logs: %s', { error: e }));
     }
 
-    this.scheduleNextTick(SLEEP);
+    this.scheduleNextTick();
   }
 
   /**
@@ -161,15 +231,18 @@ export class CloudWatchLogEventMonitor {
   /**
    * Print out a cloudwatch event
    */
-  private print(event: CloudWatchLogEvent): void {
-    info(
+  private async print(event: CloudWatchLogEvent): Promise<void> {
+    await this.ioHost.notify(info(
+      this.action,
       util.format(
         '[%s] %s %s',
         chalk.blue(event.logGroupName),
         chalk.yellow(event.timestamp.toLocaleTimeString()),
         event.message.trim(),
       ),
-    );
+      '',
+      event,
+    ));
   }
 
   /**
diff --git a/packages/aws-cdk/lib/cli/cdk-toolkit.ts b/packages/aws-cdk/lib/cli/cdk-toolkit.ts
@@ -698,10 +698,13 @@ export class CdkToolkit {
     // --------------                --------  'cdk deploy' done  --------------  'cdk deploy' done  --------------
     let latch: 'pre-ready' | 'open' | 'deploying' | 'queued' = 'pre-ready';
 
-    const cloudWatchLogMonitor = options.traceLogs ? new CloudWatchLogEventMonitor() : undefined;
+    const cloudWatchLogMonitor = options.traceLogs ? new CloudWatchLogEventMonitor({
+      action: 'watch',
+      ioHost: this.ioHost,
+    }) : undefined;
     const deployAndWatch = async () => {
       latch = 'deploying';
-      cloudWatchLogMonitor?.deactivate();
+      await cloudWatchLogMonitor?.deactivate();
 
       await this.invokeDeployFromWatch(options, cloudWatchLogMonitor);
 
@@ -715,7 +718,7 @@ export class CdkToolkit {
         await this.invokeDeployFromWatch(options, cloudWatchLogMonitor);
       }
       latch = 'open';
-      cloudWatchLogMonitor?.activate();
+      await cloudWatchLogMonitor?.activate();
     };
 
     chokidar
