feat(toolkit-lib): structured log monitoring events (#218)

Moves more of the APIs to use modern IO infrastructure.

---
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: mrgrain

packages/@aws-cdk/tmp-toolkit-helpers/src/api/io/payloads/index.ts

@@ -12,3 +12,4 @@ export * from './progress';
 export * from './watch';
 export * from './stack-details';
 export * from './diff';
+export * from './logs-monitor';

packages/@aws-cdk/tmp-toolkit-helpers/src/api/io/payloads/logs-monitor.ts

@@ -0,0 +1,38 @@
+
+/**
+ * Payload when stack monitoring is starting or stopping for a given stack deployment.
+ */
+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
+ */
+export interface CloudWatchLogEvent {
+  /**
+   * The log event message
+   */
+  readonly message: string;
+
+  /**
+   * The name of the log group
+   */
+  readonly logGroupName: string;
+
+  /**
+   * The time at which the event occurred
+   */
+  readonly timestamp: Date;
+}

packages/@aws-cdk/tmp-toolkit-helpers/src/api/io/private/messages.ts

@@ -6,6 +6,7 @@ import type { MissingContext, UpdatedContext } from '../payloads/context';
 import type { BuildAsset, DeployConfirmationRequest, PublishAsset, StackDeployProgress, SuccessfulDeployStackResult } from '../payloads/deploy';
 import type { StackDestroy, StackDestroyProgress } from '../payloads/destroy';
 import type { StackDetailsPayload } from '../payloads/list';
+import type { CloudWatchLogEvent, CloudWatchLogMonitorControlEvent } from '../payloads/logs-monitor';
 import type { StackRollbackProgress } from '../payloads/rollback';
 import type { SdkTrace } from '../payloads/sdk-trace';
 import type { StackActivity, StackMonitoringControlEvent } from '../payloads/stack-activity';
@@ -98,6 +99,26 @@ export const IO = {
     code: 'CDK_TOOLKIT_I5031',
     description: 'Informs about any log groups that are traced as part of the deployment',
   }),
+  CDK_TOOLKIT_I5032: make.debug<CloudWatchLogMonitorControlEvent>({
+    code: 'CDK_TOOLKIT_I5032',
+    description: 'Start monitoring log groups',
+    interface: 'CloudWatchLogMonitorControlEvent',
+  }),
+  CDK_TOOLKIT_I5033: make.info<CloudWatchLogEvent>({
+    code: 'CDK_TOOLKIT_I5033',
+    description: 'A log event received from Cloud Watch',
+    interface: 'CloudWatchLogEvent',
+  }),
+  CDK_TOOLKIT_I5034: make.debug<CloudWatchLogMonitorControlEvent>({
+    code: 'CDK_TOOLKIT_I5034',
+    description: 'Stop monitoring log groups',
+    interface: 'CloudWatchLogMonitorControlEvent',
+  }),
+  CDK_TOOLKIT_E5035: make.error<ErrorPayload>({
+    code: 'CDK_TOOLKIT_E5035',
+    description: 'A log monitoring error',
+    interface: 'ErrorPayload',
+  }),
   CDK_TOOLKIT_I5050: make.confirm<ConfirmationRequest>({
     code: 'CDK_TOOLKIT_I5050',
     description: 'Confirm rollback during deployment',

packages/@aws-cdk/toolkit-lib/docs/message-registry.md

@@ -21,6 +21,10 @@ group: Documents
 | `CDK_TOOLKIT_W5021` | Empty non-existent stack, deployment is skipped | `warn` | n/a |
 | `CDK_TOOLKIT_W5022` | Empty existing stack, stack will be destroyed | `warn` | n/a |
 | `CDK_TOOLKIT_I5031` | Informs about any log groups that are traced as part of the deployment | `info` | n/a |
+| `CDK_TOOLKIT_I5032` | Start monitoring log groups | `debug` | {@link CloudWatchLogMonitorControlEvent} |
+| `CDK_TOOLKIT_I5033` | A log event received from Cloud Watch | `info` | {@link CloudWatchLogEvent} |
+| `CDK_TOOLKIT_I5034` | Stop monitoring log groups | `debug` | {@link CloudWatchLogMonitorControlEvent} |
+| `CDK_TOOLKIT_E5035` | A log monitoring error | `error` | {@link ErrorPayload} |
 | `CDK_TOOLKIT_I5050` | Confirm rollback during deployment | `info` | {@link ConfirmationRequest} |
 | `CDK_TOOLKIT_I5060` | Confirm deploy security sensitive changes | `info` | {@link DeployConfirmationRequest} |
 | `CDK_TOOLKIT_I5100` | Stack deploy progress | `info` | {@link StackDeployProgress} |

packages/@aws-cdk/toolkit-lib/lib/toolkit/toolkit.ts

@@ -519,7 +519,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({ ioHelper });
           const foundLogGroupsResult = await findCloudWatchLogGroups(await this.sdkProvider('deploy'), ioHelper, stack);
           cloudWatchLogMonitor.addLogGroups(
             foundLogGroupsResult.env,
@@ -637,10 +637,10 @@ export class Toolkit extends CloudAssemblySourceBuilder implements AsyncDisposab
     type LatchState = 'pre-ready' | 'open' | 'deploying' | 'queued';
     let latch: LatchState = 'pre-ready';
 
-    const cloudWatchLogMonitor = options.traceLogs ? new CloudWatchLogEventMonitor() : undefined;
+    const cloudWatchLogMonitor = options.traceLogs ? new CloudWatchLogEventMonitor({ ioHelper }) : undefined;
     const deployAndWatch = async () => {
       latch = 'deploying' as LatchState;
-      cloudWatchLogMonitor?.deactivate();
+      await cloudWatchLogMonitor?.deactivate();
 
       await this.invokeDeployFromWatch(assembly, options, cloudWatchLogMonitor);
 
@@ -654,7 +654,7 @@ export class Toolkit extends CloudAssemblySourceBuilder implements AsyncDisposab
         await this.invokeDeployFromWatch(assembly, options, cloudWatchLogMonitor);
       }
       latch = 'open';
-      cloudWatchLogMonitor?.activate();
+      await cloudWatchLogMonitor?.activate();
     };
 
     chokidar

packages/aws-cdk/lib/api/logs/find-cloudwatch-logs.ts

@@ -1,7 +1,6 @@
 import type { CloudFormationStackArtifact, Environment } from '@aws-cdk/cx-api';
 import type { StackResourceSummary } from '@aws-sdk/client-cloudformation';
-import { IoHelper } from '../../../../@aws-cdk/tmp-toolkit-helpers/src/api/io/private';
-import { debug } from '../../logging';
+import { IO, IoHelper } from '../../../../@aws-cdk/tmp-toolkit-helpers/src/api/io/private';
 import { formatErrorMessage } from '../../util';
 import type { SDK, SdkProvider } from '../aws-auth';
 import { EnvironmentAccess } from '../environment';
@@ -47,7 +46,7 @@ export async function findCloudWatchLogGroups(
   try {
     sdk = (await new EnvironmentAccess(sdkProvider, DEFAULT_TOOLKIT_STACK_NAME, ioHelper).accessStackForLookup(stackArtifact)).sdk;
   } catch (e: any) {
-    debug(`Failed to access SDK environment: ${formatErrorMessage(e)}`);
+    await ioHelper.notify(IO.DEFAULT_TOOLKIT_DEBUG.msg(`Failed to access SDK environment: ${formatErrorMessage(e)}`));
     sdk = (await sdkProvider.forEnvironment(resolvedEnv, Mode.ForReading)).sdk;
   }
 

packages/aws-cdk/lib/api/logs/logs-monitor.ts

@@ -1,40 +1,12 @@
 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 type { CloudWatchLogEvent } from '../../../../@aws-cdk/tmp-toolkit-helpers/src/api/io';
+import { IO, IoHelper } from '../../../../@aws-cdk/tmp-toolkit-helpers/src/api/io/private';
 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
- */
-const SLEEP = 2_000;
-
-/**
- * Represents a CloudWatch Log Event that will be
- * printed to the terminal
- */
-interface CloudWatchLogEvent {
-  /**
-   * The log event message
-   */
-  readonly message: string;
-
-  /**
-   * The name of the log group
-   */
-  readonly logGroupName: string;
-
-  /**
-   * The time at which the event occurred
-   */
-  readonly timestamp: Date;
-}
-
 /**
  * Configuration tracking information on the log groups that are
  * being monitored
@@ -54,6 +26,20 @@ interface LogGroupsAccessSettings {
   readonly logGroupsStartTimes: { [logGroupName: string]: number };
 }
 
+export interface CloudWatchLogEventMonitorProps {
+  /**
+   * The IoHost used for messaging
+   */
+  readonly ioHelper: IoHelper;
+
+  /**
+   * 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 +51,36 @@ 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;
+
+  public monitorId?: string;
+  private readonly ioHelper: IoHelper;
 
-  constructor(startTime?: Date) {
-    this.startTime = startTime?.getTime() ?? Date.now();
+  constructor(props: CloudWatchLogEventMonitorProps) {
+    this.startTime = props.startTime?.getTime() ?? Date.now();
+    this.ioHelper = props.ioHelper;
   }
 
   /**
    * resume reading/printing events
    */
-  public activate(): void {
-    this.active = true;
-    this.scheduleNextTick(0);
+  public async activate(): Promise<void> {
+    this.monitorId = uuid.v4();
+
+    await this.ioHelper.notify(IO.CDK_TOOLKIT_I5032.msg('Start monitoring log groups', {
+      monitor: this.monitorId,
+      logGroupNames: this.logGroupNames(),
+    }));
+
+    await this.tick();
+    this.scheduleNextTick();
   }
 
   /**
@@ -88,9 +92,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.ioHelper.notify(IO.CDK_TOOLKIT_I5034.msg('Stopped monitoring log groups', {
+      monitor: oldMonitorId,
+      logGroupNames: this.logGroupNames(),
+    }));
+
     this.envsLogGroupsAccessSettings.clear();
   }
 
@@ -119,27 +130,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);
-      });
-    } catch (e) {
-      error('Error occurred while monitoring logs: %s', e);
+      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: any) {
+      await this.ioHelper.notify(IO.CDK_TOOLKIT_E5035.msg('Error occurred while monitoring logs: %s', { error: e }));
     }
 
-    this.scheduleNextTick(SLEEP);
+    this.scheduleNextTick();
   }
 
   /**
@@ -161,15 +186,16 @@ export class CloudWatchLogEventMonitor {
   /**
    * Print out a cloudwatch event
    */
-  private print(event: CloudWatchLogEvent): void {
-    info(
+  private async print(event: CloudWatchLogEvent): Promise<void> {
+    await this.ioHelper.notify(IO.CDK_TOOLKIT_I5033.msg(
       util.format(
         '[%s] %s %s',
         chalk.blue(event.logGroupName),
         chalk.yellow(event.timestamp.toLocaleTimeString()),
         event.message.trim(),
       ),
-    );
+      event,
+    ));
   }
 
   /**

packages/aws-cdk/lib/cli/cdk-toolkit.ts

@@ -649,6 +649,7 @@ export class CdkToolkit {
 
   public async watch(options: WatchOptions) {
     const rootDir = path.dirname(path.resolve(PROJECT_CONFIG));
+    const ioHelper = asIoHelper(this.ioHost, 'watch');
     debug("root directory used for 'watch' is: %s", rootDir);
 
     const watchSettings: { include?: string | string[]; exclude: string | string[] } | undefined =
@@ -697,10 +698,12 @@ 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({
+      ioHelper,
+    }) : undefined;
     const deployAndWatch = async () => {
       latch = 'deploying';
-      cloudWatchLogMonitor?.deactivate();
+      await cloudWatchLogMonitor?.deactivate();
 
       await this.invokeDeployFromWatch(options, cloudWatchLogMonitor);
 
@@ -714,7 +717,7 @@ export class CdkToolkit {
         await this.invokeDeployFromWatch(options, cloudWatchLogMonitor);
       }
       latch = 'open';
-      cloudWatchLogMonitor?.activate();
+      await cloudWatchLogMonitor?.activate();
     };
 
     chokidar