From 2f3b7bcf1ab846d0d61c155f3f2d5244ffd68ffa Mon Sep 17 00:00:00 2001 From: Joshua T Corbin Date: Mon, 29 Feb 2016 11:22:24 -0800 Subject: [PATCH 1/7] test/peer-churn: plant each turnip as soon as possible --- test/peer-churn.js | 33 +++++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/test/peer-churn.js b/test/peer-churn.js index 1066651..6d7e63f 100644 --- a/test/peer-churn.js +++ b/test/peer-churn.js @@ -25,6 +25,7 @@ var CountedReadySignal = require('ready-signal/counted'); var timers = require('timers'); var util = require('util'); +var Turnip = require('./lib/turnip').Turnip; var Turnips = require('./lib/turnip').Turnips; var allocCluster = require('./lib/test-cluster.js'); var CollapsedAssert = require('./lib/collapsed-assert.js'); @@ -130,15 +131,36 @@ allocCluster.test('peer churn', { function thenShutdownFirstRound() { assert.comment('- thenShutdownFirstRound'); + + turnips = new Turnips(); + turnips.ready = new CountedReadySignal(first.length); + collectParallel(first, function each(remote, i, done) { + var hostPort = remote.channel.hostPort; + var parts = hostPort.split(':'); + var host = parts[0]; + var port = parseInt(parts[1], 10); + // half of them do an unad then destroy, the other half just go away if (i % 2 === 0) { remote.doUnregister(function unreged(err) { assert.ifError(err, 'expected no unregister error'); - remote.destroy(done); + remote.destroy(plant); }); } else { - remote.destroy(done); + remote.destroy(plant); + } + + function plant() { + assert.comment('-- planting turnip[' + i + '] on ' + hostPort); + turnips.turnips[i] = new Turnip(port, host, ready); + done(); + } + + function ready(err) { + assert.comment('-- turnip[' + i + '] ready on ' + hostPort); + assert.ifError(err, 'no unexpected turnip error'); + turnips.ready.signal(); } }, thenWaitForReaper); } @@ -188,12 +210,7 @@ allocCluster.test('peer churn', { ].indexOf(record.msg) >= 0, 'expected peer churn logs'); }); - turnips = Turnips.forAll(first, function getRemotePortHost(remote) { - var parts = remote.channel.hostPort.split(':'); - var host = parts[0]; - var port = parseInt(parts[1], 10); - return [port, host]; - }, thenWaitAndSee); + turnips.ready(thenWaitAndSee); } function thenWaitAndSee() { From d92f18dccc432c0176e4d956cf80b4b759d00d79 Mon Sep 17 00:00:00 2001 From: Joshua T Corbin Date: Tue, 1 Mar 2016 14:00:12 -0800 Subject: [PATCH 2/7] serviceProxy.ensurePartialConnections: simplify log extension --- service-proxy.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service-proxy.js b/service-proxy.js index f10fd8b..c5bfffe 100644 --- a/service-proxy.js +++ b/service-proxy.js @@ -937,7 +937,7 @@ function ensurePartialConnections(serviceChannel, serviceName, hostPort, reason, if (!result.noop) { self.logger.info( 'implementing affinity change', - self.extendLogInfo(partialRange.extendLogInfo({ + result.extendLogInfo({ serviceName: serviceName, reason: reason, causingWorker: hostPort, From 7f0e03d06a68f064274dc2e1d4be2fe4c1bda2af Mon Sep 17 00:00:00 2001 From: Joshua T Corbin Date: Tue, 1 Mar 2016 14:01:23 -0800 Subject: [PATCH 3/7] serviceProxy: inline .deletePeerIndex --- service-proxy.js | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/service-proxy.js b/service-proxy.js index c5bfffe..2b936c0 100644 --- a/service-proxy.js +++ b/service-proxy.js @@ -696,17 +696,6 @@ function refreshServicePeer(serviceName, hostPort) { self.ensurePeerConnected(serviceName, peer, 'service peer refresh', now); }; -ServiceDispatchHandler.prototype.deletePeerIndex = -function deletePeerIndex(serviceName, hostPort) { - var self = this; - - if (self.partialAffinityEnabled) { - deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort); - deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName); - } - deleteIndexEntry(self.knownPeers, hostPort, serviceName); -}; - ServiceDispatchHandler.prototype.ensurePeerConnected = function ensurePeerConnected(serviceName, peer, reason, now) { var self = this; @@ -1406,7 +1395,13 @@ function reapSinglePeer(hostPort, serviceNames, now) { if (serviceChannel) { serviceChannel.peers.delete(hostPort); } - self.deletePeerIndex(serviceName, hostPort); + + if (self.partialAffinityEnabled) { + deleteIndexEntry(self.connectedServicePeers, serviceName, hostPort); + deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName); + } + deleteIndexEntry(self.knownPeers, hostPort, serviceName); + var partialRange = self.partialRanges[serviceName]; if (partialRange) { partialRange.removeWorker(hostPort, now); From 9de9a5bec716ffa51c062b931ebfefd98ba83762 Mon Sep 17 00:00:00 2001 From: Joshua T Corbin Date: Mon, 29 Feb 2016 12:34:54 -0800 Subject: [PATCH 4/7] WIP: add sustained traffic to peer churn test --- test/peer-churn.js | 164 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 160 insertions(+), 4 deletions(-) diff --git a/test/peer-churn.js b/test/peer-churn.js index 6d7e63f..abf823b 100644 --- a/test/peer-churn.js +++ b/test/peer-churn.js @@ -39,6 +39,7 @@ var CLUSTER_SIZE = 10; var CHURN_FACTOR = 0.5; var K_VALUE = 5; var SERVICE_SIZE = 10; +var SUSTAIN_PARALLEL = 10 * SERVICE_SIZE; var ENDPOINT_DELAY = 0.5 * PERIOD; var ENDPOINT_DELAY_FUZZ = 0.50; @@ -66,6 +67,21 @@ allocCluster.test('peer churn', { // XXX do we care about the rare: // AUTOBAHN INFO: ignoring outresponse.send on a closed connection ~ { responseId: 5, + // TODO: provoked by sustained traffic + ['info', 'OutResponse.send() after inreq timed out'], + ['info', 'error for timed out outgoing response'], + ['info', 'expected error while forwarding'], + ['info', 'forwarding expected error frame'], + ['info', 'got call response for timed out call request'], + ['info', 'got expected errorframe without call request'], + ['info', 'ignoring outresponse.send on a closed connection'], + ['info', 'popOutReq received for unknown or lost id'], + ['warn', 'Relay advertise failed with expected err'], + ['warn', 'forwarding error frame'], + ['warn', 'got errorframe without call request'], + ['warn', 'got unexpected call response without call request'], + ['warn', 'mismatched expired operation tombstone'], + ['warn', 'affinity change failed audit'], ['warn', 'removing stale partial affinity worker'], @@ -102,6 +118,8 @@ allocCluster.test('peer churn', { var first = null; var second = null; var turnips = null; + var sustained = null; + var cohort = null; function clientReady() { assert.comment('- clientReady'); @@ -126,7 +144,40 @@ allocCluster.test('peer churn', { assert.comment('- gotSecondRound'); checkAllLogs(cluster, assert, checkConnectingLog); checkExitsTo(cluster, 'lucy', second, 'the second peers', assert); - checkRequestsTo('lucy', first.concat(second), 'either generation', client.clientChannel, assert, thenShutdownFirstRound); + checkRequestsTo('lucy', first.concat(second), 'either generation', client.clientChannel, assert, thenStartSustainedTraffic); + } + + function thenStartSustainedTraffic() { + assert.comment('- thenStartSustainedTraffic'); + sustained = checkSustainedTo('lucy', client.clientChannel, normalSustainedCheck); + cohort = first.concat(second); + + timers.setTimeout(thenCheckFirstRound, fuzzedPeriods(SETTLE_PERIODS)); + } + + function normalSustainedCheck(err, delay, serverHostPort, cassert) { + cassert.ifError(err, 'no unexpected sustained request error'); + cassert.ok(cohort.some(function isit(remote) { + return remote.hostPort === serverHostPort; + }), 'expected sustained request to be served by cohort'); + } + + function sustainedCheckDuringShutdown(err, delay, serverHostPort, cassert) { + if (err && err.type === 'tchannel.request.timeout' || + err && err.type === 'tchannel.timeout') { + cassert.ok('expected timeout error'); // TODO: why do we get these? could we not? + return; + } + normalSustainedCheck(err, delay, serverHostPort, cassert); + } + + function thenCheckFirstRound() { + assert.comment('- thenCheckFirstRound'); + + sustained.checkpoint().report(assert, 'expected sustained requests to be served by either generation'); + sustained.check = sustainedCheckDuringShutdown; + + thenShutdownFirstRound(); } function thenShutdownFirstRound() { @@ -180,6 +231,12 @@ allocCluster.test('peer churn', { 'expected registration failure during destruction', 'resetting connection', + // happens for destructive with sustained + 'error while forwarding', + 'forwarding error frame', + 'expected error while forwarding', + 'ignoring outresponse.send on a closed connection', // TODO: can we fix this one? + // these happen due to the unads 'Peer drained and closed due to unadvertisement', 'canceling peer drain', @@ -196,7 +253,9 @@ allocCluster.test('peer churn', { function thenTurnip() { assert.comment('- thenTurnip'); - // checkNoLogs('reap', cluster, assert); + sustained.checkpoint().report(assert, 'expected sustained requests to be served by either generation'); + cohort = second; + sustained.check = normalSustainedCheck; checkAllLogs(cluster, assert, function checkEachLog(record) { assert.ok([ @@ -206,7 +265,23 @@ allocCluster.test('peer churn', { 'reaping dead peers', 'reaping dead peer', 'draining peer', - 'pruning peers' + 'pruning peers', + + // these happen for the unad peers + 'Peer drained and closed due to unadvertisement', + + // happens for destructive with sustained + 'expected error while forwarding', + 'forwarding expected error frame', + 'error while forwarding', + 'forwarding error frame', + 'got call response for timed out call request', + + // TODO: can we fix these? + 'ignoring outresponse.send on a closed connection', + 'error for timed out outgoing response', + 'OutResponse.send() after inreq timed out' + ].indexOf(record.msg) >= 0, 'expected peer churn logs'); }); @@ -225,7 +300,12 @@ allocCluster.test('peer churn', { assert.comment('- thenTendAndSend'); checkNoLogs('turnip tending', cluster, assert); checkTurnips(); - sendAfterChurn(); + thenStopSustaining(); + } + + function thenStopSustaining() { + assert.comment('- thenStopSustaining'); + sustained.cancel(sendAfterChurn); } function checkTurnips() { @@ -242,6 +322,9 @@ allocCluster.test('peer churn', { function sendAfterChurn() { assert.comment('- sendAfterChurn'); + + sustained.checkpoint().report(assert, 'expected sustained requests to be served by second'); + checkRequestsTo('lucy', second, 'the second peers', client.clientChannel, assert, thenDestroySecondRound); } @@ -343,6 +426,43 @@ function checkRequestsTo(serviceName, cohort, desc, chan, assert, cb) { }); } +function checkSustainedTo(serviceName, chan, check) { + var cassert = CollapsedAssert(); + var cancel = sustainMany(chan, SUSTAIN_PARALLEL, { + serviceName: serviceName, + timeout: REQUEST_TIMEOUT + }, 'who', '', '', sent); + + var finished = null; + var sustain = { + checkpoint: checkpoint, + check: check, + cancel: finish + }; + return sustain; + + function sent(err, res, arg2, arg3) { + sustain.check(err, parseFloat(arg2), String(arg3), cassert); + } + + function checkpoint() { + var prior = cassert; + cassert = CollapsedAssert(); + return prior; + } + + function finish(finishedCb) { + if (!finished) { + finished = finishedCb; + cancel(thenReport); + } + } + + function thenReport() { + finished(); + } +} + function sendMany(chan, N, opts, arg1, arg2, arg3, check, cb) { var sendsDone = CountedReadySignal(N); sendsDone(cb); @@ -360,6 +480,42 @@ function sendMany(chan, N, opts, arg1, arg2, arg3, check, cb) { } } +function sustainMany(chan, N, opts, arg1, arg2, arg3, check) { + var ongoing = 0; + var drained = null; + // - callback to be called once ongoing <= 0 + // - stops making new requests once drained !== null + fill(); + return cancel; + + function cancel(drainedCb) { + drained = drainedCb; + } + + function fill() { + if (drained) { + if (ongoing === 0) { + drained(); + } + return; + } + while (ongoing < N) { + doRequest(); + } + } + + function doRequest() { + ++ongoing; + chan.request(opts).send(arg1, arg2, arg3, done); + } + + function done(err, res, resArg2, resArg3) { + --ongoing; + check(err, res, resArg2, resArg3); + fill(); + } +} + function takeLogs(dbgLogtron) { var records = dbgLogtron._backend.records; dbgLogtron._backend.records = []; From b8c823307da093f625990d273581159d9138bb2c Mon Sep 17 00:00:00 2001 From: Joshua T Corbin Date: Tue, 1 Mar 2016 13:54:51 -0800 Subject: [PATCH 5/7] DEBUG: log dump --- test/peer-churn.js | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/test/peer-churn.js b/test/peer-churn.js index abf823b..3c96e7e 100644 --- a/test/peer-churn.js +++ b/test/peer-churn.js @@ -537,6 +537,10 @@ function checkAllLogs(cluster, assert, check) { record.levelName, record.msg, record._logData.fields )); } + assert.comment(util.format( + 'LOG: %s %s: %j', + record.levelName, record.msg, record._logData.fields + )); } assert.removeListener('result', onResult); @@ -556,6 +560,9 @@ function checkNoLogs(desc, cluster, assert) { assert.comment(util.format('UNEXPECTED LOG: %s %s: %j', record.levelName, record.msg, record._logData.fields )); + assert.comment(util.format('LOG: %s %s: %j', + record.levelName, record.msg, record._logData.fields + )); } } } From f06794ad74e879226065fd5c6f84886a9d5c2b12 Mon Sep 17 00:00:00 2001 From: Joshua T Corbin Date: Tue, 1 Mar 2016 13:59:32 -0800 Subject: [PATCH 6/7] DEBUG: sum logs --- service-proxy.js | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/service-proxy.js b/service-proxy.js index 2b936c0..f38a19b 100644 --- a/service-proxy.js +++ b/service-proxy.js @@ -822,6 +822,12 @@ function addNewPartialPeer(serviceChannel, hostPort, now) { partialRange.addWorker(hostPort, now); } + self.logger.info('ADD NEW PARTIAL PEER KNOWS', { + serviceName: serviceName, + hostPort: hostPort, + now: now + }); + // Unmark recently seen peers, so they don't get reaped deleteIndexEntry(self.peersToReap, hostPort, serviceName); // Mark known peers, so they are candidates for future reaping @@ -864,6 +870,12 @@ function freshenPartialPeer(peer, serviceName, now) { deleteIndexEntry(self.connectedPeerServices, hostPort, serviceName); } + self.logger.info('FRESHEN PARTIAL PEER KNOWS', { + serviceName: serviceName, + hostPort: hostPort, + now: now + }); + // Unmark recently seen peers, so they don't get reaped deleteIndexEntry(self.peersToReap, peer.hostPort, serviceName); // Mark known peers, so they are candidates for future reaping @@ -932,7 +944,7 @@ function ensurePartialConnections(serviceChannel, serviceName, hostPort, reason, causingWorker: hostPort, numToConnect: result.toConnect.length, numToDisconnect: result.toDisconnect.length - })) + }) ); result.implement(); } @@ -1362,6 +1374,11 @@ function reapSinglePeer(hostPort, serviceNames, now) { var self = this; if (self.knownPeers[hostPort]) { + self.logger.info('REAP ABORT, WHO KNOWS', { + serviceNames: serviceNames, + hostPort: hostPort, + now: now + }); return; } @@ -1752,7 +1769,15 @@ function audit() { worker = this.toDisconnect[i]; peer = this.serviceChannel.peers.get(worker); if (!peer) { + // this.proxy.logger.warn( + // 'DEBUG workers before', + // {workers: this.partialRange.workers.join(','), worker: worker} + // ); this.removeWorker(worker, 'toDisconnect'); + // this.proxy.logger.warn( + // 'DEBUG workers after', + // {workers: this.partialRange.workers.join(',')} + // ); ++this.staleToDisconnect; } } From 2440401b947b32b7b65de1bb944f44e17de00410 Mon Sep 17 00:00:00 2001 From: Joshua T Corbin Date: Mon, 7 Mar 2016 16:01:40 -0800 Subject: [PATCH 7/7] WIP --- service-proxy.js | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/service-proxy.js b/service-proxy.js index f38a19b..564363b 100644 --- a/service-proxy.js +++ b/service-proxy.js @@ -732,24 +732,53 @@ function connectSinglePeer(hostPort, connectInfo) { var self = this; if (self.peersToPrune[hostPort]) { + + self.logger.info('SKIPPING PEER CONNECTION: PRUNED', self.extendLogInfo({ + hostPort: hostPort, + connectInfo: connectInfo + })); + return; } var serviceName = connectInfo.serviceName; var serviceChannel = self.getServiceChannel(serviceName); if (!serviceChannel) { + + self.logger.info('SKIPPING PEER CONNECTION: NO SERVICE CHANNEL', self.extendLogInfo({ + hostPort: hostPort, + connectInfo: connectInfo + })); + return; } var peer = serviceChannel.peers.get(hostPort); if (!peer) { + + self.logger.info('SKIPPING PEER CONNECTION: NO PEER', self.extendLogInfo({ + hostPort: hostPort, + connectInfo: connectInfo + })); + return; } if (peer.draining) { + + self.logger.info('SKIPPING PEER CONNECTION: PEER DRAINING', self.extendLogInfo({ + hostPort: hostPort, + connectInfo: connectInfo + })); + return; } + self.logger.info('CONNECTING TO PEER', self.extendLogInfo({ + hostPort: hostPort, + connectInfo: connectInfo + })); + peer.connectTo(); };