Skip to content

Commit

Permalink
Log warnings when cluster state publication failed to some nodes (#31233
Browse files Browse the repository at this point in the history 
)

If the publishing of a cluster state to a node fails, we currently only log it as debug information and 
only on the master. This makes it hard to see the cause of (test) failures when logging is set to 
default levels. This PR adds a warn level log on the node receiving the cluster state when it fails to 
deserialise the cluster state and a warn level log on the master with a list of nodes for which 
publication failed.
  • Loading branch information
@bleskes
bleskes committed Jun 13, 2018
1 parent d7b11f6 commit be72fbc
Show file tree
Hide file tree
Showing 3 changed files with 52 additions and 20 deletions.
12 changes: 12 additions & 0 deletions ...src/main/java/org/elasticsearch/discovery/BlockingClusterStatePublishResponseHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;

import java.util.Collections;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
Expand All @@ -35,6 +36,7 @@ public class BlockingClusterStatePublishResponseHandler {

private final CountDownLatch latch;
private final Set<DiscoveryNode> pendingNodes;
private final Set<DiscoveryNode> failedNodes;

/**
* Creates a new BlockingClusterStatePublishResponseHandler
Expand All @@ -44,6 +46,7 @@ public BlockingClusterStatePublishResponseHandler(Set<DiscoveryNode> publishingT
this.pendingNodes = ConcurrentCollections.newConcurrentSet();
this.pendingNodes.addAll(publishingToNodes);
this.latch = new CountDownLatch(pendingNodes.size());
this.failedNodes = ConcurrentCollections.newConcurrentSet();
}

/**
Expand All @@ -64,6 +67,8 @@ public void onResponse(DiscoveryNode node) {
public void onFailure(DiscoveryNode node, Exception e) {
boolean found = pendingNodes.remove(node);
assert found : "node [" + node + "] already responded or failed";
boolean added = failedNodes.add(node);
assert added : "duplicate failures for " + node;
latch.countDown();
}

Expand All @@ -86,4 +91,11 @@ public DiscoveryNode[] pendingNodes() {
// nulls if some nodes responded in the meanwhile
return pendingNodes.toArray(new DiscoveryNode[0]);
}

/**
* returns a set of nodes for which publication has failed.
*/
public Set<DiscoveryNode> getFailedNodes() {
return Collections.unmodifiableSet(failedNodes);
}
}
41 changes: 25 additions & 16 deletions server/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
package org.elasticsearch.discovery.zen;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.Version;
import org.elasticsearch.action.ActionListener;
Expand All @@ -41,6 +40,7 @@
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.discovery.AckClusterStatePublishResponseHandler;
import org.elasticsearch.discovery.BlockingClusterStatePublishResponseHandler;
import org.elasticsearch.discovery.Discovery;
Expand Down Expand Up @@ -207,6 +207,12 @@ private void innerPublish(final ClusterChangedEvent clusterChangedEvent, final S
clusterState.version(), publishTimeout, pendingNodes);
}
}
// The failure is logged under debug when a sending failed. we now log a summary.
Set<DiscoveryNode> failedNodes = publishResponseHandler.getFailedNodes();
if (failedNodes.isEmpty() == false) {
logger.warn("publishing cluster state with version [{}] failed for the following nodes: [{}]",
clusterChangedEvent.state().version(), failedNodes);
}
} catch (InterruptedException e) {
// ignore & restore interrupt
Thread.currentThread().interrupt();
Expand Down Expand Up @@ -367,14 +373,14 @@ public static BytesReference serializeDiffClusterState(Diff diff, Version nodeVe
protected void handleIncomingClusterStateRequest(BytesTransportRequest request, TransportChannel channel) throws IOException {
Compressor compressor = CompressorFactory.compressor(request.bytes());
StreamInput in = request.bytes().streamInput();
try {
if (compressor != null) {
in = compressor.streamInput(in);
}
in = new NamedWriteableAwareStreamInput(in, namedWriteableRegistry);
in.setVersion(request.version());
synchronized (lastSeenClusterStateMutex) {
final ClusterState incomingState;
final ClusterState incomingState;
synchronized (lastSeenClusterStateMutex) {
try {
if (compressor != null) {
in = compressor.streamInput(in);
}
in = new NamedWriteableAwareStreamInput(in, namedWriteableRegistry);
in.setVersion(request.version());
// If true we received full cluster state - otherwise diffs
if (in.readBoolean()) {
incomingState = ClusterState.readFrom(in, transportService.getLocalNode());
Expand All @@ -391,14 +397,17 @@ protected void handleIncomingClusterStateRequest(BytesTransportRequest request,
logger.debug("received diff for but don't have any local cluster state - requesting full state");
throw new IncompatibleClusterStateVersionException("have no local cluster state");
}
incomingClusterStateListener.onIncomingClusterState(incomingState);
lastSeenClusterState = incomingState;
} catch (IncompatibleClusterStateVersionException e) {
incompatibleClusterStateDiffReceivedCount.incrementAndGet();
throw e;
} catch (Exception e) {
logger.warn("unexpected error while deserializing an incoming cluster state", e);
throw e;
} finally {
IOUtils.close(in);
}
} catch (IncompatibleClusterStateVersionException e) {
incompatibleClusterStateDiffReceivedCount.incrementAndGet();
throw e;
} finally {
IOUtils.close(in);
incomingClusterStateListener.onIncomingClusterState(incomingState);
lastSeenClusterState = incomingState;
}
channel.sendResponse(TransportResponse.Empty.INSTANCE);
}
Expand Down
19 changes: 15 additions & 4 deletions ...est/java/org/elasticsearch/discovery/BlockingClusterStatePublishResponseHandlerTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,10 +85,16 @@ public void testConcurrentAccess() throws InterruptedException {
int firstRound = randomIntBetween(5, nodeCount - 1);
Thread[] threads = new Thread[firstRound];
CyclicBarrier barrier = new CyclicBarrier(firstRound);
Set<DiscoveryNode> expectedFailures = new HashSet<>();
Set<DiscoveryNode> completedNodes = new HashSet<>();
for (int i = 0; i < threads.length; i++) {
completedNodes.add(allNodes[i]);
threads[i] = new Thread(new PublishResponder(randomBoolean(), allNodes[i], barrier, logger, handler));
final DiscoveryNode node = allNodes[i];
completedNodes.add(node);
final boolean fail = randomBoolean();
if (fail) {
expectedFailures.add(node);
}
threads[i] = new Thread(new PublishResponder(fail, node, barrier, logger, handler));
threads[i].start();
}
// wait on the threads to finish
Expand All @@ -105,7 +111,12 @@ public void testConcurrentAccess() throws InterruptedException {
barrier = new CyclicBarrier(secondRound);

for (int i = 0; i < threads.length; i++) {
threads[i] = new Thread(new PublishResponder(randomBoolean(), allNodes[firstRound + i], barrier, logger, handler));
final DiscoveryNode node = allNodes[firstRound + i];
final boolean fail = randomBoolean();
if (fail) {
expectedFailures.add(node);
}
threads[i] = new Thread(new PublishResponder(fail, node, barrier, logger, handler));
threads[i].start();
}
// wait on the threads to finish
Expand All @@ -114,6 +125,6 @@ public void testConcurrentAccess() throws InterruptedException {
}
assertTrue("expected handler not to timeout as all nodes responded", handler.awaitAllNodes(new TimeValue(10)));
assertThat(handler.pendingNodes(), arrayWithSize(0));

assertThat(handler.getFailedNodes(), equalTo(expectedFailures));
}
}

0 comments on commit be72fbc

Please sign in to comment.