feat(NODE-4686): Add log messages to CLAM (#3955)

Co-authored-by: Durran Jordan <durran@gmail.com>
Co-authored-by: Alena Khineika <alena.khineika@gmail.com>

src/bson.ts

@@ -13,6 +13,8 @@ export {
   deserialize,
   Document,
   Double,
+  EJSON,
+  EJSONOptions,
   Int32,
   Long,
   MaxKey,

src/cmap/connect.ts

@@ -186,6 +186,10 @@ async function performInitialHandshake(
       throw error;
     }
   }
+
+  // Connection establishment is socket creation (tcp handshake, tls handshake, MongoDB handshake (saslStart, saslContinue))
+  // Once connection is established, command logging can log events (if enabled)
+  conn.established = true;
 }
 
 export interface HandshakeDocument extends Document {

src/cmap/connection.ts

@@ -24,6 +24,7 @@ import {
   MongoWriteConcernError
 } from '../error';
 import type { ServerApi, SupportedNodeConnectionOptions } from '../mongo_client';
+import { MongoLoggableComponent, type MongoLogger, SeverityLevel } from '../mongo_logger';
 import { type CancellationToken, TypedEventEmitter } from '../mongo_types';
 import type { ReadPreferenceLike } from '../read_preference';
 import { applySession, type ClientSession, updateSessionFromResponse } from '../sessions';
@@ -114,6 +115,8 @@ export interface ConnectionOptions
   socketTimeoutMS?: number;
   cancellationToken?: CancellationToken;
   metadata: ClientMetadata;
+  /** @internal */
+  mongoLogger?: MongoLogger | undefined;
 }
 
 /** @internal */
@@ -165,6 +168,16 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
   public delayedTimeoutId: NodeJS.Timeout | null = null;
   public generation: number;
   public readonly description: Readonly<StreamDescription>;
+  /**
+   * @public
+   * Represents if the connection has been established:
+   *  - TCP handshake
+   *  - TLS negotiated
+   *  - mongodb handshake (saslStart, saslContinue), includes authentication
+   *
+   * Once connection is established, command logging can log events (if enabled)
+   */
+  public established: boolean;
 
   private lastUseTime: number;
   private socketTimeoutMS: number;
@@ -174,6 +187,8 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
   private messageStream: Readable;
   private socketWrite: (buffer: Uint8Array) => Promise<void>;
   private clusterTime: Document | null = null;
+  /** @internal */
+  override mongoLogger: MongoLogger | undefined;
 
   /** @event */
   static readonly COMMAND_STARTED = COMMAND_STARTED;
@@ -198,6 +213,8 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     this.socketTimeoutMS = options.socketTimeoutMS ?? 0;
     this.monitorCommands = options.monitorCommands;
     this.serverApi = options.serverApi;
+    this.mongoLogger = options.mongoLogger;
+    this.established = false;
 
     this.description = new StreamDescription(this.address, options);
     this.generation = options.generation;
@@ -258,6 +275,16 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     );
   }
 
+  private get shouldEmitAndLogCommand(): boolean {
+    return (
+      (this.monitorCommands ||
+        (this.established &&
+          !this.authContext?.reauthenticating &&
+          this.mongoLogger?.willLog(SeverityLevel.DEBUG, MongoLoggableComponent.COMMAND))) ??
+      false
+    );
+  }
+
   public markAvailable(): void {
     this.lastUseTime = now();
   }
@@ -441,10 +468,13 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     const message = this.prepareCommand(ns.db, command, options);
 
     let started = 0;
-    if (this.monitorCommands) {
+    if (this.shouldEmitAndLogCommand) {
       started = now();
-      this.emit(
+      this.emitAndLogCommand(
+        this.monitorCommands,
         Connection.COMMAND_STARTED,
+        message.databaseName,
+        this.established,
         new CommandStartedEvent(this, message, this.description.serverConnectionId)
       );
     }
@@ -464,9 +494,12 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
           throw new MongoServerError(document);
         }
 
-        if (this.monitorCommands) {
-          this.emit(
+        if (this.shouldEmitAndLogCommand) {
+          this.emitAndLogCommand(
+            this.monitorCommands,
             Connection.COMMAND_SUCCEEDED,
+            message.databaseName,
+            this.established,
             new CommandSucceededEvent(
               this,
               message,
@@ -481,10 +514,13 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
         this.controller.signal.throwIfAborted();
       }
     } catch (error) {
-      if (this.monitorCommands) {
+      if (this.shouldEmitAndLogCommand) {
         if (error.name === 'MongoWriteConcernError') {
-          this.emit(
+          this.emitAndLogCommand(
+            this.monitorCommands,
             Connection.COMMAND_SUCCEEDED,
+            message.databaseName,
+            this.established,
             new CommandSucceededEvent(
               this,
               message,
@@ -494,8 +530,11 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
             )
           );
         } else {
-          this.emit(
+          this.emitAndLogCommand(
+            this.monitorCommands,
             Connection.COMMAND_FAILED,
+            message.databaseName,
+            this.established,
             new CommandFailedEvent(
               this,
               message,

src/cmap/connection_pool.ts

@@ -699,7 +699,8 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       ...this.options,
       id: this[kConnectionCounter].next().value,
       generation: this[kGeneration],
-      cancellationToken: this[kCancellationToken]
+      cancellationToken: this[kCancellationToken],
+      mongoLogger: this.mongoLogger
     };
 
     this[kPending]++;

src/constants.ts

@@ -60,8 +60,11 @@ export const CONNECTION_CHECKED_OUT = 'connectionCheckedOut' as const;
 /** @internal */
 export const CONNECTION_CHECKED_IN = 'connectionCheckedIn' as const;
 export const CLUSTER_TIME_RECEIVED = 'clusterTimeReceived' as const;
+/** @internal */
 export const COMMAND_STARTED = 'commandStarted' as const;
+/** @internal */
 export const COMMAND_SUCCEEDED = 'commandSucceeded' as const;
+/** @internal */
 export const COMMAND_FAILED = 'commandFailed' as const;
 /** @internal */
 export const SERVER_HEARTBEAT_STARTED = 'serverHeartbeatStarted' as const;

src/index.ts

@@ -292,6 +292,9 @@ export type { StreamDescription, StreamDescriptionOptions } from './cmap/stream_
 export type { CompressorName } from './cmap/wire_protocol/compression';
 export type { CollectionOptions, CollectionPrivate, ModifyResult } from './collection';
 export type {
+  COMMAND_FAILED,
+  COMMAND_STARTED,
+  COMMAND_SUCCEEDED,
   CONNECTION_CHECK_OUT_FAILED,
   CONNECTION_CHECK_OUT_STARTED,
   CONNECTION_CHECKED_IN,
@@ -367,6 +370,8 @@ export type {
   LogComponentSeveritiesClientOptions,
   LogConvertible,
   Loggable,
+  LoggableCommandFailedEvent,
+  LoggableCommandSucceededEvent,
   LoggableEvent,
   LoggableServerHeartbeatFailedEvent,
   LoggableServerHeartbeatStartedEvent,

src/mongo_logger.ts

@@ -1,12 +1,8 @@
-import { type Document, EJSON, type EJSONOptions } from 'bson';
 import type { Writable } from 'stream';
 import { inspect } from 'util';
 
-import type {
-  CommandFailedEvent,
-  CommandStartedEvent,
-  CommandSucceededEvent
-} from './cmap/command_monitoring_events';
+import { type Document, EJSON, type EJSONOptions, type ObjectId } from './bson';
+import type { CommandStartedEvent } from './cmap/command_monitoring_events';
 import type {
   ConnectionCheckedInEvent,
   ConnectionCheckedOutEvent,
@@ -295,6 +291,40 @@ function compareSeverity(s0: SeverityLevel, s1: SeverityLevel): 1 | 0 | -1 {
   return s0Num < s1Num ? -1 : s0Num > s1Num ? 1 : 0;
 }
 
+/**
+ * @internal
+ * Must be separate from Events API due to differences in spec requirements for logging a command success
+ */
+export type LoggableCommandSucceededEvent = {
+  address: string;
+  connectionId?: string | number;
+  requestId: number;
+  duration: number;
+  commandName: string;
+  reply: Document | undefined;
+  serviceId?: ObjectId;
+  name: typeof COMMAND_SUCCEEDED;
+  serverConnectionId: bigint | null;
+  databaseName: string;
+};
+
+/**
+ * @internal
+ * Must be separate from Events API due to differences in spec requirements for logging a command failure
+ */
+export type LoggableCommandFailedEvent = {
+  address: string;
+  connectionId?: string | number;
+  requestId: number;
+  duration: number;
+  commandName: string;
+  failure: Error;
+  serviceId?: ObjectId;
+  name: typeof COMMAND_FAILED;
+  serverConnectionId: bigint | null;
+  databaseName: string;
+};
+
 /**
  * @internal
  * Must be separate from Events API due to differences in spec requirements for logging server heartbeat beginning
@@ -350,8 +380,8 @@ export type LoggableEvent =
   | ServerSelectionSucceededEvent
   | WaitingForSuitableServerEvent
   | CommandStartedEvent
-  | CommandSucceededEvent
-  | CommandFailedEvent
+  | LoggableCommandSucceededEvent
+  | LoggableCommandFailedEvent
   | ConnectionPoolCreatedEvent
   | ConnectionPoolReadyEvent
   | ConnectionPoolClosedEvent
@@ -383,7 +413,8 @@ export function stringifyWithMaxLen(
   maxDocumentLength: number,
   options: EJSONOptions = {}
 ): string {
-  let strToTruncate: string;
+  let strToTruncate = '';
+
   if (typeof value === 'function') {
     strToTruncate = value.toString();
   } else {
@@ -420,7 +451,7 @@ function attachServerSelectionFields(
 
 function attachCommandFields(
   log: Record<string, any>,
-  commandEvent: CommandStartedEvent | CommandSucceededEvent | CommandFailedEvent
+  commandEvent: CommandStartedEvent | LoggableCommandSucceededEvent | LoggableCommandFailedEvent
 ) {
   log.commandName = commandEvent.commandName;
   log.requestId = commandEvent.requestId;
@@ -431,6 +462,8 @@ function attachCommandFields(
   if (commandEvent?.serviceId) {
     log.serviceId = commandEvent.serviceId.toHexString();
   }
+  log.databaseName = commandEvent.databaseName;
+  log.serverConnectionId = commandEvent?.serverConnectionId;
 
   return log;
 }
@@ -490,20 +523,20 @@ function defaultLogTransform(
     case COMMAND_STARTED:
       log = attachCommandFields(log, logObject);
       log.message = 'Command started';
-      log.command = stringifyWithMaxLen(logObject.command, maxDocumentLength);
+      log.command = stringifyWithMaxLen(logObject.command, maxDocumentLength, { relaxed: true });
       log.databaseName = logObject.databaseName;
       return log;
     case COMMAND_SUCCEEDED:
       log = attachCommandFields(log, logObject);
       log.message = 'Command succeeded';
       log.durationMS = logObject.duration;
-      log.reply = stringifyWithMaxLen(logObject.reply, maxDocumentLength);
+      log.reply = stringifyWithMaxLen(logObject.reply, maxDocumentLength, { relaxed: true });
       return log;
     case COMMAND_FAILED:
       log = attachCommandFields(log, logObject);
       log.message = 'Command failed';
       log.durationMS = logObject.duration;
-      log.failure = logObject.failure;
+      log.failure = logObject.failure.message ?? '(redacted)';
       return log;
     case CONNECTION_POOL_CREATED:
       log = attachConnectionFields(log, logObject);
@@ -701,12 +734,16 @@ export class MongoLogger {
     this.logDestination = options.logDestination;
   }
 
+  willLog(severity: SeverityLevel, component: MongoLoggableComponent): boolean {
+    return compareSeverity(severity, this.componentSeverities[component]) <= 0;
+  }
+
   private log(
     severity: SeverityLevel,
     component: MongoLoggableComponent,
     message: Loggable | string
   ): void {
-    if (compareSeverity(severity, this.componentSeverities[component]) > 0) return;
+    if (!this.willLog(severity, component)) return;
 
     let logMessage: Log = { t: new Date(), c: component, s: severity };
     if (typeof message === 'string') {

src/mongo_types.ts

@@ -12,12 +12,15 @@ import type {
   ObjectId,
   Timestamp
 } from './bson';
-import type {
-  LoggableServerHeartbeatFailedEvent,
-  LoggableServerHeartbeatStartedEvent,
-  LoggableServerHeartbeatSucceededEvent,
+import { type CommandStartedEvent } from './cmap/command_monitoring_events';
+import {
+  type LoggableCommandFailedEvent,
+  type LoggableCommandSucceededEvent,
+  type LoggableServerHeartbeatFailedEvent,
+  type LoggableServerHeartbeatStartedEvent,
+  type LoggableServerHeartbeatSucceededEvent,
   MongoLoggableComponent,
-  MongoLogger
+  type MongoLogger
 } from './mongo_logger';
 import type { Sort } from './sort';
 
@@ -442,6 +445,28 @@ export class TypedEventEmitter<Events extends EventsDescription> extends EventEm
       this.mongoLogger?.debug(this.component, loggableHeartbeatEvent);
     }
   }
+  /** @internal */
+  emitAndLogCommand<EventKey extends keyof Events>(
+    monitorCommands: boolean,
+    event: EventKey | symbol,
+    databaseName: string,
+    connectionEstablished: boolean,
+    ...args: Parameters<Events[EventKey]>
+  ): void {
+    if (monitorCommands) {
+      this.emit(event, ...args);
+    }
+    if (connectionEstablished) {
+      const loggableCommandEvent:
+        | CommandStartedEvent
+        | LoggableCommandFailedEvent
+        | LoggableCommandSucceededEvent = {
+        databaseName: databaseName,
+        ...args[0]
+      };
+      this.mongoLogger?.debug(MongoLoggableComponent.COMMAND, loggableCommandEvent);
+    }
+  }
 }
 
 /** @public */