[CCR] Added history uuid validation (#33546)

For correctness we need to verify whether the history uuid of the leader
index shards never changes while that index is being followed.

* The history UUIDs are recorded as custom index metadata in the follow index.
* The follow api validates whether the current history UUIDs of the leader
  index shards are the same as the recorded history UUIDs.
  If not the follow api fails.
* While a follow index is following a leader index; shard follow tasks
  on each shard changes api call verify whether their current history uuid
  is the same as the recorded history uuid.

Relates to #30086 

Co-authored-by: Nhat Nguyen <nhat.nguyen@elastic.co>

test/framework/src/main/java/org/elasticsearch/index/replication/ESIndexLevelReplicationTestCase.java

@@ -443,6 +443,10 @@ public IndexShard getPrimary() {
             return primary;
         }
 
+        public synchronized void reinitPrimaryShard() throws IOException {
+            primary = reinitShard(primary);
+        }
+
         public void syncGlobalCheckpoint() {
             PlainActionFuture<ReplicationResponse> listener = new PlainActionFuture<>();
             try {

x-pack/plugin/ccr/qa/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/ccr/FollowIndexSecurityIT.java

@@ -113,7 +113,7 @@ public void testFollowIndex() throws Exception {
 
             e = expectThrows(ResponseException.class,
                 () -> followIndex("leader_cluster:" + unallowedIndex, unallowedIndex));
-            assertThat(e.getMessage(), containsString("follow index [" + unallowedIndex + "] does not exist"));
+            assertThat(e.getMessage(), containsString("action [indices:monitor/stats] is unauthorized for user [test_ccr]"));
             assertThat(indexExists(adminClient(), unallowedIndex), is(false));
             assertBusy(() -> assertThat(countCcrNodeTasks(), equalTo(0)));
         }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/Ccr.java

@@ -85,6 +85,8 @@
 public class Ccr extends Plugin implements ActionPlugin, PersistentTaskPlugin, EnginePlugin {
 
     public static final String CCR_THREAD_POOL_NAME = "ccr";
+    public static final String CCR_CUSTOM_METADATA_KEY = "ccr";
+    public static final String CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS = "leader_index_shard_history_uuids";
 
     private final boolean enabled;
     private final Settings settings;

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/CcrLicenseChecker.java

@@ -10,9 +10,18 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
+import org.elasticsearch.action.admin.indices.stats.IndexShardStats;
+import org.elasticsearch.action.admin.indices.stats.IndexStats;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
+import org.elasticsearch.action.admin.indices.stats.ShardStats;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.index.engine.CommitStats;
+import org.elasticsearch.index.engine.Engine;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.license.RemoteClusterLicenseChecker;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.rest.RestStatus;
@@ -21,6 +30,7 @@
 import java.util.Collections;
 import java.util.Locale;
 import java.util.Objects;
+import java.util.function.BiConsumer;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -58,23 +68,24 @@ public boolean isCcrAllowed() {
     }
 
     /**
-     * Fetches the leader index metadata from the remote cluster. Before fetching the index metadata, the remote cluster is checked for
-     * license compatibility with CCR. If the remote cluster is not licensed for CCR, the {@code onFailure} consumer is is invoked.
-     * Otherwise, the specified consumer is invoked with the leader index metadata fetched from the remote cluster.
+     * Fetches the leader index metadata and history UUIDs for leader index shards from the remote cluster.
+     * Before fetching the index metadata, the remote cluster is checked for license compatibility with CCR.
+     * If the remote cluster is not licensed for CCR, the {@code onFailure} consumer is is invoked. Otherwise,
+     * the specified consumer is invoked with the leader index metadata fetched from the remote cluster.
      *
-     * @param client                      the client
-     * @param clusterAlias                the remote cluster alias
-     * @param leaderIndex                 the name of the leader index
-     * @param onFailure                   the failure consumer
-     * @param leaderIndexMetadataConsumer the leader index metadata consumer
-     * @param <T>                         the type of response the listener is waiting for
+     * @param client        the client
+     * @param clusterAlias  the remote cluster alias
+     * @param leaderIndex   the name of the leader index
+     * @param onFailure     the failure consumer
+     * @param consumer      the consumer for supplying the leader index metadata and historyUUIDs of all leader shards
+     * @param <T>           the type of response the listener is waiting for
      */
-    public <T> void checkRemoteClusterLicenseAndFetchLeaderIndexMetadata(
+    public <T> void checkRemoteClusterLicenseAndFetchLeaderIndexMetadataAndHistoryUUIDs(
             final Client client,
             final String clusterAlias,
             final String leaderIndex,
             final Consumer<Exception> onFailure,
-            final Consumer<IndexMetaData> leaderIndexMetadataConsumer) {
+            final BiConsumer<String[], IndexMetaData> consumer) {
 
         final ClusterStateRequest request = new ClusterStateRequest();
         request.clear();
@@ -85,7 +96,13 @@ public <T> void checkRemoteClusterLicenseAndFetchLeaderIndexMetadata(
                 clusterAlias,
                 request,
                 onFailure,
-                leaderClusterState -> leaderIndexMetadataConsumer.accept(leaderClusterState.getMetaData().index(leaderIndex)),
+                leaderClusterState -> {
+                    IndexMetaData leaderIndexMetaData = leaderClusterState.getMetaData().index(leaderIndex);
+                    final Client leaderClient = client.getRemoteClusterClient(clusterAlias);
+                    fetchLeaderHistoryUUIDs(leaderClient, leaderIndexMetaData, onFailure, historyUUIDs -> {
+                        consumer.accept(historyUUIDs, leaderIndexMetaData);
+                    });
+                },
                 licenseCheck -> indexMetadataNonCompliantRemoteLicense(leaderIndex, licenseCheck),
                 e -> indexMetadataUnknownRemoteLicense(leaderIndex, clusterAlias, e));
     }
@@ -168,6 +185,58 @@ public void onFailure(final Exception e) {
                 });
     }
 
+    /**
+     * Fetches the history UUIDs for leader index on per shard basis using the specified leaderClient.
+     *
+     * @param leaderClient                              the leader client
+     * @param leaderIndexMetaData                       the leader index metadata
+     * @param onFailure                                 the failure consumer
+     * @param historyUUIDConsumer                       the leader index history uuid and consumer
+     */
+    // NOTE: Placed this method here; in order to avoid duplication of logic for fetching history UUIDs
+    // in case of following a local or a remote cluster.
+    public void fetchLeaderHistoryUUIDs(
+        final Client leaderClient,
+        final IndexMetaData leaderIndexMetaData,
+        final Consumer<Exception> onFailure,
+        final Consumer<String[]> historyUUIDConsumer) {
+
+        String leaderIndex = leaderIndexMetaData.getIndex().getName();
+        CheckedConsumer<IndicesStatsResponse, Exception> indicesStatsHandler = indicesStatsResponse -> {
+            IndexStats indexStats = indicesStatsResponse.getIndices().get(leaderIndex);
+            String[] historyUUIDs = new String[leaderIndexMetaData.getNumberOfShards()];
+            for (IndexShardStats indexShardStats : indexStats) {
+                for (ShardStats shardStats : indexShardStats) {
+                    // Ignore replica shards as they may not have yet started and
+                    // we just end up overwriting slots in historyUUIDs
+                    if (shardStats.getShardRouting().primary() == false) {
+                        continue;
+                    }
+
+                    CommitStats commitStats = shardStats.getCommitStats();
+                    if (commitStats == null) {
+                        onFailure.accept(new IllegalArgumentException("leader index's commit stats are missing"));
+                        return;
+                    }
+                    String historyUUID = commitStats.getUserData().get(Engine.HISTORY_UUID_KEY);
+                    ShardId shardId = shardStats.getShardRouting().shardId();
+                    historyUUIDs[shardId.id()] = historyUUID;
+                }
+            }
+            for (int i = 0; i < historyUUIDs.length; i++) {
+                if (historyUUIDs[i] == null) {
+                    onFailure.accept(new IllegalArgumentException("no history uuid for [" + leaderIndex + "][" + i + "]"));
+                    return;
+                }
+            }
+            historyUUIDConsumer.accept(historyUUIDs);
+        };
+        IndicesStatsRequest request = new IndicesStatsRequest();
+        request.clear();
+        request.indices(leaderIndex);
+        leaderClient.admin().indices().stats(request, ActionListener.wrap(indicesStatsHandler, onFailure));
+    }
+
     private static ElasticsearchStatusException indexMetadataNonCompliantRemoteLicense(
             final String leaderIndex, final RemoteClusterLicenseChecker.LicenseCheck licenseCheck) {
         final String clusterAlias = licenseCheck.remoteClusterLicenseInfo().clusterAlias();

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardChangesAction.java

@@ -58,11 +58,13 @@ public static class Request extends SingleShardRequest<Request> {
         private long fromSeqNo;
         private int maxOperationCount;
         private ShardId shardId;
+        private String expectedHistoryUUID;
         private long maxOperationSizeInBytes = FollowIndexAction.DEFAULT_MAX_BATCH_SIZE_IN_BYTES;
 
-        public Request(ShardId shardId) {
+        public Request(ShardId shardId, String expectedHistoryUUID) {
             super(shardId.getIndexName());
             this.shardId = shardId;
+            this.expectedHistoryUUID = expectedHistoryUUID;
         }
 
         Request() {
@@ -96,6 +98,10 @@ public void setMaxOperationSizeInBytes(long maxOperationSizeInBytes) {
             this.maxOperationSizeInBytes = maxOperationSizeInBytes;
         }
 
+        public String getExpectedHistoryUUID() {
+            return expectedHistoryUUID;
+        }
+
         @Override
         public ActionRequestValidationException validate() {
             ActionRequestValidationException validationException = null;
@@ -119,6 +125,7 @@ public void readFrom(StreamInput in) throws IOException {
             fromSeqNo = in.readVLong();
             maxOperationCount = in.readVInt();
             shardId = ShardId.readShardId(in);
+            expectedHistoryUUID = in.readString();
             maxOperationSizeInBytes = in.readVLong();
         }
 
@@ -128,6 +135,7 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeVLong(fromSeqNo);
             out.writeVInt(maxOperationCount);
             shardId.writeTo(out);
+            out.writeString(expectedHistoryUUID);
             out.writeVLong(maxOperationSizeInBytes);
         }
 
@@ -140,12 +148,13 @@ public boolean equals(final Object o) {
             return fromSeqNo == request.fromSeqNo &&
                     maxOperationCount == request.maxOperationCount &&
                     Objects.equals(shardId, request.shardId) &&
+                    Objects.equals(expectedHistoryUUID, request.expectedHistoryUUID) &&
                     maxOperationSizeInBytes == request.maxOperationSizeInBytes;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(fromSeqNo, maxOperationCount, shardId, maxOperationSizeInBytes);
+            return Objects.hash(fromSeqNo, maxOperationCount, shardId, expectedHistoryUUID, maxOperationSizeInBytes);
         }
 
         @Override
@@ -154,6 +163,7 @@ public String toString() {
                     "fromSeqNo=" + fromSeqNo +
                     ", maxOperationCount=" + maxOperationCount +
                     ", shardId=" + shardId +
+                    ", expectedHistoryUUID=" + expectedHistoryUUID +
                     ", maxOperationSizeInBytes=" + maxOperationSizeInBytes +
                     '}';
         }
@@ -189,7 +199,12 @@ public Translog.Operation[] getOperations() {
         Response() {
         }
 
-        Response(final long mappingVersion, final long globalCheckpoint, final long maxSeqNo, final Translog.Operation[] operations) {
+        Response(
+            final long mappingVersion,
+            final long globalCheckpoint,
+            final long maxSeqNo,
+            final Translog.Operation[] operations) {
+
             this.mappingVersion = mappingVersion;
             this.globalCheckpoint = globalCheckpoint;
             this.maxSeqNo = maxSeqNo;
@@ -260,6 +275,7 @@ protected Response shardOperation(Request request, ShardId shardId) throws IOExc
                     seqNoStats.getGlobalCheckpoint(),
                     request.fromSeqNo,
                     request.maxOperationCount,
+                    request.expectedHistoryUUID,
                     request.maxOperationSizeInBytes);
             return new Response(mappingVersion, seqNoStats.getGlobalCheckpoint(), seqNoStats.getMaxSeqNo(), operations);
         }
@@ -293,11 +309,20 @@ protected Response newResponse() {
      * Also if the sum of collected operations' size is above the specified maxOperationSizeInBytes then this method
      * stops collecting more operations and returns what has been collected so far.
      */
-    static Translog.Operation[] getOperations(IndexShard indexShard, long globalCheckpoint, long fromSeqNo, int maxOperationCount,
+    static Translog.Operation[] getOperations(IndexShard indexShard,
+                                              long globalCheckpoint,
+                                              long fromSeqNo,
+                                              int maxOperationCount,
+                                              String expectedHistoryUUID,
                                               long maxOperationSizeInBytes) throws IOException {
         if (indexShard.state() != IndexShardState.STARTED) {
             throw new IndexShardNotStartedException(indexShard.shardId(), indexShard.state());
         }
+        final String historyUUID = indexShard.getHistoryUUID();
+        if (historyUUID.equals(expectedHistoryUUID) == false) {
+            throw new IllegalStateException("unexpected history uuid, expected [" + expectedHistoryUUID + "], actual [" +
+                historyUUID + "]");
+        }
         if (fromSeqNo > globalCheckpoint) {
             return EMPTY_OPERATIONS_ARRAY;
         }