Skip to content
This repository was archived by the owner on Feb 18, 2021. It is now read-only.

WIP: Ensure connected out and drain in #90

Open
jcorbin wants to merge 3 commits into
base: master
Choose a base branch
from
Open

WIP: Ensure connected out and drain in #90

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5ff98be
ServiceDispatchHandler: drain incoming connections under .ensurePeerC…
Oct 22, 2015
00184d5
ServiceDispatchHandler: factor #drainInOnceConnected out of #ensurePe…
Nov 18, 2015
3e7d23d
Add feature flag for draining incoming connections
Nov 18, 2015
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
7 changes: 7 additions & 0 deletions clients/index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -365,6 +365,7 @@ ApplicationClients.prototype.onRemoteConfigUpdate = function onRemoteConfigUpdat
self.updateReservoir();
self.updateReapPeersPeriod();
self.updatePrunePeersPeriod();
self.updateDrainIncomingConnections();
self.updatePartialAffinityEnabled();
self.setMaximumRelayTTL();
self.updatePeerHeapEnabled();
Expand Down Expand Up @@ -444,6 +445,12 @@ function updatePrunePeersPeriod() {
self.serviceProxy.setPrunePeersPeriod(period);
};

ApplicationClients.prototype.updateDrainIncomingConnections = function updateDrainIncomingConnections() {
var self = this;
var enabled = self.remoteConfig.get('drainIncomingConnections.enabled', false);
self.serviceProxy.setDrainIncomingConnections(enabled);
};

ApplicationClients.prototype.updatePartialAffinityEnabled = function updatePartialAffinityEnabled() {
var self = this;
var enabled = self.remoteConfig.get('partialAffinity.enabled', false);
Expand Down
77 changes: 76 additions & 1 deletion service-proxy.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,7 @@ function ServiceDispatchHandler(options) {
});
self.rateLimiterEnabled = options.rateLimiterEnabled;

self.drainIncomingConnections = !!options.drainIncomingConnections;
self.partialAffinityEnabled = !!options.partialAffinityEnabled;
self.minPeersPerWorker = options.minPeersPerWorker || DEFAULT_MIN_PEERS_PER_WORKER;
self.minPeersPerRelay = options.minPeersPerRelay || DEFAULT_MIN_PEERS_PER_RELAY;
Expand Down Expand Up @@ -635,7 +636,75 @@ function ensurePeerConnected(serviceName, peer, reason, now) {
peer.clearDrain('canceled to ensure peer connection');
}

peer.connectTo();
var conn = peer.connectTo();
if (self.drainIncomingConnections) {
self.drainInOnceConnected(peer, conn, reason);
}
};

ServiceDispatchHandler.prototype.drainInOnceConnected =
function drainInOnceConnected(peer, conn, reason) {
var self = this;

peer.waitForIdentified(conn, onConnIded);

function onConnIded(err) {
if (err) {
self.logger.warn(
'failed to ensure outgoing connection to service peer',
self.extendLogInfo({
error: err,
peerHostPort: peer.hostPort,
refreshReason: reason
}));
return;
}
peer.drain({
reason: reason,
Copy link
Copy Markdown
Contributor

@Raynos Raynos Oct 26, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We either need to communicate to every one to expect declined messages about draining.

Or make a change in the hyperbahn client in all languages to silence the draining error frame log.

Copy link
Copy Markdown
Contributor Author

@jcorbin jcorbin Oct 26, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yup, that's a good educational moment to have anyhow, since we'll drain and decline during graceful shutdown anyhow, which we want them to handle.

Plus partial affinity in #66 will Soon ™ drain ex-affinity peers, so clients just really need to get used to this.

direction: 'in',
timeout: self.drainTimeout
}, connectDrainDone);
}

function connectDrainDone(err) {
if (err &&
err.type === 'tchannel.drain.peer.timed-out') {
// TODO: stat?
self.logger.warn(
'forcibly closing drained peer',
self.extendLogInfo({
error: err,
drainReason: reason
})
);
err = null;
}
if (err) {
self.logger.warn(
'failed to drain incoming connections from service peer',
self.extendLogInfo({
error: err,
peerHostPort: peer.hostPort
})
);
peer.clearDrain();
return;
}
peer.closeDrainedConnections(connectDrainCloseDone);
}

function connectDrainCloseDone(err) {
if (err) {
self.logger.warn(
'failed to close drained incoming connections from service peer',
self.extendLogInfo({
error: err,
peerHostPort: peer.hostPort
})
);
}
peer.clearDrain();
}
};

ServiceDispatchHandler.prototype.getPartialRange =
Expand Down Expand Up @@ -1538,6 +1607,12 @@ function disableRateLimiter() {
self.rateLimiterEnabled = false;
};

ServiceDispatchHandler.prototype.setDrainIncomingConnections =
function setDrainIncomingConnections(enabled) {
var self = this;
self.drainIncomingConnections = !!enabled;
};

ServiceDispatchHandler.prototype.setPartialAffinityEnabled =
function enablePartialAffinity(enabled) {
var self = this;
Expand Down