Ensure ILM policies run safely on leader indices

[gwbrown]

Adds a Step to the Shrink and Delete actions which prevents those
actions from running on a leader index - all follower indices must first
unfollow the leader index before these actions can run. This prevents
the loss of history before follower indices are ready, which might
otherwise result in the loss of data.

Relates to #34648

---

There will likely be a change to the integration test, as I'm still seeing occasional failures in it, which don't appear to be "real" failures (just hitting timeouts sometimes), but the code is otherwise ready to review.

[elasticmachine]

Pinging @elastic/es-core-features

[gwbrown]

This will also need to be updated after #37951 is merged, as that PR contains changes which will cause this one to fail to build. The changes should be pretty minor, though.

[martijnvg]

I chatted about this test @gwbrown. The stopping of ilm in this test can be problematic for other tests that run after it. First all tests are executed as part of the leader cluster test task and then as part of the follower cluster test task (this while leader test cluster is running). The other which the test methods are executed is randomized by the test framework.

In order to avoid this problem the entire test can be ran just against the follow cluster test task, like is done below.
Also what also is possible is to remove the stop and start ilm calls and in instead setup the leader index first in leader cluster, execute put follow in follower cluster and then create the leader ilm policy.

if ("leader".equals(targetCluster)) {
            return;
        } else if ("follow".equals(targetCluster) == false) {
            throw  new AssertionError("unexpected target cluster [" + targetCluster + "]");
        }

        String indexName = "shrink-leader-test";
        String shrunkenIndexName = "shrink-" + indexName;
        String policyName = "shrink-leader-test-policy";
        try (RestClient leaderClient = buildLeaderClient()) {

            putShrinkOnlyPolicy(leaderClient, policyName);

            // Stop ILM to let us set up the follower before the policy swings into action and shrinks the index
            stopILM(client());
            assertBusy(() -> assertEquals("STOPPED", ilmStatus(client())));
            Settings indexSettings = Settings.builder()
                .put("index.soft_deletes.enabled", true)
                .put("index.number_of_shards", 2)
                .put("index.number_of_replicas", 0)
                .put("index.lifecycle.name", policyName)
                .build();
            createIndex(indexName, indexSettings, "", "", leaderClient); // <-- method does not yet exist
            ensureGreen(indexName, leaderClient); // <-- method does not yet exist

            // Policy with the same name must exist in follower cluster too:
            putUnfollowOnlyPolicy(client(), policyName);
            followIndex(indexName, indexName);
            ensureGreen(indexName);

            startILM(leaderClient);
            index(leaderClient, indexName, "1");
            assertDocumentExists(leaderClient, indexName, "1");

            assertBusy(() -> {
                assertDocumentExists(client(), indexName, "1");
                // Sanity check that following_index setting has been set, so that we can verify later that this setting has been unset:
                assertThat(getIndexSetting(client(), indexName, "index.xpack.ccr.following_index"), equalTo("true"));

                // We should get into a state with these policies where both leader and followers are waiting on each other
                assertILMPolicy(leaderClient, indexName, policyName, "warm", "shrink", "wait-for-shard-history-leases");
                assertILMPolicy(client(), indexName, policyName, "hot", "unfollow", "wait-for-indexing-complete");
            });

            // Manually set this to kick the process
            updateIndexSettings(leaderClient, indexName, Settings.builder()
                .put("index.lifecycle.indexing_complete", true)
                .build()
            );

            assertBusy(() -> {
                // The shrunken index should now be created on the leader...
                Response shrunkenIndexExistsResponse = leaderClient.performRequest(new Request("HEAD", "/" + shrunkenIndexName));
                assertEquals(RestStatus.OK.getStatus(), shrunkenIndexExistsResponse.getStatusLine().getStatusCode());

                // And both of these should now finish their policies
                assertILMPolicy(leaderClient, shrunkenIndexName, policyName, "completed");
                assertILMPolicy(client(), indexName, policyName, "completed");
            });
        }

[dakrone]

I left three really minor comments, otherwise LGTM

[dakrone]

Can you add javadoc please?

[dakrone]

this can probably be a static final field since it's hardcoded?

[dakrone]

This can be return true; since the messages are never different?

[gwbrown]

@elasticmachine run elasticsearch-ci/default-distro

[gwbrown]

Thanks for the reviews and thanks for handling the merge conflict @jasontedor!