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

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,15 @@ 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 +186,10 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
   private messageStream: Readable;
   private socketWrite: (buffer: Uint8Array) => Promise<void>;
   private clusterTime: Document | null = null;
+  /** @internal  */
+  override component = MongoLoggableComponent.COMMAND;
+  /** @internal */
+  override mongoLogger: MongoLogger | undefined;
 
   /** @event */
   static readonly COMMAND_STARTED = COMMAND_STARTED;
@@ -198,6 +214,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;
@@ -441,10 +459,16 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     const message = this.prepareCommand(ns.db, command, options);
 
     let started = 0;
-    if (this.monitorCommands) {
+    if (
+      this.monitorCommands ||
+      (this.established && this.mongoLogger?.willLog(SeverityLevel.DEBUG, this.component))
+    ) {
       started = now();
-      this.emit(
+      this.emitAndLogCommand(
+        this.monitorCommands,
         Connection.COMMAND_STARTED,
+        message.databaseName,
+        this.established,
         new CommandStartedEvent(this, message, this.description.serverConnectionId)
       );
     }
@@ -464,9 +488,15 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
           throw new MongoServerError(document);
         }
 
-        if (this.monitorCommands) {
-          this.emit(
+        if (
+          this.monitorCommands ||
+          (this.established && this.mongoLogger?.willLog(SeverityLevel.DEBUG, this.component))
+        ) {
+          this.emitAndLogCommand(
+            this.monitorCommands,
             Connection.COMMAND_SUCCEEDED,
+            message.databaseName,
+            this.established,
             new CommandSucceededEvent(
               this,
               message,
@@ -481,10 +511,16 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
         this.controller.signal.throwIfAborted();
       }
     } catch (error) {
-      if (this.monitorCommands) {
+      if (
+        this.monitorCommands ||
+        (this.established && this.mongoLogger?.willLog(SeverityLevel.DEBUG, this.component))
+      ) {
         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/cmap/message_stream.ts

@@ -0,0 +1,220 @@
+import { Duplex, type DuplexOptions } from 'stream';
+
+import type { BSONSerializeOptions, Document } from '../bson';
+import { MongoDecompressionError, MongoParseError } from '../error';
+import type { ClientSession } from '../sessions';
+import { BufferPool, type Callback } from '../utils';
+import {
+  type MessageHeader,
+  OpCompressedRequest,
+  OpMsgResponse,
+  OpQueryResponse,
+  type WriteProtocolMessageType
+} from './commands';
+import { compress, Compressor, type CompressorName, decompress } from './wire_protocol/compression';
+import { OP_COMPRESSED, OP_MSG } from './wire_protocol/constants';
+
+const MESSAGE_HEADER_SIZE = 16;
+const COMPRESSION_DETAILS_SIZE = 9; // originalOpcode + uncompressedSize, compressorID
+
+const kDefaultMaxBsonMessageSize = 1024 * 1024 * 16 * 4;
+/** @internal */
+const kBuffer = Symbol('buffer');
+
+/** @internal */
+export interface MessageStreamOptions extends DuplexOptions {
+  maxBsonMessageSize?: number;
+}
+
+/** @internal */
+export interface OperationDescription extends BSONSerializeOptions {
+  started: number;
+  cb: Callback<Document>;
+  documentsReturnedIn?: string;
+  noResponse: boolean;
+  raw: boolean;
+  requestId: number;
+  session?: ClientSession;
+  agreedCompressor?: CompressorName;
+  zlibCompressionLevel?: number;
+  $clusterTime?: Document;
+}
+
+/**
+ * A duplex stream that is capable of reading and writing raw wire protocol messages, with
+ * support for optional compression
+ * @internal
+ */
+export class MessageStream extends Duplex {
+  /** @internal */
+  maxBsonMessageSize: number;
+  /** @internal */
+  [kBuffer]: BufferPool;
+  /** @internal */
+  isMonitoringConnection = false;
+
+  constructor(options: MessageStreamOptions = {}) {
+    super(options);
+    this.maxBsonMessageSize = options.maxBsonMessageSize || kDefaultMaxBsonMessageSize;
+    this[kBuffer] = new BufferPool();
+  }
+
+  get buffer(): BufferPool {
+    return this[kBuffer];
+  }
+
+  override _write(chunk: Buffer, _: unknown, callback: Callback<Buffer>): void {
+    this[kBuffer].append(chunk);
+    processIncomingData(this, callback);
+  }
+
+  override _read(/* size */): void {
+    // NOTE: This implementation is empty because we explicitly push data to be read
+    //       when `writeMessage` is called.
+    return;
+  }
+
+  writeCommand(
+    command: WriteProtocolMessageType,
+    operationDescription: OperationDescription
+  ): void {
+    const agreedCompressor = operationDescription.agreedCompressor ?? 'none';
+    if (agreedCompressor === 'none' || !OpCompressedRequest.canCompress(command)) {
+      const data = command.toBin();
+      this.push(Array.isArray(data) ? Buffer.concat(data) : data);
+      return;
+    }
+    // otherwise, compress the message
+    const concatenatedOriginalCommandBuffer = Buffer.concat(command.toBin());
+    const messageToBeCompressed = concatenatedOriginalCommandBuffer.slice(MESSAGE_HEADER_SIZE);
+
+    // Extract information needed for OP_COMPRESSED from the uncompressed message
+    const originalCommandOpCode = concatenatedOriginalCommandBuffer.readInt32LE(12);
+
+    const options = {
+      agreedCompressor,
+      zlibCompressionLevel: operationDescription.zlibCompressionLevel ?? 0
+    };
+    // Compress the message body
+    compress(options, messageToBeCompressed).then(
+      compressedMessage => {
+        // Create the msgHeader of OP_COMPRESSED
+        const msgHeader = Buffer.alloc(MESSAGE_HEADER_SIZE);
+        msgHeader.writeInt32LE(
+          MESSAGE_HEADER_SIZE + COMPRESSION_DETAILS_SIZE + compressedMessage.length,
+          0
+        ); // messageLength
+        msgHeader.writeInt32LE(command.requestId, 4); // requestID
+        msgHeader.writeInt32LE(0, 8); // responseTo (zero)
+        msgHeader.writeInt32LE(OP_COMPRESSED, 12); // opCode
+
+        // Create the compression details of OP_COMPRESSED
+        const compressionDetails = Buffer.alloc(COMPRESSION_DETAILS_SIZE);
+        compressionDetails.writeInt32LE(originalCommandOpCode, 0); // originalOpcode
+        compressionDetails.writeInt32LE(messageToBeCompressed.length, 4); // Size of the uncompressed compressedMessage, excluding the MsgHeader
+        compressionDetails.writeUInt8(Compressor[agreedCompressor], 8); // compressorID
+        this.push(Buffer.concat([msgHeader, compressionDetails, compressedMessage]));
+      },
+      error => {
+        operationDescription.cb(error);
+      }
+    );
+  }
+}
+
+function processIncomingData(stream: MessageStream, callback: Callback<Buffer>): void {
+  const buffer = stream[kBuffer];
+  const sizeOfMessage = buffer.getInt32();
+
+  if (sizeOfMessage == null) {
+    return callback();
+  }
+
+  if (sizeOfMessage < 0) {
+    return callback(new MongoParseError(`Invalid message size: ${sizeOfMessage}`));
+  }
+
+  if (sizeOfMessage > stream.maxBsonMessageSize) {
+    return callback(
+      new MongoParseError(
+        `Invalid message size: ${sizeOfMessage}, max allowed: ${stream.maxBsonMessageSize}`
+      )
+    );
+  }
+
+  if (sizeOfMessage > buffer.length) {
+    return callback();
+  }
+
+  const message = buffer.read(sizeOfMessage);
+  const messageHeader: MessageHeader = {
+    length: message.readInt32LE(0),
+    requestId: message.readInt32LE(4),
+    responseTo: message.readInt32LE(8),
+    opCode: message.readInt32LE(12)
+  };
+
+  const monitorHasAnotherHello = () => {
+    if (stream.isMonitoringConnection) {
+      // Can we read the next message size?
+      const sizeOfMessage = buffer.getInt32();
+      if (sizeOfMessage != null && sizeOfMessage <= buffer.length) {
+        return true;
+      }
+    }
+    return false;
+  };
+
+  let ResponseType = messageHeader.opCode === OP_MSG ? OpMsgResponse : OpQueryResponse;
+  if (messageHeader.opCode !== OP_COMPRESSED) {
+    const messageBody = message.subarray(MESSAGE_HEADER_SIZE);
+
+    // If we are a monitoring connection message stream and
+    // there is more in the buffer that can be read, skip processing since we
+    // want the last hello command response that is in the buffer.
+    if (monitorHasAnotherHello()) {
+      return processIncomingData(stream, callback);
+    }
+
+    stream.emit('message', new ResponseType(message, messageHeader, messageBody));
+
+    if (buffer.length >= 4) {
+      return processIncomingData(stream, callback);
+    }
+    return callback();
+  }
+
+  messageHeader.fromCompressed = true;
+  messageHeader.opCode = message.readInt32LE(MESSAGE_HEADER_SIZE);
+  messageHeader.length = message.readInt32LE(MESSAGE_HEADER_SIZE + 4);
+  const compressorID = message[MESSAGE_HEADER_SIZE + 8];
+  const compressedBuffer = message.slice(MESSAGE_HEADER_SIZE + 9);
+
+  // recalculate based on wrapped opcode
+  ResponseType = messageHeader.opCode === OP_MSG ? OpMsgResponse : OpQueryResponse;
+  decompress(compressorID, compressedBuffer).then(
+    messageBody => {
+      if (messageBody.length !== messageHeader.length) {
+        return callback(
+          new MongoDecompressionError('Message body and message header must be the same length')
+        );
+      }
+
+      // If we are a monitoring connection message stream and
+      // there is more in the buffer that can be read, skip processing since we
+      // want the last hello command response that is in the buffer.
+      if (monitorHasAnotherHello()) {
+        return processIncomingData(stream, callback);
+      }
+      stream.emit('message', new ResponseType(message, messageHeader, messageBody));
+
+      if (buffer.length >= 4) {
+        return processIncomingData(stream, callback);
+      }
+      return callback();
+    },
+    error => {
+      return callback(error);
+    }
+  );
+}

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;