ILM: ignore clusterChanged events if state not recovered (#67507)

IndexLifecycleService will listen for clusterChanged events in order to
trigger the async actions when a new master is elected. During the same
event processing routine determines when a new master was elected (to avoid
triggering the async actions multiple times).
However, if it doesn't exclude the events based on not fully recovered
cluster states it will get into a situation where it records the "new master elected"
but it cannot trigger the async actions because the cluster state does not
contain the `IndexLifecycleMetadata` custom.

Author: andreidan

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycleService.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.shard.IndexEventListener;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -222,6 +223,11 @@ private synchronized void maybeScheduleJob() {
 
     @Override
     public void clusterChanged(ClusterChangedEvent event) {
+        // wait for the cluster state to be recovered so the ILM policies are present
+        if (event.state().blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
+            return;
+        }
+
         // Instead of using a LocalNodeMasterListener to track master changes, this service will
         // track them here to avoid conditions where master listener events run after other
         // listeners that depend on what happened in the master listener

x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/ilm/IndexLifecycleServiceTests.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -59,6 +60,8 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import static java.time.Clock.systemUTC;
+import static org.elasticsearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
 import static org.elasticsearch.xpack.core.ilm.AbstractStepTestCase.randomStepKey;
 import static org.elasticsearch.xpack.core.ilm.LifecycleExecutionState.ILM_CUSTOM_METADATA_KEY;
 import static org.elasticsearch.xpack.ilm.LifecyclePolicyTestsUtils.newTestLifecyclePolicy;
@@ -449,6 +452,27 @@ public void doTestExceptionStillProcessesOtherIndices(boolean useOnMaster) {
         }
     }
 
+    public void testClusterChangedWaitsForTheStateToBeRecovered() {
+        IndexLifecycleService ilmService = new IndexLifecycleService(Settings.EMPTY, mock(Client.class), clusterService, threadPool,
+            systemUTC(), () -> now, null, null) {
+
+            @Override
+            void onMaster(ClusterState clusterState) {
+                fail("IndexLifecycleService ignored the global [state not recovered / initialized] cluster block");
+            }
+
+            @Override
+            void triggerPolicies(ClusterState clusterState, boolean fromClusterStateChange) {
+                fail("IndexLifecycleService ignored the global [state not recovered / initialized] cluster block");
+            }
+        };
+
+        ClusterState currentState = ClusterState.builder(ClusterName.DEFAULT)
+            .blocks(ClusterBlocks.builder().addGlobalBlock(STATE_NOT_RECOVERED_BLOCK).build())
+            .build();
+        ilmService.clusterChanged(new ClusterChangedEvent("_source", currentState, ClusterState.EMPTY_STATE));
+    }
+
     public void testTriggeredDifferentJob() {
         Mockito.reset(clusterService);
         SchedulerEngine.Event schedulerEvent = new SchedulerEngine.Event("foo", randomLong(), randomLong());