'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 = MONGODB_ERROR_CODES.IllegalOperation;
|
|
/**
|
* Emit event if it exists
|
* @method
|
*/
|
function emitSDAMEvent(self, event, description) {
|
if (self.listeners(event).length > 0) {
|
self.emit(event, description);
|
}
|
}
|
|
function createCompressionInfo(options) {
|
if (!options.compression || !options.compression.compressors) {
|
return [];
|
}
|
|
// Check that all supplied compressors are valid
|
options.compression.compressors.forEach(function(compressor) {
|
if (compressor !== 'snappy' && compressor !== 'zlib') {
|
throw new Error('compressors must be at least one of snappy or zlib');
|
}
|
});
|
|
return options.compression.compressors;
|
}
|
|
function clone(object) {
|
return JSON.parse(JSON.stringify(object));
|
}
|
|
var getPreviousDescription = function(self) {
|
if (!self.s.serverDescription) {
|
self.s.serverDescription = {
|
address: self.name,
|
arbiters: [],
|
hosts: [],
|
passives: [],
|
type: 'Unknown'
|
};
|
}
|
|
return self.s.serverDescription;
|
};
|
|
var emitServerDescriptionChanged = function(self, description) {
|
if (self.listeners('serverDescriptionChanged').length > 0) {
|
// Emit the server description changed events
|
self.emit('serverDescriptionChanged', {
|
topologyId: self.s.topologyId !== -1 ? self.s.topologyId : self.id,
|
address: self.name,
|
previousDescription: getPreviousDescription(self),
|
newDescription: description
|
});
|
|
self.s.serverDescription = description;
|
}
|
};
|
|
var getPreviousTopologyDescription = function(self) {
|
if (!self.s.topologyDescription) {
|
self.s.topologyDescription = {
|
topologyType: 'Unknown',
|
servers: [
|
{
|
address: self.name,
|
arbiters: [],
|
hosts: [],
|
passives: [],
|
type: 'Unknown'
|
}
|
]
|
};
|
}
|
|
return self.s.topologyDescription;
|
};
|
|
var emitTopologyDescriptionChanged = function(self, description) {
|
if (self.listeners('topologyDescriptionChanged').length > 0) {
|
// Emit the server description changed events
|
self.emit('topologyDescriptionChanged', {
|
topologyId: self.s.topologyId !== -1 ? self.s.topologyId : self.id,
|
address: self.name,
|
previousDescription: getPreviousTopologyDescription(self),
|
newDescription: description
|
});
|
|
self.s.serverDescription = description;
|
}
|
};
|
|
var changedIsMaster = function(self, currentIsmaster, ismaster) {
|
var currentType = getTopologyType(self, currentIsmaster);
|
var newType = getTopologyType(self, ismaster);
|
if (newType !== currentType) return true;
|
return false;
|
};
|
|
var getTopologyType = function(self, ismaster) {
|
if (!ismaster) {
|
ismaster = self.ismaster;
|
}
|
|
if (!ismaster) return 'Unknown';
|
if (ismaster.ismaster && ismaster.msg === 'isdbgrid') return 'Mongos';
|
if (ismaster.ismaster && !ismaster.hosts) return 'Standalone';
|
if (ismaster.ismaster) return 'RSPrimary';
|
if (ismaster.secondary) return 'RSSecondary';
|
if (ismaster.arbiterOnly) return 'RSArbiter';
|
return 'Unknown';
|
};
|
|
var inquireServerState = function(self) {
|
return function(callback) {
|
if (self.s.state === 'destroyed') return;
|
// Record response time
|
var start = new Date().getTime();
|
|
// emitSDAMEvent
|
emitSDAMEvent(self, 'serverHeartbeatStarted', { connectionId: self.name });
|
|
// Attempt to execute ismaster command
|
self.command('admin.$cmd', { ismaster: true }, { monitoring: true }, function(err, r) {
|
if (!err) {
|
// Legacy event sender
|
self.emit('ismaster', r, self);
|
|
// Calculate latencyMS
|
var latencyMS = new Date().getTime() - start;
|
|
// Server heart beat event
|
emitSDAMEvent(self, 'serverHeartbeatSucceeded', {
|
durationMS: latencyMS,
|
reply: r.result,
|
connectionId: self.name
|
});
|
|
// Did the server change
|
if (changedIsMaster(self, self.s.ismaster, r.result)) {
|
// Emit server description changed if something listening
|
emitServerDescriptionChanged(self, {
|
address: self.name,
|
arbiters: [],
|
hosts: [],
|
passives: [],
|
type: !self.s.inTopology ? 'Standalone' : getTopologyType(self)
|
});
|
}
|
|
// Updat ismaster view
|
self.s.ismaster = r.result;
|
|
// Set server response time
|
self.s.isMasterLatencyMS = latencyMS;
|
} else {
|
emitSDAMEvent(self, 'serverHeartbeatFailed', {
|
durationMS: latencyMS,
|
failure: err,
|
connectionId: self.name
|
});
|
}
|
|
// Peforming an ismaster monitoring callback operation
|
if (typeof callback === 'function') {
|
return callback(err, r);
|
}
|
|
// Perform another sweep
|
self.s.inquireServerStateTimeout = setTimeout(inquireServerState(self), self.s.haInterval);
|
});
|
};
|
};
|
|
//
|
// Clone the options
|
var cloneOptions = function(options) {
|
var opts = {};
|
for (var name in options) {
|
opts[name] = options[name];
|
}
|
return opts;
|
};
|
|
function Interval(fn, time) {
|
var timer = false;
|
|
this.start = function() {
|
if (!this.isRunning()) {
|
timer = setInterval(fn, time);
|
}
|
|
return this;
|
};
|
|
this.stop = function() {
|
clearInterval(timer);
|
timer = false;
|
return this;
|
};
|
|
this.isRunning = function() {
|
return timer !== false;
|
};
|
}
|
|
function Timeout(fn, time) {
|
var timer = false;
|
var func = () => {
|
if (timer) {
|
clearTimeout(timer);
|
timer = false;
|
|
fn();
|
}
|
};
|
|
this.start = function() {
|
if (!this.isRunning()) {
|
timer = setTimeout(func, time);
|
}
|
return this;
|
};
|
|
this.stop = function() {
|
clearTimeout(timer);
|
timer = false;
|
return this;
|
};
|
|
this.isRunning = function() {
|
return timer !== false;
|
};
|
}
|
|
function diff(previous, current) {
|
// Difference document
|
var diff = {
|
servers: []
|
};
|
|
// Previous entry
|
if (!previous) {
|
previous = { servers: [] };
|
}
|
|
// Check if we have any previous servers missing in the current ones
|
for (var i = 0; i < previous.servers.length; i++) {
|
var found = false;
|
|
for (var j = 0; j < current.servers.length; j++) {
|
if (current.servers[j].address.toLowerCase() === previous.servers[i].address.toLowerCase()) {
|
found = true;
|
break;
|
}
|
}
|
|
if (!found) {
|
// Add to the diff
|
diff.servers.push({
|
address: previous.servers[i].address,
|
from: previous.servers[i].type,
|
to: 'Unknown'
|
});
|
}
|
}
|
|
// Check if there are any severs that don't exist
|
for (j = 0; j < current.servers.length; j++) {
|
found = false;
|
|
// Go over all the previous servers
|
for (i = 0; i < previous.servers.length; i++) {
|
if (previous.servers[i].address.toLowerCase() === current.servers[j].address.toLowerCase()) {
|
found = true;
|
break;
|
}
|
}
|
|
// Add the server to the diff
|
if (!found) {
|
diff.servers.push({
|
address: current.servers[j].address,
|
from: 'Unknown',
|
to: current.servers[j].type
|
});
|
}
|
}
|
|
// Got through all the servers
|
for (i = 0; i < previous.servers.length; i++) {
|
var prevServer = previous.servers[i];
|
|
// Go through all current servers
|
for (j = 0; j < current.servers.length; j++) {
|
var currServer = current.servers[j];
|
|
// Matching server
|
if (prevServer.address.toLowerCase() === currServer.address.toLowerCase()) {
|
// We had a change in state
|
if (prevServer.type !== currServer.type) {
|
diff.servers.push({
|
address: prevServer.address,
|
from: prevServer.type,
|
to: currServer.type
|
});
|
}
|
}
|
}
|
}
|
|
// Return difference
|
return diff;
|
}
|
|
/**
|
* Shared function to determine clusterTime for a given topology
|
*
|
* @param {*} topology
|
* @param {*} clusterTime
|
*/
|
function resolveClusterTime(topology, $clusterTime) {
|
if (topology.clusterTime == null) {
|
topology.clusterTime = $clusterTime;
|
} else {
|
if ($clusterTime.clusterTime.greaterThan(topology.clusterTime.clusterTime)) {
|
topology.clusterTime = $clusterTime;
|
}
|
}
|
}
|
|
// NOTE: this is a temporary move until the topologies can be more formally refactored
|
// to share code.
|
const SessionMixins = {
|
endSessions: function(sessions, callback) {
|
if (!Array.isArray(sessions)) {
|
sessions = [sessions];
|
}
|
|
// TODO:
|
// When connected to a sharded cluster the endSessions command
|
// can be sent to any mongos. When connected to a replica set the
|
// endSessions command MUST be sent to the primary if the primary
|
// is available, otherwise it MUST be sent to any available secondary.
|
// Is it enough to use: ReadPreference.primaryPreferred ?
|
this.command(
|
'admin.$cmd',
|
{ endSessions: sessions },
|
{ readPreference: ReadPreference.primaryPreferred },
|
() => {
|
// intentionally ignored, per spec
|
if (typeof callback === 'function') callback();
|
}
|
);
|
}
|
};
|
|
function topologyType(topology) {
|
if (topology.description) {
|
return topology.description.type;
|
}
|
|
if (topology.type === 'mongos') {
|
return TopologyType.Sharded;
|
} else if (topology.type === 'replset') {
|
return TopologyType.ReplicaSetWithPrimary;
|
}
|
|
return TopologyType.Single;
|
}
|
|
const RETRYABLE_WIRE_VERSION = 6;
|
|
/**
|
* Determines whether the provided topology supports retryable writes
|
*
|
* @param {Mongos|Replset} topology
|
*/
|
const isRetryableWritesSupported = function(topology) {
|
const maxWireVersion = topology.lastIsMaster().maxWireVersion;
|
if (maxWireVersion < RETRYABLE_WIRE_VERSION) {
|
return false;
|
}
|
|
if (!topology.logicalSessionTimeoutMinutes) {
|
return false;
|
}
|
|
if (topologyType(topology) === TopologyType.Single) {
|
return false;
|
}
|
|
return true;
|
};
|
|
const MMAPv1_RETRY_WRITES_ERROR_MESSAGE =
|
'This MongoDB deployment does not support retryable writes. Please add retryWrites=false to your connection string.';
|
|
function getMMAPError(err) {
|
if (err.code !== MMAPv1_RETRY_WRITES_ERROR_CODE || !err.errmsg.includes('Transaction numbers')) {
|
return err;
|
}
|
|
// According to the retryable writes spec, we must replace the error message in this case.
|
// We need to replace err.message so the thrown message is correct and we need to replace err.errmsg to meet the spec requirement.
|
const newErr = new MongoError({
|
message: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
|
errmsg: MMAPv1_RETRY_WRITES_ERROR_MESSAGE,
|
originalError: err
|
});
|
return newErr;
|
}
|
|
// NOTE: only used for legacy topology types
|
function legacyIsRetryableWriteError(err, topology) {
|
if (!(err instanceof MongoError)) {
|
return false;
|
}
|
|
// if pre-4.4 server, then add error label if its a retryable write error
|
if (
|
isRetryableWritesSupported(topology) &&
|
(err instanceof MongoNetworkError ||
|
(maxWireVersion(topology) < 9 && isRetryableWriteError(err)))
|
) {
|
err.addErrorLabel('RetryableWriteError');
|
}
|
|
return err.hasErrorLabel('RetryableWriteError');
|
}
|
|
module.exports = {
|
SessionMixins,
|
resolveClusterTime,
|
inquireServerState,
|
getTopologyType,
|
emitServerDescriptionChanged,
|
emitTopologyDescriptionChanged,
|
cloneOptions,
|
createCompressionInfo,
|
clone,
|
diff,
|
Interval,
|
Timeout,
|
isRetryableWritesSupported,
|
getMMAPError,
|
topologyType,
|
legacyIsRetryableWriteError
|
};
|
