add logging

Author: pxsalehi

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/allocation/AllocationFailuresResetIT.java

@@ -9,9 +9,12 @@
 
 package org.elasticsearch.cluster.routing.allocation;
 
+import org.apache.logging.log4j.Level;
+import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.ShardRoutingState;
 import org.elasticsearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.shard.IndexEventListener;
@@ -20,6 +23,7 @@
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
 import org.elasticsearch.test.MockIndexEventListener;
+import org.elasticsearch.test.MockLog;
 
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -86,8 +90,20 @@ public void testResetAllocationFailuresOnNodeJoin() throws Exception {
  prepareCreate(INDEX, indexSettings(1, 0)).execute();
  awaitShardAllocMaxRetries();
  removeAllocationFailuresInjection(node1);
- internalCluster().startNode();
- awaitShardAllocSucceed();
+ try (var mockLog = MockLog.capture(RoutingNodes.class)) {
+ var shardId = internalCluster().clusterService().state().routingTable().index(INDEX).shard(SHARD).shardId();
+ mockLog.addExpectation(
+ new MockLog.SeenEventExpectation(
+ "log resetting failed allocations",
+ RoutingNodes.class.getName(),
+ Level.INFO,
+ Strings.format(RoutingNodes.RESET_FAILED_ALLOCATION_COUNTER_LOG_MSG, 1, List.of(shardId))
+ )
+ );
+ internalCluster().startNode();
+ awaitShardAllocSucceed();
+ mockLog.assertAllExpectationsMatched();
+ }
  }
 
  public void testResetRelocationFailuresOnNodeJoin() throws Exception {
@@ -122,12 +138,23 @@ public void beforeIndexCreated(Index index, Settings indexSettings) {
  assertThat(state.nodes().get(shard.currentNodeId()).getName(), equalTo(node1));
  failRelocation.set(false);
  // A new node joining should reset the counter and allow more relocation retries
- internalCluster().startNode();
- assertBusy(() -> {
- var stateAfterNodeJoin = internalCluster().clusterService().state();
- var relocatedShard = stateAfterNodeJoin.routingTable().index(INDEX).shard(SHARD).primaryShard();
- assertThat(relocatedShard, notNullValue());
- assertThat(stateAfterNodeJoin.nodes().get(relocatedShard.currentNodeId()).getName(), not(equalTo(node1)));
- });
+ try (var mockLog = MockLog.capture(RoutingNodes.class)) {
+ mockLog.addExpectation(
+ new MockLog.SeenEventExpectation(
+ "log resetting failed relocations",
+ RoutingNodes.class.getName(),
+ Level.INFO,
+ Strings.format(RoutingNodes.RESET_FAILED_RELOCATION_COUNTER_LOG_MSG, 1, List.of(shard.shardId()))
+ )
+ );
+ internalCluster().startNode();
+ assertBusy(() -> {
+ var stateAfterNodeJoin = internalCluster().clusterService().state();
+ var relocatedShard = stateAfterNodeJoin.routingTable().index(INDEX).shard(SHARD).primaryShard();
+ assertThat(relocatedShard, notNullValue());
+ assertThat(stateAfterNodeJoin.nodes().get(relocatedShard.currentNodeId()).getName(), not(equalTo(node1)));
+ });
+ mockLog.assertAllExpectationsMatched();
+ }
  }
 }

server/src/main/java/org/elasticsearch/cluster/routing/RoutingNodes.java

@@ -16,14 +16,19 @@
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.UnassignedInfo.AllocationStatus;
 import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
+import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.allocator.DesiredBalanceMetrics;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
 
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -44,6 +49,8 @@
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+import static org.elasticsearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider.SETTING_ALLOCATION_MAX_RETRY;
+
 /**
  * {@link RoutingNodes} represents a copy the routing information contained in the {@link ClusterState cluster state}.
  * It can be either initialized as mutable or immutable allowing or disallowing changes to its elements.
@@ -60,6 +67,13 @@
  */
 public class RoutingNodes implements Iterable<RoutingNode> {
 
+ private static final Logger logger = LogManager.getLogger(RoutingNodes.class);
+ public static final String RESET_FAILED_ALLOCATION_COUNTER_LOG_MSG =
+ "Resetting failure counter for %d shard(s) that have reached their max allocation retires (%s)";
+ public static final String RESET_FAILED_RELOCATION_COUNTER_LOG_MSG =
+ "Resetting failure counter for %d shard(s) that have reached their max relocation retries (%s)";
+ private static final int MAX_SHARDS_IN_LOG_MSG = 20;
+
  private final Map<String, RoutingNode> nodesToShards;
 
  private final UnassignedShards unassignedShards;
@@ -1309,14 +1323,36 @@ public boolean hasRelocationFailures() {
  return false;
  }
 
- public void resetFailedCounter(RoutingChangesObserver routingChangesObserver) {
+ public void resetFailedCounter(RoutingAllocation allocation) {
+ final var observer = allocation.changes();
+ int shardsWithMaxFailedAllocations = 0;
+ int shardsWithMaxFailedRelocations = 0;
+ List<ShardId> topShardIdsWithFailedAllocations = new ArrayList<>();
+ List<ShardId> topShardIdsWithFailedRelocations = new ArrayList<>();
+
  final var unassignedIterator = unassigned().iterator();
  while (unassignedIterator.hasNext()) {
  ShardRouting shardRouting = unassignedIterator.next();
  UnassignedInfo unassignedInfo = shardRouting.unassignedInfo();
+ int failedAllocations = unassignedInfo.failedAllocations();
+ if (failedAllocations > 0) {
+ try {
+ final var maxRetry = SETTING_ALLOCATION_MAX_RETRY.get(
+ allocation.metadata().getIndexSafe(shardRouting.index()).getSettings()
+ );
+ if (failedAllocations >= maxRetry) {
+ shardsWithMaxFailedAllocations++;
+ if (topShardIdsWithFailedAllocations.size() <= MAX_SHARDS_IN_LOG_MSG) {
+ topShardIdsWithFailedAllocations.add(shardRouting.shardId());
+ }
+ }
+ } catch (IndexNotFoundException e) {
+ // ignore
+ }
+ }
  unassignedIterator.updateUnassigned(
  new UnassignedInfo(
- unassignedInfo.failedAllocations() > 0 ? UnassignedInfo.Reason.MANUAL_ALLOCATION : unassignedInfo.reason(),
+ failedAllocations > 0 ? UnassignedInfo.Reason.MANUAL_ALLOCATION : unassignedInfo.reason(),
  unassignedInfo.message(),
  unassignedInfo.failure(),
  0,
@@ -1328,7 +1364,7 @@ public void resetFailedCounter(RoutingChangesObserver routingChangesObserver) {
  unassignedInfo.lastAllocatedNodeId()
  ),
  shardRouting.recoverySource(),
- routingChangesObserver
+ observer
  );
  }
 
@@ -1337,6 +1373,20 @@ public void resetFailedCounter(RoutingChangesObserver routingChangesObserver) {
  for (ShardRouting shardRouting : routingNode) {
  if (shardRouting.relocationFailureInfo() != null && shardRouting.relocationFailureInfo().failedRelocations() > 0) {
  shardsWithRelocationFailures.add(shardRouting);
+ try {
+ int failedRelocations = shardRouting.relocationFailureInfo().failedRelocations();
+ final var maxRetry = SETTING_ALLOCATION_MAX_RETRY.get(
+ allocation.metadata().getIndexSafe(shardRouting.index()).getSettings()
+ );
+ if (failedRelocations >= maxRetry) {
+ shardsWithMaxFailedRelocations++;
+ if (topShardIdsWithFailedRelocations.size() <= MAX_SHARDS_IN_LOG_MSG) {
+ topShardIdsWithFailedRelocations.add(shardRouting.shardId());
+ }
+ }
+ } catch (IndexNotFoundException e) {
+ // ignore
+ }
  }
  }
 
@@ -1347,6 +1397,17 @@ public void resetFailedCounter(RoutingChangesObserver routingChangesObserver) {
  assignedShardsAdd(updated);
  }
  }
+
+ if (shardsWithMaxFailedAllocations > 0) {
+ logger.info(
+ Strings.format(RESET_FAILED_ALLOCATION_COUNTER_LOG_MSG, shardsWithMaxFailedAllocations, topShardIdsWithFailedAllocations)
+ );
+ }
+ if (shardsWithMaxFailedRelocations > 0) {
+ logger.info(
+ Strings.format(RESET_FAILED_RELOCATION_COUNTER_LOG_MSG, shardsWithMaxFailedRelocations, topShardIdsWithFailedRelocations)
+ );
+ }
  }
 
  /**

server/src/main/java/org/elasticsearch/cluster/routing/allocation/AllocationService.java

@@ -639,7 +639,7 @@ private boolean shouldResetAllocationFailures(ClusterChangedEvent changeEvent) {
 
  private ClusterState rerouteWithResetFailedCounter(ClusterState clusterState) {
  RoutingAllocation allocation = createRoutingAllocation(clusterState, currentNanoTime());
- allocation.routingNodes().resetFailedCounter(allocation.changes());
+ allocation.routingNodes().resetFailedCounter(allocation);
  reroute(allocation, routingAllocation -> shardsAllocator.allocate(routingAllocation, ActionListener.noop()));
  return buildResultAndLogHealthChange(clusterState, allocation, "reroute with reset failed counter");
  }

server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/ShardsAllocator.java

@@ -62,7 +62,7 @@ default RoutingExplanations execute(RoutingAllocation allocation, AllocationComm
 
  try {
  if (retryFailed) {
- allocation.routingNodes().resetFailedCounter(allocation.changes());
+ allocation.routingNodes().resetFailedCounter(allocation);
  }
  return commands.execute(allocation, explain);
  } finally {