Skip to content

Commit b27f385

Browse files
W-A-JamesW-A-James
authored
feat(NODE-4720): Add log messages to CMAP spec (#3645)
1 parent 2264fbb commit b27f385

102 files changed

Lines changed: 1959 additions & 202 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎src/cmap/connection_pool.ts‎

Lines changed: 26 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -251,8 +251,11 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
251251
this[kMetrics] = new ConnectionPoolMetrics();
252252
this[kProcessingWaitQueue] = false;
253253

254+
this.mongoLogger = this[kServer].topology.client.mongoLogger;
255+
this.component = 'connection';
256+
254257
process.nextTick(() => {
255-
this.emit(ConnectionPool.CONNECTION_POOL_CREATED, new ConnectionPoolCreatedEvent(this));
258+
this.emitAndLog(ConnectionPool.CONNECTION_POOL_CREATED, new ConnectionPoolCreatedEvent(this));
256259
});
257260
}
258261

@@ -337,7 +340,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
337340
return;
338341
}
339342
this[kPoolState] = PoolState.ready;
340-
this.emit(ConnectionPool.CONNECTION_POOL_READY, new ConnectionPoolReadyEvent(this));
343+
this.emitAndLog(ConnectionPool.CONNECTION_POOL_READY, new ConnectionPoolReadyEvent(this));
341344
clearTimeout(this[kMinPoolSizeTimer]);
342345
this.ensureMinPoolSize();
343346
}
@@ -348,7 +351,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
348351
* explicitly destroyed by the new owner.
349352
*/
350353
checkOut(callback: Callback<Connection>): void {
351-
this.emit(
354+
this.emitAndLog(
352355
ConnectionPool.CONNECTION_CHECK_OUT_STARTED,
353356
new ConnectionCheckOutStartedEvent(this)
354357
);
@@ -360,7 +363,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
360363
waitQueueMember[kCancelled] = true;
361364
waitQueueMember.timer = undefined;
362365

363-
this.emit(
366+
this.emitAndLog(
364367
ConnectionPool.CONNECTION_CHECK_OUT_FAILED,
365368
new ConnectionCheckOutFailedEvent(this, 'timeout')
366369
);
@@ -398,7 +401,10 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
398401
}
399402

400403
this[kCheckedOut].delete(connection);
401-
this.emit(ConnectionPool.CONNECTION_CHECKED_IN, new ConnectionCheckedInEvent(this, connection));
404+
this.emitAndLog(
405+
ConnectionPool.CONNECTION_CHECKED_IN,
406+
new ConnectionCheckedInEvent(this, connection)
407+
);
402408

403409
if (willDestroy) {
404410
const reason = connection.closed ? 'error' : poolClosed ? 'poolClosed' : 'stale';
@@ -437,7 +443,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
437443
// Increment the generation for the service id.
438444
this.serviceGenerations.set(sid, generation + 1);
439445
}
440-
this.emit(
446+
this.emitAndLog(
441447
ConnectionPool.CONNECTION_POOL_CLEARED,
442448
new ConnectionPoolClearedEvent(this, { serviceId })
443449
);
@@ -452,9 +458,11 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
452458

453459
this.clearMinPoolSizeTimer();
454460
if (!alreadyPaused) {
455-
this.emit(
461+
this.emitAndLog(
456462
ConnectionPool.CONNECTION_POOL_CLEARED,
457-
new ConnectionPoolClearedEvent(this, { interruptInUseConnections })
463+
new ConnectionPoolClearedEvent(this, {
464+
interruptInUseConnections
465+
})
458466
);
459467
}
460468

@@ -509,15 +517,15 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
509517
eachAsync<Connection>(
510518
this[kConnections].toArray(),
511519
(conn, cb) => {
512-
this.emit(
520+
this.emitAndLog(
513521
ConnectionPool.CONNECTION_CLOSED,
514522
new ConnectionClosedEvent(this, conn, 'poolClosed')
515523
);
516524
conn.destroy({ force: !!options.force }, cb);
517525
},
518526
err => {
519527
this[kConnections].clear();
520-
this.emit(ConnectionPool.CONNECTION_POOL_CLOSED, new ConnectionPoolClosedEvent(this));
528+
this.emitAndLog(ConnectionPool.CONNECTION_POOL_CLOSED, new ConnectionPoolClosedEvent(this));
521529
callback(err);
522530
}
523531
);
@@ -645,7 +653,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
645653
connection: Connection,
646654
reason: 'error' | 'idle' | 'stale' | 'poolClosed'
647655
) {
648-
this.emit(
656+
this.emitAndLog(
649657
ConnectionPool.CONNECTION_CLOSED,
650658
new ConnectionClosedEvent(this, connection, reason)
651659
);
@@ -694,15 +702,15 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
694702

695703
this[kPending]++;
696704
// This is our version of a "virtual" no-I/O connection as the spec requires
697-
this.emit(
705+
this.emitAndLog(
698706
ConnectionPool.CONNECTION_CREATED,
699707
new ConnectionCreatedEvent(this, { id: connectOptions.id })
700708
);
701709

702710
connect(connectOptions, (err, connection) => {
703711
if (err || !connection) {
704712
this[kPending]--;
705-
this.emit(
713+
this.emitAndLog(
706714
ConnectionPool.CONNECTION_CLOSED,
707715
new ConnectionClosedEvent(
708716
this,
@@ -750,7 +758,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
750758
}
751759

752760
connection.markAvailable();
753-
this.emit(ConnectionPool.CONNECTION_READY, new ConnectionReadyEvent(this, connection));
761+
this.emitAndLog(ConnectionPool.CONNECTION_READY, new ConnectionReadyEvent(this, connection));
754762

755763
this[kPending]--;
756764
callback(undefined, connection);
@@ -819,7 +827,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
819827
if (this[kPoolState] !== PoolState.ready) {
820828
const reason = this.closed ? 'poolClosed' : 'connectionError';
821829
const error = this.closed ? new PoolClosedError(this) : new PoolClearedError(this);
822-
this.emit(
830+
this.emitAndLog(
823831
ConnectionPool.CONNECTION_CHECK_OUT_FAILED,
824832
new ConnectionCheckOutFailedEvent(this, reason, error)
825833
);
@@ -842,7 +850,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
842850

843851
if (!this.destroyConnectionIfPerished(connection)) {
844852
this[kCheckedOut].add(connection);
845-
this.emit(
853+
this.emitAndLog(
846854
ConnectionPool.CONNECTION_CHECKED_OUT,
847855
new ConnectionCheckedOutEvent(this, connection)
848856
);
@@ -872,14 +880,14 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
872880
}
873881
} else {
874882
if (err) {
875-
this.emit(
883+
this.emitAndLog(
876884
ConnectionPool.CONNECTION_CHECK_OUT_FAILED,
877885
// TODO(NODE-5192): Remove this cast
878886
new ConnectionCheckOutFailedEvent(this, 'connectionError', err as MongoError)
879887
);
880888
} else if (connection) {
881889
this[kCheckedOut].add(connection);
882-
this.emit(
890+
this.emitAndLog(
883891
ConnectionPool.CONNECTION_CHECKED_OUT,
884892
new ConnectionCheckedOutEvent(this, connection)
885893
);

‎src/mongo_client.ts‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -325,7 +325,7 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
325325
/** @internal */
326326
topology?: Topology;
327327
/** @internal */
328-
readonly mongoLogger: MongoLogger;
328+
override readonly mongoLogger: MongoLogger;
329329
/** @internal */
330330
private connectionLock?: Promise<this>;
331331

@@ -471,23 +471,21 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
471471
}
472472
}
473473

474-
const topology = new Topology(options.hosts, options);
474+
this.topology = new Topology(this, options.hosts, options);
475475
// Events can be emitted before initialization is complete so we have to
476476
// save the reference to the topology on the client ASAP if the event handlers need to access it
477-
this.topology = topology;
478-
topology.client = this;
479477

480-
topology.once(Topology.OPEN, () => this.emit('open', this));
478+
this.topology.once(Topology.OPEN, () => this.emit('open', this));
481479

482480
for (const event of MONGO_CLIENT_EVENTS) {
483-
topology.on(event, (...args: any[]) => this.emit(event, ...(args as any)));
481+
this.topology.on(event, (...args: any[]) => this.emit(event, ...(args as any)));
484482
}
485483

486484
const topologyConnect = async () => {
487485
try {
488-
await promisify(callback => topology.connect(options, callback))();
486+
await promisify(callback => this.topology?.connect(options, callback))();
489487
} catch (error) {
490-
topology.close({ force: true });
488+
this.topology?.close({ force: true });
491489
throw error;
492490
}
493491
};

‎src/mongo_types.ts‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import type {
1212
ObjectId,
1313
Timestamp
1414
} from './bson';
15+
import type { MongoLoggableComponent, MongoLogger } from './mongo_logger';
1516
import type { Sort } from './sort';
1617

1718
/** @internal */
@@ -397,8 +398,21 @@ export declare interface TypedEventEmitter<Events extends EventsDescription> ext
397398
* Typescript type safe event emitter
398399
* @public
399400
*/
400-
// eslint-disable-next-line @typescript-eslint/no-unused-vars
401-
export class TypedEventEmitter<Events extends EventsDescription> extends EventEmitter {}
401+
402+
export class TypedEventEmitter<Events extends EventsDescription> extends EventEmitter {
403+
/** @internal */
404+
protected mongoLogger?: MongoLogger;
405+
/** @internal */
406+
protected component?: MongoLoggableComponent;
407+
/** @internal */
408+
protected emitAndLog<EventKey extends keyof Events>(
409+
event: EventKey | symbol,
410+
...args: Parameters<Events[EventKey]>
411+
): void {
412+
this.emit(event, ...args);
413+
if (this.component) this.mongoLogger?.debug(this.component, args[0]);
414+
}
415+
}
402416

403417
/** @public */
404418
export class CancellationToken extends TypedEventEmitter<{ cancel(): void }> {}

‎src/sdam/monitor.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,7 @@ export class Monitor extends TypedEventEmitter<MonitorEvents> {
111111
const connectOptions = Object.assign(
112112
{
113113
id: '<monitor>' as const,
114-
generation: server.s.pool.generation,
114+
generation: server.pool.generation,
115115
connectionType: Connection,
116116
cancellationToken,
117117
hostAddress: server.description.hostAddress

‎src/sdam/server.ts‎

Lines changed: 21 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -87,10 +87,6 @@ export interface ServerPrivate {
8787
options: ServerOptions;
8888
/** The current state of the Server */
8989
state: string;
90-
/** The topology this server is a part of */
91-
topology: Topology;
92-
/** A connection pool for this server */
93-
pool: ConnectionPool;
9490
/** MongoDB server API version */
9591
serverApi?: ServerApi;
9692
/** A count of the operations currently running against the server. */
@@ -114,6 +110,10 @@ export type ServerEvents = {
114110
export class Server extends TypedEventEmitter<ServerEvents> {
115111
/** @internal */
116112
s: ServerPrivate;
113+
/** @internal */
114+
topology: Topology;
115+
/** @internal */
116+
pool: ConnectionPool;
117117
serverApi?: ServerApi;
118118
hello?: Document;
119119
[kMonitor]: Monitor | null;
@@ -143,20 +143,21 @@ export class Server extends TypedEventEmitter<ServerEvents> {
143143

144144
const poolOptions = { hostAddress: description.hostAddress, ...options };
145145

146+
this.topology = topology;
147+
this.pool = new ConnectionPool(this, poolOptions);
148+
146149
this.s = {
147150
description,
148151
options,
149152
state: STATE_CLOSED,
150-
topology,
151-
pool: new ConnectionPool(this, poolOptions),
152153
operationCount: 0
153154
};
154155

155156
for (const event of [...CMAP_EVENTS, ...APM_EVENTS]) {
156-
this.s.pool.on(event, (e: any) => this.emit(event, e));
157+
this.pool.on(event, (e: any) => this.emit(event, e));
157158
}
158159

159-
this.s.pool.on(Connection.CLUSTER_TIME_RECEIVED, (clusterTime: ClusterTime) => {
160+
this.pool.on(Connection.CLUSTER_TIME_RECEIVED, (clusterTime: ClusterTime) => {
160161
this.clusterTime = clusterTime;
161162
});
162163

@@ -192,11 +193,11 @@ export class Server extends TypedEventEmitter<ServerEvents> {
192193
}
193194

194195
get clusterTime(): ClusterTime | undefined {
195-
return this.s.topology.clusterTime;
196+
return this.topology.clusterTime;
196197
}
197198

198199
set clusterTime(clusterTime: ClusterTime | undefined) {
199-
this.s.topology.clusterTime = clusterTime;
200+
this.topology.clusterTime = clusterTime;
200201
}
201202

202203
get description(): ServerDescription {
@@ -215,7 +216,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
215216
}
216217

217218
get loadBalanced(): boolean {
218-
return this.s.topology.description.type === TopologyType.LoadBalanced;
219+
return this.topology.description.type === TopologyType.LoadBalanced;
219220
}
220221

221222
/**
@@ -261,7 +262,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
261262
this[kMonitor]?.close();
262263
}
263264

264-
this.s.pool.close(options, err => {
265+
this.pool.close(options, err => {
265266
stateTransition(this, STATE_CLOSED);
266267
this.emit('closed');
267268
if (typeof callback === 'function') {
@@ -330,7 +331,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
330331
// balanced code makes a recursive call). Instead, we increment the count after this
331332
// check.
332333
if (this.loadBalanced && session && conn == null && isPinnableCommand(cmd, session)) {
333-
this.s.pool.checkOut((err, checkedOut) => {
334+
this.pool.checkOut((err, checkedOut) => {
334335
if (err || checkedOut == null) {
335336
if (callback) return callback(err);
336337
return;
@@ -344,7 +345,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
344345

345346
this.s.operationCount += 1;
346347

347-
this.s.pool.withConnection(
348+
this.pool.withConnection(
348349
conn,
349350
(err, conn, cb) => {
350351
if (err || !conn) {
@@ -382,7 +383,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
382383
}
383384

384385
const isStaleError =
385-
error.connectionGeneration && error.connectionGeneration < this.s.pool.generation;
386+
error.connectionGeneration && error.connectionGeneration < this.pool.generation;
386387
if (isStaleError) {
387388
return;
388389
}
@@ -398,14 +399,14 @@ export class Server extends TypedEventEmitter<ServerEvents> {
398399
error.addErrorLabel(MongoErrorLabel.ResetPool);
399400
markServerUnknown(this, error);
400401
} else if (connection) {
401-
this.s.pool.clear({ serviceId: connection.serviceId });
402+
this.pool.clear({ serviceId: connection.serviceId });
402403
}
403404
} else {
404405
if (isSDAMUnrecoverableError(error)) {
405406
if (shouldHandleStateChangeError(this, error)) {
406407
const shouldClearPool = maxWireVersion(this) <= 7 || isNodeShuttingDownError(error);
407408
if (this.loadBalanced && connection && shouldClearPool) {
408-
this.s.pool.clear({ serviceId: connection.serviceId });
409+
this.pool.clear({ serviceId: connection.serviceId });
409410
}
410411

411412
if (!this.loadBalanced) {
@@ -514,7 +515,7 @@ function makeOperationHandler(
514515
return callback(error);
515516
}
516517

517-
if (connectionIsStale(server.s.pool, connection)) {
518+
if (connectionIsStale(server.pool, connection)) {
518519
return callback(error);
519520
}
520521

@@ -532,15 +533,15 @@ function makeOperationHandler(
532533
}
533534

534535
if (
535-
(isRetryableWritesEnabled(server.s.topology) || isTransactionCommand(cmd)) &&
536+
(isRetryableWritesEnabled(server.topology) || isTransactionCommand(cmd)) &&
536537
supportsRetryableWrites(server) &&
537538
!inActiveTransaction(session, cmd)
538539
) {
539540
error.addErrorLabel(MongoErrorLabel.RetryableWriteError);
540541
}
541542
} else {
542543
if (
543-
(isRetryableWritesEnabled(server.s.topology) || isTransactionCommand(cmd)) &&
544+
(isRetryableWritesEnabled(server.topology) || isTransactionCommand(cmd)) &&
544545
needsRetryableWriteLabel(error, maxWireVersion(server)) &&
545546
!inActiveTransaction(session, cmd)
546547
) {

0 commit comments

Comments
 (0)