feat(mongodb-runner): pass log events through to caller COMPASS-10017 (#593)

Author: addaleax

packages/mongodb-runner/src/index.ts

@@ -1,5 +1,13 @@
-export { MongoServer, MongoServerOptions } from './mongoserver';
-
-export { MongoCluster, MongoClusterOptions } from './mongocluster';
+export {
+  MongoServer,
+  type MongoServerEvents,
+  MongoServerOptions,
+} from './mongoserver';
+export {
+  MongoCluster,
+  type MongoClusterEvents,
+  MongoClusterOptions,
+} from './mongocluster';
+export type { LogEntry } from './mongologreader';
 export type { ConnectionString } from 'mongodb-connection-string-url';
 export { prune, start, stop, exec, instances } from './runner-helpers';

packages/mongodb-runner/src/mongocluster.spec.ts

@@ -5,6 +5,7 @@ import path from 'path';
 import os from 'os';
 import createDebug from 'debug';
 import sinon from 'sinon';
+import type { LogEntry } from './mongologreader';
 
 if (process.env.CI) {
   createDebug.enable('mongodb-runner,mongodb-downloader');
@@ -298,4 +299,48 @@ describe('MongoCluster', function () {
     expect(doc?._id).to.be.a('string');
     await cluster.close();
   });
+
+  it('can let callers listen for server log events', async function () {
+    cluster = await MongoCluster.start({
+      version: '6.x',
+      topology: 'replset',
+      tmpDir,
+      secondaries: 1,
+    });
+    const logs: LogEntry[] = [];
+    cluster.on('mongoLog', (uuid, entry) => logs.push(entry));
+    await cluster.withClient(async (client) => {
+      const coll = await client.db('test').createCollection<any>('test', {
+        validationAction: 'warn',
+        validationLevel: 'strict',
+        validator: {
+          $jsonSchema: {
+            bsonType: 'object',
+            required: ['phone'],
+            properties: {
+              phone: {
+                bsonType: 'string',
+              },
+            },
+          },
+        },
+      });
+      await coll.insertOne({ _id: 42, baddoc: 1 });
+    });
+    expect(
+      logs.find(
+        (entry) =>
+          entry.id === 20320 /* create collection */ &&
+          entry.attr.namespace === 'test.test',
+      ),
+    ).to.exist;
+    const validatorLogEntry = logs.find(
+      (entry) => entry.id === 20294 /* fail validation */,
+    );
+    expect(validatorLogEntry?.attr.namespace).to.equal('test.test');
+    expect(validatorLogEntry?.attr.document).to.deep.equal({
+      _id: 42,
+      baddoc: 1,
+    });
+  });
 });

packages/mongodb-runner/src/mongocluster.ts

@@ -1,4 +1,4 @@
-import type { MongoServerOptions } from './mongoserver';
+import type { MongoServerEvents, MongoServerOptions } from './mongoserver';
 import { MongoServer } from './mongoserver';
 import { ConnectionString } from 'mongodb-connection-string-url';
 import type { DownloadOptions } from '@mongodb-js/mongodb-downloader';
@@ -7,6 +7,7 @@ import type { MongoClientOptions } from 'mongodb';
 import { MongoClient } from 'mongodb';
 import { sleep, range, uuid, debug } from './util';
 import { OIDCMockProviderProcess } from './oidc';
+import { EventEmitter } from 'events';
 
 export interface MongoClusterOptions
   extends Pick<
@@ -23,14 +24,34 @@ export interface MongoClusterOptions
   oidc?: string;
 }
 
-export class MongoCluster {
+export type MongoClusterEvents = {
+  [k in keyof MongoServerEvents]: [serverUUID: string, ...MongoServerEvents[k]];
+} & {
+  newListener: [keyof MongoClusterEvents];
+  removeListener: [keyof MongoClusterEvents];
+};
+
+export class MongoCluster extends EventEmitter<MongoClusterEvents> {
   private topology: MongoClusterOptions['topology'] = 'standalone';
   private replSetName?: string;
   private servers: MongoServer[] = []; // mongod/mongos
   private shards: MongoCluster[] = []; // replsets
   private oidcMockProviderProcess?: OIDCMockProviderProcess;
 
   private constructor() {
+    super();
+    // NB: This will not retroactively add listeners to new server instances.
+    // This should be fine, as we only pass "fully initialized" clusters to
+    // callers, with all child clusters and individual servers already in place.
+    this.on('newListener', (name) => {
+      if (name === 'newListener' || name === 'removeListener') return;
+      if (this.listenerCount(name) === 0) {
+        for (const child of this.servers)
+          child.on(name, (...args) => this.emit(name, child.id, ...args));
+        for (const child of this.shards)
+          child.on(name, (...args) => this.emit(name, ...args));
+      }
+    });
     /* see .start() */
   }
 

packages/mongodb-runner/src/mongoserver.ts

@@ -12,7 +12,7 @@ import { Readable } from 'stream';
 import type { Document, MongoClientOptions } from 'mongodb';
 import { MongoClient } from 'mongodb';
 import path from 'path';
-import { once } from 'events';
+import { EventEmitter, once } from 'events';
 import { uuid, debug, pick, debugVerbose } from './util';
 
 export interface MongoServerOptions {
@@ -33,8 +33,12 @@ interface SerializedServerProperties {
   hasInsertedMetadataCollEntry: boolean;
 }
 
-export class MongoServer {
-  private uuid: string = uuid();
+export interface MongoServerEvents {
+  mongoLog: [LogEntry];
+}
+
+export class MongoServer extends EventEmitter<MongoServerEvents> {
+  public uuid: string = uuid();
   private buildInfo?: Document;
   private childProcess?: ChildProcess;
   private pid?: number;
@@ -44,7 +48,12 @@ export class MongoServer {
   private startTime = new Date().toISOString();
   private hasInsertedMetadataCollEntry = false;
 
+  get id(): string {
+    return this.uuid;
+  }
+
   private constructor() {
+    super();
     /* see .start() */
   }
 
@@ -212,7 +221,8 @@ export class MongoServer {
     const errorLogEntries: LogEntry[] = [];
     try {
       const logEntryStream = Readable.from(createLogEntryIterator(stdout));
-      logEntryStream.on('data', (entry) => {
+      logEntryStream.on('data', (entry: LogEntry) => {
+        srv.emit('mongoLog', entry);
         if (!srv.closing && ['E', 'F'].includes(entry.severity)) {
           errorLogEntries.push(entry);
           debug('mongodb server output', entry);