Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ lib/
*.d.ts
# type definition tests
!test/types
!global.d.ts

.vscode
output
Expand Down
1 change: 1 addition & 0 deletions .mocharc.json
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,6 @@
"recursive": true,
"timeout": 60000,
"reporter": "test/tools/reporter/mongodb_reporter.js",
"sort": true,
"color": true
}
241 changes: 241 additions & 0 deletions etc/sdam_viz.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,241 @@
/* eslint-disable no-console */
/* eslint-disable @typescript-eslint/no-var-requires */

// run this file with ts-node:
// npx ts-node etc/sdam_viz.js -h

const { MongoClient } = require('../src');
const { now, calculateDurationInMs, arrayStrictEqual, errorStrictEqual } = require('../src/utils');

const util = require('util');
const chalk = require('chalk');
const argv = require('yargs')
.usage('Usage: $0 [options] <connection string>')
.demandCommand(1)
.help('h')
.describe('workload', 'Simulate a read workload')
.describe('writeWorkload', 'Simulate a write workload')
.describe('writeWorkloadInterval', 'Time interval between write workload write attempts')
.describe('writeWorkloadSampleSize', 'Sample size between status display for write workload')
.describe('legacy', 'Use the legacy topology types')
.alias('l', 'legacy')
.alias('w', 'workload')
.alias('h', 'help').argv;

function print(msg) {
console.log(`${chalk.white(new Date().toISOString())} ${msg}`);
}

const uri = argv._[0];
const client = new MongoClient(uri);

function diff(lhs, rhs, fields, comparator) {
return fields.reduce((diff, field) => {
if ((lhs[field] == null || rhs[field] == null) && field !== 'error') {
return diff;
}

if (!comparator(lhs[field], rhs[field])) {
diff.push(
` ${field}: ${chalk.green(`${util.inspect(lhs[field])}`)} => ${chalk.green(
`${util.inspect(rhs[field])}`
)}`
);
}

return diff;
}, []);
}

function serverDescriptionDiff(lhs, rhs) {
const objectIdFields = ['electionId'];
const arrayFields = ['hosts', 'tags'];
const simpleFields = [
'type',
'minWireVersion',
'me',
'setName',
'setVersion',
'electionId',
'primary',
'logicalSessionTimeoutMinutes'
];

return diff(lhs, rhs, simpleFields, (x, y) => x === y)
.concat(diff(lhs, rhs, ['error'], (x, y) => errorStrictEqual(x, y)))
.concat(diff(lhs, rhs, arrayFields, (x, y) => arrayStrictEqual(x, y)))
.concat(diff(lhs, rhs, objectIdFields, (x, y) => x.equals(y)))
.join(',\n');
}

function topologyDescriptionDiff(lhs, rhs) {
const simpleFields = [
'type',
'setName',
'maxSetVersion',
'stale',
'compatible',
'compatibilityError',
'logicalSessionTimeoutMinutes',
'error',
'commonWireVersion'
];

return diff(lhs, rhs, simpleFields, (x, y) => x === y).join(',\n');
}

function visualizeMonitoringEvents(client) {
function print(msg) {
console.error(`${chalk.white(new Date().toISOString())} ${msg}`);
}

client.on('serverHeartbeatStarted', event =>
print(`${chalk.yellow('heartbeat')} ${chalk.bold('started')} host: '${event.connectionId}`)
);

client.on('serverHeartbeatSucceeded', event =>
print(
`${chalk.yellow('heartbeat')} ${chalk.green('succeeded')} host: '${
event.connectionId
}' ${chalk.gray(`(${event.duration} ms)`)}`
)
);

client.on('serverHeartbeatFailed', event =>
print(
`${chalk.yellow('heartbeat')} ${chalk.red('failed')} host: '${
event.connectionId
}' ${chalk.gray(`(${event.duration} ms)`)}`
)
);

// server information
client.on('serverOpening', event => {
print(
`${chalk.cyan('server')} [${event.address}] ${chalk.bold('opening')} in topology#${
event.topologyId
}`
);
});

client.on('serverClosed', event => {
print(
`${chalk.cyan('server')} [${event.address}] ${chalk.bold('closed')} in topology#${
event.topologyId
}`
);
});

client.on('serverDescriptionChanged', event => {
print(`${chalk.cyan('server')} [${event.address}] changed:`);
console.error(serverDescriptionDiff(event.previousDescription, event.newDescription));
});

// topology information
client.on('topologyOpening', event => {
print(`${chalk.magenta('topology')} adding topology#${event.topologyId}`);
});

client.on('topologyClosed', event => {
print(`${chalk.magenta('topology')} removing topology#${event.topologyId}`);
});

client.on('topologyDescriptionChanged', event => {
const diff = topologyDescriptionDiff(event.previousDescription, event.newDescription);
if (diff !== '') {
print(`${chalk.magenta('topology')} [topology#${event.topologyId}] changed:`);
console.error(diff);
}
});
}

async function run() {
print(`connecting to: ${chalk.bold(uri)}`);

visualizeMonitoringEvents(client);
await client.connect();

if (argv.workload) {
scheduleWorkload(client);
}

if (argv.writeWorkload) {
scheduleWriteWorkload(client);
}
}

let workloadTimer;
let workloadCounter = 0;
let workloadInterrupt = false;
async function scheduleWorkload(client) {
if (!workloadInterrupt) {
// immediately reschedule work
workloadTimer = setTimeout(() => scheduleWorkload(client), 7000);
}

const currentWorkload = workloadCounter++;

try {
print(`${chalk.yellow(`workload#${currentWorkload}`)} issuing find...`);
const result = await client
.db('test')
.collection('test')
.find({}, { socketTimeoutMS: 2000 })
.limit(1)
.toArray();

print(
`${chalk.yellow(`workload#${currentWorkload}`)} find completed: ${JSON.stringify(result)}`
);
} catch (e) {
print(`${chalk.yellow(`workload#${currentWorkload}`)} find failed: ${e.message}`);
}
}

let writeWorkloadTimer;
let writeWorkloadCounter = 0;
let averageWriteMS = 0;
let completedWriteWorkloads = 0;
const writeWorkloadSampleSize = argv.writeWorkloadSampleSize || 100;
const writeWorkloadInterval = argv.writeWorkloadInterval || 100;
async function scheduleWriteWorkload(client) {
if (!workloadInterrupt) {
// immediately reschedule work
writeWorkloadTimer = setTimeout(() => scheduleWriteWorkload(client), writeWorkloadInterval);
}

const currentWriteWorkload = writeWorkloadCounter++;

try {
const start = now();
await client.db('test').collection('test').insertOne({ a: 42 });
averageWriteMS = 0.2 * calculateDurationInMs(start) + 0.8 * averageWriteMS;

completedWriteWorkloads++;
if (completedWriteWorkloads % writeWorkloadSampleSize === 0) {
print(
`${chalk.yellow(
`workload#${currentWriteWorkload}`
)} completed ${completedWriteWorkloads} writes with average time: ${averageWriteMS}`
);
}
} catch (e) {
print(`${chalk.yellow(`workload#${currentWriteWorkload}`)} write failed: ${e.message}`);
}
}

let exitRequestCount = 0;
process.on('SIGINT', async function () {
exitRequestCount++;
if (exitRequestCount > 3) {
console.log('force quitting...');
process.exit(1);
}

workloadInterrupt = true;
clearTimeout(workloadTimer);
clearTimeout(writeWorkloadTimer);
await client.close();
});

run().catch(error => console.log('Caught', error));
65 changes: 65 additions & 0 deletions global.d.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
import type { TestConfiguration } from './test/tools/unified-spec-runner/runner';

/** Defined in test/tools/runner/filters/mongodb_topology_filter.js (topologyTypeToString) */
type TopologyTypes = 'single' | 'replicaset' | 'sharded' | 'load-balanced';

interface MongoDBMetadataUI {
requires?: {
topology?: TopologyTypes | TopologyTypes[];
mongodb?: string;
os?: NodeJS.Platform | `!${NodeJS.Platform}`;
apiVersion?: '1';
clientSideEncryption?: boolean;
serverless?: 'forbid' | 'allow' | 'require';
};

sessions?: {
skipLeakTests?: boolean;
};
}

interface MetadataAndTest<Fn> {
metadata: MongoDBMetadataUI;
test: Fn;
}

declare global {
namespace Mocha {
interface TestFunction {
(title: string, metadata: MongoDBMetadataUI, fn: Mocha.Func): Mocha.Test;
(title: string, metadata: MongoDBMetadataUI, fn: Mocha.AsyncFunc): Mocha.Test;

(title: string, testAndMetadata: MetadataAndTest<Mocha.Func>): Mocha.Test;
(title: string, testAndMetadata: MetadataAndTest<Mocha.AsyncFunc>): Mocha.Test;
}

interface Context {
configuration: TestConfiguration;
}

interface Test {
metadata: MongoDBMetadataUI;
}

interface Runnable {
/**
* An optional string the test author can attach to print out why a test is skipped
*
* @example
* ```
* it.skip('my test', () => {
* //...
* }).skipReason = 'TODO(NODE-XXXX): Feature implementation impending!';
* ```
*
* The reporter (`test/tools/reporter/mongodb_reporter.js`) will print out the skipReason
* indented directly below the test name.
* ```
* - my test
* - TODO(NODE-XXXX): Feature implementation impending!
* ```
*/
skipReason?: string;
}
}
}
6 changes: 3 additions & 3 deletions test/functional/unit-sdam/monitoring.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ describe('monitoring', function () {
return mock.createServer().then(server => (mockServer = server));
});

// TODO: NODE-3819: Unskip flaky tests.
// TODO(NODE-3819): Unskip flaky tests.
it.skip('should record roundTripTime', function (done) {
mockServer.setMessageHandler(request => {
const doc = request.document;
Expand All @@ -49,7 +49,7 @@ describe('monitoring', function () {
topology.close(done);
}, 500);
});
});
}).skipReason = 'TODO(NODE-3819): Unskip flaky tests';

// TODO(NODE-3600): Unskip flaky test
it.skip('should recover on error during initial connect', function (done) {
Expand Down Expand Up @@ -88,7 +88,7 @@ describe('monitoring', function () {

topology.close(done);
});
});
}).skipReason = 'TODO(NODE-3600): Unskip flaky tests';

describe('Monitor', function () {
it('should connect and issue an initial server check', function (done) {
Expand Down
Loading