Skip to content

Commit

Permalink
feat(PoolCluster): restoreNodeTimeout implementation (#3218)
Browse files Browse the repository at this point in the history
  • Loading branch information
@Lupennat
Lupennat authored Dec 4, 2024
1 parent 2e62d46 commit 9a38601
Show file tree
Hide file tree
Showing 10 changed files with 520 additions and 71 deletions.
180 changes: 131 additions & 49 deletions lib/pool_cluster.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,17 +13,30 @@ const EventEmitter = require('events').EventEmitter;
const makeSelector = {
RR() {
let index = 0;
return clusterIds => clusterIds[index++ % clusterIds.length];
return (clusterIds) => clusterIds[index++ % clusterIds.length];
},
RANDOM() {
return clusterIds =>
return (clusterIds) =>
clusterIds[Math.floor(Math.random() * clusterIds.length)];
},
ORDER() {
return clusterIds => clusterIds[0];
return (clusterIds) => clusterIds[0];
}
};

const getMonotonicMilliseconds = function () {
let ms;

if (typeof process.hrtime === 'function') {
ms = process.hrtime();
ms = ms[0] * 1e3 + ms[1] * 1e-6;
} else {
ms = process.uptime() * 1000;
}

return Math.floor(ms);
};

class PoolNamespace {
constructor(cluster, pattern, selector) {
this._cluster = cluster;
Expand All @@ -34,15 +47,28 @@ class PoolNamespace {
getConnection(cb) {
const clusterNode = this._getClusterNode();
if (clusterNode === null) {
return cb(new Error('Pool does Not exists.'));
let err = new Error('Pool does Not exist.');
err.code = 'POOL_NOEXIST';

if (this._cluster._findNodeIds(this._pattern, true).length !== 0) {
err = new Error('Pool does Not have online node.');
err.code = 'POOL_NONEONLINE';
}

return cb(err);
}
return this._cluster._getConnection(clusterNode, (err, connection) => {
if (err) {
if (
this._cluster._canRetry &&
this._cluster._findNodeIds(this._pattern).length !== 0
) {
this._cluster.emit('warn', err);
return this.getConnection(cb);
}

return cb(err);
}
if (connection === 'retry') {
return this.getConnection(cb);
}
return cb(null, connection);
});
}
Expand Down Expand Up @@ -79,9 +105,9 @@ class PoolNamespace {

/**
* pool cluster execute
* @param {*} sql
* @param {*} values
* @param {*} cb
* @param {*} sql
* @param {*} values
* @param {*} cb
*/
execute(sql, values, cb) {
if (typeof values === 'function') {
Expand Down Expand Up @@ -123,6 +149,7 @@ class PoolCluster extends EventEmitter {
this._canRetry =
typeof config.canRetry === 'undefined' ? true : config.canRetry;
this._removeNodeErrorCount = config.removeNodeErrorCount || 5;
this._restoreNodeTimeout = config.restoreNodeTimeout || 0;
this._defaultSelector = config.defaultSelector || 'RR';
this._closed = false;
this._lastId = 0;
Expand Down Expand Up @@ -155,13 +182,26 @@ class PoolCluster extends EventEmitter {
this._nodes[id] = {
id: id,
errorCount: 0,
pool: new Pool({ config: new PoolConfig(config) })
pool: new Pool({ config: new PoolConfig(config) }),
_offlineUntil: 0
};
this._serviceableNodeIds.push(id);
this._clearFindCaches();
}
}

remove(pattern) {
const foundNodeIds = this._findNodeIds(pattern, true);

for (let i = 0; i < foundNodeIds.length; i++) {
const node = this._getNode(foundNodeIds[i]);

if (node) {
this._removeNode(node);
}
}
}

getConnection(pattern, selector, cb) {
let namespace;
if (typeof pattern === 'function') {
Expand All @@ -181,7 +221,7 @@ class PoolCluster extends EventEmitter {
const cb =
callback !== undefined
? callback
: err => {
: (err) => {
if (err) {
throw err;
}
Expand All @@ -190,11 +230,12 @@ class PoolCluster extends EventEmitter {
process.nextTick(cb);
return;
}

this._closed = true;

let calledBack = false;
let waitingClose = 0;
const onEnd = err => {
const onEnd = (err) => {
if (!calledBack && (err || --waitingClose <= 0)) {
calledBack = true;
return cb(err);
Expand All @@ -205,67 +246,98 @@ class PoolCluster extends EventEmitter {
waitingClose++;
this._nodes[id].pool.end(onEnd);
}

if (waitingClose === 0) {
process.nextTick(onEnd);
}
}

_findNodeIds(pattern) {
if (typeof this._findCaches[pattern] !== 'undefined') {
return this._findCaches[pattern];
}
let foundNodeIds;
if (pattern === '*') {
// all
foundNodeIds = this._serviceableNodeIds;
} else if (this._serviceableNodeIds.indexOf(pattern) !== -1) {
// one
foundNodeIds = [pattern];
} else {
// wild matching
const keyword = pattern.substring(pattern.length - 1, 0);
foundNodeIds = this._serviceableNodeIds.filter(id =>
id.startsWith(keyword)
);
_findNodeIds(pattern, includeOffline) {
let currentTime = 0;
let foundNodeIds = this._findCaches[pattern];

if (typeof this._findCaches[pattern] === 'undefined') {
if (pattern === '*') {
// all
foundNodeIds = this._serviceableNodeIds;
} else if (this._serviceableNodeIds.indexOf(pattern) !== -1) {
// one
foundNodeIds = [pattern];
} else {
// wild matching
const keyword = pattern.substring(pattern.length - 1, 0);
foundNodeIds = this._serviceableNodeIds.filter((id) =>
id.startsWith(keyword)
);
}
}

this._findCaches[pattern] = foundNodeIds;
return foundNodeIds;

if (includeOffline) {
return foundNodeIds;
}

return foundNodeIds.filter((nodeId) => {
const node = this._getNode(nodeId);

if (!node._offlineUntil) {
return true;
}

if (!currentTime) {
currentTime = getMonotonicMilliseconds();
}

return node._offlineUntil <= currentTime;
});
}

_getNode(id) {
return this._nodes[id] || null;
}

_increaseErrorCount(node) {
if (++node.errorCount >= this._removeNodeErrorCount) {
const index = this._serviceableNodeIds.indexOf(node.id);
if (index !== -1) {
this._serviceableNodeIds.splice(index, 1);
delete this._nodes[node.id];
this._clearFindCaches();
node.pool.end();
this.emit('remove', node.id);
}
const errorCount = ++node.errorCount;

if (this._removeNodeErrorCount > errorCount) {
return;
}

if (this._restoreNodeTimeout > 0) {
node._offlineUntil =
getMonotonicMilliseconds() + this._restoreNodeTimeout;
this.emit('offline', node.id);
return;
}

this._removeNode(node);
this.emit('remove', node.id);
}

_decreaseErrorCount(node) {
if (node.errorCount > 0) {
--node.errorCount;
let errorCount = node.errorCount;

if (errorCount > this._removeNodeErrorCount) {
errorCount = this._removeNodeErrorCount;
}

if (errorCount < 1) {
errorCount = 1;
}

node.errorCount = errorCount - 1;

if (node._offlineUntil) {
node._offlineUntil = 0;
this.emit('online', node.id);
}
}

_getConnection(node, cb) {
node.pool.getConnection((err, connection) => {
if (err) {
this._increaseErrorCount(node);
if (this._canRetry) {
// REVIEW: this seems wrong?
this.emit('warn', err);
// eslint-disable-next-line no-console
console.warn(`[Error] PoolCluster : ${err}`);
return cb(null, 'retry');
}
return cb(err);
}
this._decreaseErrorCount(node);
Expand All @@ -275,6 +347,16 @@ class PoolCluster extends EventEmitter {
});
}

_removeNode(node) {
const index = this._serviceableNodeIds.indexOf(node.id);
if (index !== -1) {
this._serviceableNodeIds.splice(index, 1);
delete this._nodes[node.id];
this._clearFindCaches();
node.pool.end();
}
}

_clearFindCaches() {
this._findCaches = {};
}
Expand Down
4 changes: 2 additions & 2 deletions promise.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ class PromisePoolCluster extends EventEmitter {
super();
this.poolCluster = poolCluster;
this.Promise = thePromise || Promise;
inheritEvents(poolCluster, this, ['warn', 'remove']);
inheritEvents(poolCluster, this, ['warn', 'remove' , 'online', 'offline']);
}

getConnection(pattern, selector) {
Expand Down Expand Up @@ -156,7 +156,7 @@ class PromisePoolCluster extends EventEmitter {
})(func);
}
}
})(['add']);
})(['add', 'remove']);

function createPromisePoolCluster(opts) {
const corePoolCluster = createPoolCluster(opts);
Expand Down
36 changes: 36 additions & 0 deletions test/esm/integration/pool-cluster/test-promise-wrapper.test.mjs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,4 +43,40 @@ const { createPoolCluster } = require('../../../../promise.js');

poolCluster.poolCluster.emit('remove');
});

await test(async () => {
const poolCluster = createPoolCluster();

poolCluster.once('offline', async function () {
await new Promise((resolve) => {
assert.equal(
// eslint-disable-next-line no-invalid-this
this,
poolCluster,
'should propagate offline event to promise wrapper',
);
resolve(true);
});
});

poolCluster.poolCluster.emit('offline');
});

await test(async () => {
const poolCluster = createPoolCluster();

poolCluster.once('online', async function () {
await new Promise((resolve) => {
assert.equal(
// eslint-disable-next-line no-invalid-this
this,
poolCluster,
'should propagate online event to promise wrapper',
);
resolve(true);
});
});

poolCluster.poolCluster.emit('online');
});
})();
6 changes: 6 additions & 0 deletions test/tsc-build/mysql/createPoolCluster/remove.test.ts
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
import { mysql } from '../../index.test.js';

const poolCluster = mysql.createPoolCluster();

// Overload: poolCluster.add(group, connectionUri);
poolCluster.remove('cluster1');
Loading

0 comments on commit 9a38601

Please sign in to comment.