Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Minor] Add more logging around peer disconnects #131

Merged
merged 4 commits into from
Oct 23, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -211,14 +211,20 @@ public void processMessage(final Capability cap, final Message message) {
return;
} else if (!peer.statusHasBeenReceived()) {
// Peers are required to send status messages before any other message type
LOG.debug(
"{} requires a Status ({}) message to be sent first. Instead, received message {}. Disconnecting from {}.",
this.getClass().getSimpleName(),
EthPV62.STATUS,
message.getData().getCode(),
peer);
peer.disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
return;
}

// Dispatch eth message
final EthMessage ethMessage = new EthMessage(peer, message.getData());
if (!peer.validateReceivedMessage(ethMessage)) {
LOG.warn("Unsolicited message received from {}, disconnecting", peer);
LOG.debug("Unsolicited message received from, disconnecting: {}", peer);
peer.disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
return;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,8 @@ private void handleGetBlockHeaders(final EthMessage message) {
ethereumWireProtocolConfiguration.getMaxGetBlockHeaders());
message.getPeer().send(response);
} catch (final RLPException e) {
LOG.debug(
"Received malformed GET_BLOCK_HEADERS message, disconnecting: {}", message.getPeer(), e);
message.getPeer().disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
} catch (final PeerNotConnected peerNotConnected) {
// Peer disconnected before we could respond - nothing to do
Expand All @@ -100,6 +102,8 @@ private void handleGetBlockBodies(final EthMessage message) {
ethereumWireProtocolConfiguration.getMaxGetBlockBodies());
message.getPeer().send(response);
} catch (final RLPException e) {
LOG.debug(
"Received malformed GET_BLOCK_BODIES message, disconnecting: {}", message.getPeer(), e);
message.getPeer().disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
} catch (final PeerNotConnected peerNotConnected) {
// Peer disconnected before we could respond - nothing to do
Expand All @@ -114,6 +118,7 @@ private void handleGetReceipts(final EthMessage message) {
blockchain, message.getData(), ethereumWireProtocolConfiguration.getMaxGetReceipts());
message.getPeer().send(response);
} catch (final RLPException e) {
LOG.debug("Received malformed GET_RECEIPTS message, disconnecting: {}", message.getPeer(), e);
message.getPeer().disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
} catch (final PeerNotConnected peerNotConnected) {
// Peer disconnected before we could respond - nothing to do
Expand All @@ -130,6 +135,8 @@ private void handleGetNodeData(final EthMessage message) {
ethereumWireProtocolConfiguration.getMaxGetNodeData());
message.getPeer().send(response);
} catch (final RLPException e) {
LOG.debug(
"Received malformed GET_NODE_DATA message, disconnecting: {}", message.getPeer(), e);
message.getPeer().disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
} catch (final PeerNotConnected peerNotConnected) {
// Peer disconnected before we could respond - nothing to do
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,11 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public abstract class AbstractPeerRequestTask<R> extends AbstractPeerTask<R> {
private static final Logger LOG = LogManager.getLogger();
private static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(5);

private Duration timeout = DEFAULT_TIMEOUT;
Expand Down Expand Up @@ -100,6 +104,7 @@ private void handleMessage(
result.ifPresent(promise::complete);
} catch (final RLPException e) {
// Peer sent us malformed data - disconnect
LOG.debug("Disconnecting with BREACH_OF_PROTOCOL due to malformed message: {}", peer, e);
peer.disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
promise.completeExceptionally(new PeerBreachedProtocolException());
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,10 @@ private void handleNewBlockFromNetwork(final EthMessage message) {

importOrSavePendingBlock(block);
} catch (final RLPException e) {
LOG.debug(
"Malformed NEW_BLOCK message received from peer, disconnecting: {}",
message.getPeer(),
e);
message.getPeer().disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
}
}
Expand Down Expand Up @@ -226,6 +230,10 @@ private void handleNewBlockHashesFromNetwork(final EthMessage message) {
.whenComplete((r, t) -> requestedBlocks.remove(newBlock.hash()));
}
} catch (final RLPException e) {
LOG.debug(
"Malformed NEW_BLOCK_HASHES message received from peer, disconnecting: {}",
message.getPeer(),
e);
message.getPeer().disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,8 @@ public void onPeerConnected(final EthPeer peer) {
peer.chainState().update(chainHeadHeader);
trailingPeerLimiter.enforceTrailingPeerLimit();
} else {
LOG.debug("Failed to retrieve chain head information for " + peer, error);
LOG.debug(
"Failed to retrieve chain head information. Disconnecting {}. ", peer, error);
peer.disconnect(DisconnectReason.USELESS_PEER);
}
});
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,12 @@ private CompletableFuture<Optional<EthPeer>> confirmPivotBlockHeader(final EthPe
.thenApply(
result -> {
if (peerHasDifferentPivotBlock(result)) {
LOG.warn(
"Best peer has wrong pivot block (#{}) expecting {} but received {}. Disconnect: {}",
pivotBlockHeader.getNumber(),
pivotBlockHeader.getHash(),
result.size() == 1 ? result.get(0).getHash() : "invalid response",
bestPeer);
bestPeer.disconnect(DisconnectReason.USELESS_PEER);
return Optional.<EthPeer>empty();
} else {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ private void processTransactionsMessage(
transactionPool.addRemoteTransactions(transactions);
} catch (final RLPException ex) {
if (peer != null) {
LOG.debug("Malformed transaction message received, disconnecting: {}", peer, ex);
peer.disconnect(DisconnectReason.BREACH_OF_PROTOCOL);
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,7 @@ public boolean addMaintainConnectionPeer(final Peer peer) {
public boolean removeMaintainedConnectionPeer(final Peer peer) {
final boolean wasRemoved = maintainedPeers.remove(peer);
peerDiscoveryAgent.dropPeer(peer);
LOG.debug("Disconnect requested for peer {}.", peer);
rlpxAgent.disconnect(peer.getId(), DisconnectReason.REQUESTED);
return wasRemoved;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -305,6 +305,9 @@ private void handlePermissionsUpdate(
connection -> {
if (!peerPermissions.allowOngoingConnection(
connection.getPeer(), connection.initiatedRemotely())) {
LOG.debug(
"Disconnecting from peer that is not permitted to maintain ongoing connection: {}",
connection);
connection.disconnect(DisconnectReason.REQUESTED);
}
});
Expand Down Expand Up @@ -332,21 +335,28 @@ private void handleIncomingConnection(final PeerConnection peerConnection) {
final Peer peer = peerConnection.getPeer();
// Deny connection if our local node isn't ready
if (!localNode.isReady()) {
LOG.debug("Node is not ready. Disconnect incoming connection: {}", peerConnection);
peerConnection.disconnect(DisconnectReason.UNKNOWN);
return;
}
// Disconnect if too many peers
if (!peerPrivileges.canExceedConnectionLimits(peer) && getConnectionCount() >= maxConnections) {
LOG.debug("Too many peers. Disconnect incoming connection: {}", peerConnection);
peerConnection.disconnect(DisconnectReason.TOO_MANY_PEERS);
return;
}
// Disconnect if too many remotely-initiated connections
if (!peerPrivileges.canExceedConnectionLimits(peer) && remoteConnectionLimitReached()) {
LOG.debug(
"Too many remotely-initiated connections. Disconnect incoming connection: {}",
peerConnection);
peerConnection.disconnect(DisconnectReason.TOO_MANY_PEERS);
return;
}
// Disconnect if not permitted
if (!peerPermissions.allowNewInboundConnectionFrom(peer)) {
LOG.debug(
"Node is not permitted to connect. Disconnect incoming connection: {}", peerConnection);
peerConnection.disconnect(DisconnectReason.UNKNOWN);
return;
}
Expand Down Expand Up @@ -426,7 +436,13 @@ private void enforceRemoteConnectionLimits() {
.filter(RlpxConnection::initiatedRemotely)
.filter(conn -> !peerPrivileges.canExceedConnectionLimits(conn.getPeer()))
.skip(maxRemotelyInitiatedConnections)
.forEach(c -> c.disconnect(DisconnectReason.TOO_MANY_PEERS));
.forEach(
conn -> {
LOG.debug(
"Too many remotely initiated connections. Disconnect low-priority connection: {}",
conn);
conn.disconnect(DisconnectReason.TOO_MANY_PEERS);
});
}

private void enforceConnectionLimits() {
Expand All @@ -438,7 +454,11 @@ private void enforceConnectionLimits() {
getActivePrioritizedConnections()
.skip(maxConnections)
.filter(c -> !peerPrivileges.canExceedConnectionLimits(c.getPeer()))
.forEach(c -> c.disconnect(DisconnectReason.TOO_MANY_PEERS));
.forEach(
conn -> {
LOG.debug("Too many connections. Disconnect low-priority connection: {}", conn);
conn.disconnect(DisconnectReason.TOO_MANY_PEERS);
});
}

private Stream<RlpxConnection> getActivePrioritizedConnections() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -200,8 +200,8 @@ public int hashCode() {
@Override
public String toString() {
return MoreObjects.toStringHelper(this)
.add("clientId", peerInfo.getClientId())
.add("nodeId", peerInfo.getNodeId())
.add("clientId", peerInfo.getClientId())
.add(
"caps",
agreedCapabilities.stream().map(Capability::toString).collect(Collectors.joining(", ")))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,15 +82,14 @@ protected void channelRead0(final ChannelHandlerContext ctx, final MessageData o
LOG.debug(
"Received Wire DISCONNECT ({}) from peer: {}",
reason.name(),
connection.getPeerInfo().getClientId());
connection.getPeerInfo());
} catch (final RLPException e) {
LOG.debug(
"Received Wire DISCONNECT with invalid RLP. Peer: {}",
connection.getPeerInfo().getClientId());
"Received Wire DISCONNECT with invalid RLP. Peer: {}", connection.getPeerInfo());
} catch (final Exception e) {
LOG.error(
"Received Wire DISCONNECT, but unable to parse reason. Peer: {}",
connection.getPeerInfo().getClientId(),
connection.getPeerInfo(),
e);
}
connection.terminateConnection(reason, true);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -142,13 +142,13 @@ protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final L
String.format(
"Expected id %s, but got %s", expectedPeer.get().getId(), peerInfo.getNodeId());
connectFuture.completeExceptionally(new UnexpectedPeerConnectionException(unexpectedMsg));
LOG.debug("{}. Disconnecting.", unexpectedMsg);
connection.disconnect(DisconnectMessage.DisconnectReason.UNEXPECTED_ID);
}

// Check that we have shared caps
if (capabilityMultiplexer.getAgreedCapabilities().size() == 0) {
LOG.debug(
"Disconnecting from {} because no capabilities are shared.", peerInfo.getClientId());
LOG.debug("Disconnecting because no capabilities are shared: {}", peerInfo);
connectFuture.completeExceptionally(
new IncompatiblePeerException("No shared capabilities"));
connection.disconnect(DisconnectMessage.DisconnectReason.USELESS_PEER);
Expand Down