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.

Sign up for GitHub

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

Log warnings when cluster state publication failed to some nodes #31233

Merged
merged 3 commits into from
Jun 13, 2018
Merged

Log warnings when cluster state publication failed to some nodes #31233

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
afd5198
warn log publishing errors
bleskes Jun 9, 2018
e73afe0
duplicate
bleskes Jun 10, 2018
c4e1524
rejiggle
bleskes Jun 11, 2018
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
12 changes: 12 additions & 0 deletions ...src/main/java/org/elasticsearch/discovery/BlockingClusterStatePublishResponseHandler.java
View file
Open in desktop
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);
}
}
11 changes: 10 additions & 1 deletion server/src/main/java/org/elasticsearch/discovery/zen/PublishClusterStateAction.java
View file
Open in desktop
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 @@ -397,6 +403,9 @@ protected void handleIncomingClusterStateRequest(BytesTransportRequest request,
} catch (IncompatibleClusterStateVersionException e) {
incompatibleClusterStateDiffReceivedCount.incrementAndGet();
throw e;
} catch (Exception e) {
Copy link
Contributor

Choose a reason for hiding this comment

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

This will also catch exceptions when calling incomingClusterStateListener.onIncomingClusterState, which in return calls ZenDiscovery.validateIncomingState. I think it's confusing to catch these as well here and erroneously reporting them here as deserialization errors.
Maybe it would be better to reduce the scope of the catch clause...

Copy link
Contributor Author

Choose a reason for hiding this comment

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

That's a great comment. I missed that call. Let me see how I can shuffle things.

Copy link
Member

Choose a reason for hiding this comment

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

Good catch @ywelsch.

logger.warn("unexpected error while deserializing an incoming cluster state", e);
throw e;
} finally {
IOUtils.close(in);
}
Expand Down
19 changes: 15 additions & 4 deletions ...est/java/org/elasticsearch/discovery/BlockingClusterStatePublishResponseHandlerTests.java
View file
Open in desktop
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));
}
}