Skip to content

Commit

Permalink
feat(NODE-3424): use hello for monitoring commands (#2964)
Browse files Browse the repository at this point in the history
  • Loading branch information
nbbeeken committed Sep 13, 2021
1 parent 44df7d7 commit 910c564
Show file tree
Hide file tree
Showing 412 changed files with 5,589 additions and 1,057 deletions.
2 changes: 1 addition & 1 deletion .evergreen/install-dependencies.sh
Expand Up @@ -3,7 +3,7 @@
set -o errexit # Exit the script with error if any of the commands fail

NVM_WINDOWS_URL="https://github.com/coreybutler/nvm-windows/releases/download/1.1.7/nvm-noinstall.zip"
NVM_URL="https://raw.githubusercontent.com/nvm-sh/nvm/v0.35.3/install.sh"
NVM_URL="https://raw.githubusercontent.com/nvm-sh/nvm/v0.38.0/install.sh"

NODE_LTS_NAME=${NODE_LTS_NAME:-carbon}
MSVS_VERSION=${MSVS_VERSION:-2017}
Expand Down
5 changes: 3 additions & 2 deletions lib/cmap/connection_pool.js
Expand Up @@ -158,12 +158,13 @@ class ConnectionPool extends EventEmitter {
waitQueueTimeoutMS:
typeof options.waitQueueTimeoutMS === 'number' ? options.waitQueueTimeoutMS : 0,
autoEncrypter: options.autoEncrypter,
metadata: options.metadata
metadata: options.metadata,
useUnifiedTopology: options.useUnifiedTopology
});

if (options.minSize > options.maxSize) {
throw new TypeError(
'Connection pool minimum size must not be greater than maxiumum pool size'
'Connection pool minimum size must not be greater than maximum pool size'
);
}

Expand Down
19 changes: 15 additions & 4 deletions lib/core/connection/connect.js
Expand Up @@ -95,6 +95,8 @@ function performInitialHandshake(conn, options, _callback) {
handshakeOptions.socketTimeout = options.connectTimeoutMS || options.connectionTimeout;
}

handshakeDoc.helloOk = !!options.useUnifiedTopology;

const start = new Date().getTime();
conn.command('admin.$cmd', handshakeDoc, handshakeOptions, (err, result) => {
if (err) {
Expand All @@ -113,6 +115,10 @@ function performInitialHandshake(conn, options, _callback) {
response.ismaster = response.isWritablePrimary;
}

if (options.useUnifiedTopology && response.helloOk) {
conn.helloOk = true;
}

const supportedServerErr = checkSupportedServer(response, options);
if (supportedServerErr) {
callback(supportedServerErr);
Expand Down Expand Up @@ -272,12 +278,17 @@ function makeConnection(family, options, cancellationToken, _callback) {
: typeof options.connectTimeoutMS === 'number'
? options.connectTimeoutMS
: 30000;
const socketTimeout = typeof options.socketTimeout === 'number' ? options.socketTimeout : 0;
const socketTimeoutMS =
typeof options.socketTimeoutMS === 'number'
? options.socketTimeoutMS
: typeof options.socketTimeout === 'number'
? options.socketTimeout
: 0;
const rejectUnauthorized =
typeof options.rejectUnauthorized === 'boolean' ? options.rejectUnauthorized : true;

if (keepAliveInitialDelay > socketTimeout) {
keepAliveInitialDelay = Math.round(socketTimeout / 2);
if (keepAliveInitialDelay > socketTimeoutMS) {
keepAliveInitialDelay = Math.round(socketTimeoutMS / 2);
}

let socket;
Expand Down Expand Up @@ -330,7 +341,7 @@ function makeConnection(family, options, cancellationToken, _callback) {
return callback(socket.authorizationError);
}

socket.setTimeout(socketTimeout);
socket.setTimeout(socketTimeoutMS);
callback(null, socket);
}

Expand Down
1 change: 1 addition & 0 deletions lib/core/connection/connection.js
Expand Up @@ -91,6 +91,7 @@ class Connection extends EventEmitter {
this.bson = options.bson;
this.tag = options.tag;
this.maxBsonMessageSize = options.maxBsonMessageSize || DEFAULT_MAX_BSON_MESSAGE_SIZE;
this.helloOk = undefined;

this.port = options.port || 27017;
this.host = options.host || 'localhost';
Expand Down
92 changes: 46 additions & 46 deletions lib/core/error.js
@@ -1,5 +1,7 @@
'use strict';

const MONGODB_ERROR_CODES = require('../error_codes').MONGODB_ERROR_CODES;

const kErrorLabels = Symbol('errorLabels');

/**
Expand Down Expand Up @@ -216,32 +218,32 @@ class MongoWriteConcernError extends MongoError {

// see: https://github.com/mongodb/specifications/blob/master/source/retryable-writes/retryable-writes.rst#terms
const RETRYABLE_ERROR_CODES = new Set([
6, // HostUnreachable
7, // HostNotFound
89, // NetworkTimeout
91, // ShutdownInProgress
189, // PrimarySteppedDown
9001, // SocketException
10107, // NotMaster
11600, // InterruptedAtShutdown
11602, // InterruptedDueToReplStateChange
13435, // NotMasterNoSlaveOk
13436 // NotMasterOrSecondary
MONGODB_ERROR_CODES.HostUnreachable,
MONGODB_ERROR_CODES.HostNotFound,
MONGODB_ERROR_CODES.NetworkTimeout,
MONGODB_ERROR_CODES.ShutdownInProgress,
MONGODB_ERROR_CODES.PrimarySteppedDown,
MONGODB_ERROR_CODES.SocketException,
MONGODB_ERROR_CODES.NotMaster,
MONGODB_ERROR_CODES.InterruptedAtShutdown,
MONGODB_ERROR_CODES.InterruptedDueToReplStateChange,
MONGODB_ERROR_CODES.NotMasterNoSlaveOk,
MONGODB_ERROR_CODES.NotMasterOrSecondary
]);

const RETRYABLE_WRITE_ERROR_CODES = new Set([
11600, // InterruptedAtShutdown
11602, // InterruptedDueToReplStateChange
10107, // NotMaster
13435, // NotMasterNoSlaveOk
13436, // NotMasterOrSecondary
189, // PrimarySteppedDown
91, // ShutdownInProgress
7, // HostNotFound
6, // HostUnreachable
89, // NetworkTimeout
9001, // SocketException
262 // ExceededTimeLimit
MONGODB_ERROR_CODES.InterruptedAtShutdown,
MONGODB_ERROR_CODES.InterruptedDueToReplStateChange,
MONGODB_ERROR_CODES.NotMaster,
MONGODB_ERROR_CODES.NotMasterNoSlaveOk,
MONGODB_ERROR_CODES.NotMasterOrSecondary,
MONGODB_ERROR_CODES.PrimarySteppedDown,
MONGODB_ERROR_CODES.ShutdownInProgress,
MONGODB_ERROR_CODES.HostNotFound,
MONGODB_ERROR_CODES.HostUnreachable,
MONGODB_ERROR_CODES.NetworkTimeout,
MONGODB_ERROR_CODES.SocketException,
MONGODB_ERROR_CODES.ExceededTimeLimit
]);

function isRetryableWriteError(error) {
Expand Down Expand Up @@ -271,41 +273,44 @@ function isRetryableError(error) {
}

const SDAM_RECOVERING_CODES = new Set([
91, // ShutdownInProgress
189, // PrimarySteppedDown
11600, // InterruptedAtShutdown
11602, // InterruptedDueToReplStateChange
13436 // NotMasterOrSecondary
MONGODB_ERROR_CODES.ShutdownInProgress,
MONGODB_ERROR_CODES.PrimarySteppedDown,
MONGODB_ERROR_CODES.InterruptedAtShutdown,
MONGODB_ERROR_CODES.InterruptedDueToReplStateChange,
MONGODB_ERROR_CODES.NotMasterOrSecondary
]);

const SDAM_NOTMASTER_CODES = new Set([
10107, // NotMaster
13435 // NotMasterNoSlaveOk
MONGODB_ERROR_CODES.NotMaster,
MONGODB_ERROR_CODES.NotMasterNoSlaveOk,
MONGODB_ERROR_CODES.LegacyNotPrimary
]);

const SDAM_NODE_SHUTTING_DOWN_ERROR_CODES = new Set([
11600, // InterruptedAtShutdown
91 // ShutdownInProgress
MONGODB_ERROR_CODES.InterruptedAtShutdown,
MONGODB_ERROR_CODES.ShutdownInProgress
]);

function isRecoveringError(err) {
if (err.code && SDAM_RECOVERING_CODES.has(err.code)) {
return true;
if (typeof err.code === 'number') {
// If any error code exists, we ignore the error.message
return SDAM_RECOVERING_CODES.has(err.code);
}

return err.message.match(/not master or secondary/) || err.message.match(/node is recovering/);
return /not master or secondary/.test(err.message) || /node is recovering/.test(err.message);
}

function isNotMasterError(err) {
if (err.code && SDAM_NOTMASTER_CODES.has(err.code)) {
return true;
if (typeof err.code === 'number') {
// If any error code exists, we ignore the error.message
return SDAM_NOTMASTER_CODES.has(err.code);
}

if (isRecoveringError(err)) {
return false;
}

return err.message.match(/not master/);
return /not master/.test(err.message);
}

function isNodeShuttingDownError(err) {
Expand All @@ -316,10 +321,9 @@ function isNodeShuttingDownError(err) {
* Determines whether SDAM can recover from a given error. If it cannot
* then the pool will be cleared, and server state will completely reset
* locally.
*
* @ignore
* @see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#not-master-and-node-is-recovering
* @param {MongoError|Error} error
* @param {MongoError} error
* @returns {boolean}
*/
function isSDAMUnrecoverableError(error) {
// NOTE: null check is here for a strictly pre-CMAP world, a timeout or
Expand All @@ -328,11 +332,7 @@ function isSDAMUnrecoverableError(error) {
return true;
}

if (isRecoveringError(error) || isNotMasterError(error)) {
return true;
}

return false;
return isRecoveringError(error) || isNotMasterError(error);
}

module.exports = {
Expand Down
12 changes: 10 additions & 2 deletions lib/core/sdam/monitor.js
Expand Up @@ -65,7 +65,8 @@ class Monitor extends EventEmitter {
heartbeatFrequencyMS:
typeof options.heartbeatFrequencyMS === 'number' ? options.heartbeatFrequencyMS : 10000,
minHeartbeatFrequencyMS:
typeof options.minHeartbeatFrequencyMS === 'number' ? options.minHeartbeatFrequencyMS : 500
typeof options.minHeartbeatFrequencyMS === 'number' ? options.minHeartbeatFrequencyMS : 500,
useUnifiedTopology: options.useUnifiedTopology
});

// TODO: refactor this to pull it directly from the pool, requires new ConnectionPool integration
Expand Down Expand Up @@ -205,8 +206,15 @@ function checkServer(monitor, callback) {
const topologyVersion = monitor[kServer].description.topologyVersion;
const isAwaitable = topologyVersion != null;
const serverApi = monitor[kConnection].serverApi;
const helloOk = monitor[kConnection].helloOk;

const cmd = {
[serverApi || helloOk ? 'hello' : 'ismaster']: true
};

// written this way omit helloOk from the command if its false-y (do not want -> helloOk: null)
if (helloOk) cmd.helloOk = helloOk;

const cmd = { [serverApi ? 'hello' : 'ismaster']: true };
const options = { socketTimeout: connectTimeoutMS };

if (isAwaitable) {
Expand Down
4 changes: 4 additions & 0 deletions lib/core/sdam/server_description.js
Expand Up @@ -70,6 +70,10 @@ class ServerDescription {
ismaster
);

if (ismaster.isWritablePrimary != null) {
ismaster.ismaster = ismaster.isWritablePrimary;
}

this.address = address;
this.error = options.error;
this.roundTripTime = options.roundTripTime || -1;
Expand Down
8 changes: 5 additions & 3 deletions lib/core/topologies/replset_state.js
Expand Up @@ -34,7 +34,7 @@ var ReplSetState = function(options) {
// Add event listener
EventEmitter.call(this);
// Topology state
this.topologyType = TopologyType.ReplicaSetNoPrimary;
this.topologyType = options.setName ? TopologyType.ReplicaSetNoPrimary : TopologyType.Unknown;
this.setName = options.setName;

// Server set
Expand Down Expand Up @@ -218,7 +218,8 @@ const isArbiter = ismaster => ismaster.arbiterOnly && ismaster.setName;
ReplSetState.prototype.update = function(server) {
var self = this;
// Get the current ismaster
var ismaster = server.lastIsMaster();
const ismaster = server.lastIsMaster();
if (ismaster && ismaster.isWritablePrimary) ismaster.ismaster = ismaster.isWritablePrimary;

// Get the server name and lowerCase it
var serverName = server.name.toLowerCase();
Expand Down Expand Up @@ -358,7 +359,8 @@ ReplSetState.prototype.update = function(server) {
// Standalone server, destroy and return
//
if (ismaster && ismaster.ismaster && !ismaster.setName) {
this.topologyType = this.primary ? TopologyType.ReplicaSetWithPrimary : TopologyType.Unknown;
// We should not mark the topology as Unknown because of one standalone
// we should just remove this server from the set
this.remove(server, { force: true });
return false;
}
Expand Down
5 changes: 4 additions & 1 deletion lib/core/topologies/shared.js
@@ -1,11 +1,14 @@
'use strict';

const MONGODB_ERROR_CODES = require('../../error_codes').MONGODB_ERROR_CODES;
const ReadPreference = require('./read_preference');
const TopologyType = require('../sdam/common').TopologyType;
const MongoError = require('../error').MongoError;
const isRetryableWriteError = require('../error').isRetryableWriteError;
const maxWireVersion = require('../utils').maxWireVersion;
const MongoNetworkError = require('../error').MongoNetworkError;
const MMAPv1_RETRY_WRITES_ERROR_CODE = 20;

const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;

/**
* Emit event if it exists
Expand Down
9 changes: 9 additions & 0 deletions lib/core/uri_parser.js
Expand Up @@ -106,6 +106,11 @@ function parseSrvConnectionString(uri, options, callback) {
}

record = qs.parse(record[0].join(''));

if (Object.keys(record).some(k => k.toLowerCase() === 'loadbalanced')) {
return callback(new MongoParseError('Load balancer mode requires driver version 4+'));
}

if (Object.keys(record).some(key => key !== 'authSource' && key !== 'replicaSet')) {
return callback(
new MongoParseError('Text record must only set `authSource` or `replicaSet`')
Expand Down Expand Up @@ -598,6 +603,10 @@ function parseConnectionString(uri, options, callback) {

parsedOptions = Object.assign({}, parsedOptions, options);

if (Object.keys(parsedOptions).some(k => k.toLowerCase() === 'loadbalanced')) {
return callback(new MongoParseError('Load balancer mode requires driver version 4+'));
}

if (protocol === PROTOCOL_MONGODB_SRV) {
return parseSrvConnectionString(uri, parsedOptions, callback);
}
Expand Down

0 comments on commit 910c564

Please sign in to comment.