[CCR] Added history uuid validation

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

@@ -110,7 +110,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

@@ -82,6 +82,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 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,41 @@ 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) {
+                    CommitStats commitStats = shardStats.getCommitStats();
+                    String historyUUID = commitStats.getUserData().get(Engine.HISTORY_UUID_KEY);
+                    ShardId shardId = shardStats.getShardRouting().shardId();
+                    historyUUIDs[shardId.id()] = historyUUID;
+                }
+            }
+            historyUUIDConsumer.accept(historyUUIDs);
+        };
+        IndicesStatsRequest request = new IndicesStatsRequest();
+        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/CreateAndFollowIndexAction.java

@@ -43,13 +43,16 @@
 import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.ccr.Ccr;
 import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
 import org.elasticsearch.xpack.ccr.CcrSettings;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.function.Consumer;
 
 public class CreateAndFollowIndexAction extends Action<CreateAndFollowIndexAction.Response> {
 
@@ -254,25 +257,32 @@ protected void masterOperation(
         private void createFollowerIndexAndFollowLocalIndex(
                 final Request request, final ClusterState state, final ActionListener<Response> listener) {
             // following an index in local cluster, so use local cluster state to fetch leader index metadata
-            final IndexMetaData leaderIndexMetadata = state.getMetaData().index(request.getFollowRequest().getLeaderIndex());
-            createFollowerIndex(leaderIndexMetadata, request, listener);
+            final String leaderIndex = request.getFollowRequest().getLeaderIndex();
+            final IndexMetaData leaderIndexMetadata = state.getMetaData().index(leaderIndex);
+            Consumer<String[]> handler = historyUUIDs -> {
+                createFollowerIndex(leaderIndexMetadata, historyUUIDs, request, listener);
+            };
+            ccrLicenseChecker.fetchLeaderHistoryUUIDs(client, leaderIndexMetadata, listener::onFailure, handler);
         }
 
         private void createFollowerIndexAndFollowRemoteIndex(
                 final Request request,
                 final String clusterAlias,
                 final String leaderIndex,
                 final ActionListener<Response> listener) {
-            ccrLicenseChecker.checkRemoteClusterLicenseAndFetchLeaderIndexMetadata(
+            ccrLicenseChecker.checkRemoteClusterLicenseAndFetchLeaderIndexMetadataAndHistoryUUIDs(
                     client,
                     clusterAlias,
                     leaderIndex,
                     listener::onFailure,
-                    leaderIndexMetaData -> createFollowerIndex(leaderIndexMetaData, request, listener));
+                    (historyUUID, leaderIndexMetaData) -> createFollowerIndex(leaderIndexMetaData, historyUUID, request, listener));
         }
 
         private void createFollowerIndex(
-                final IndexMetaData leaderIndexMetaData, final Request request, final ActionListener<Response> listener) {
+                final IndexMetaData leaderIndexMetaData,
+                final String[] historyUUIDs,
+                final Request request,
+                final ActionListener<Response> listener) {
             if (leaderIndexMetaData == null) {
                 listener.onFailure(new IllegalArgumentException("leader index [" + request.getFollowRequest().getLeaderIndex() +
                     "] does not exist"));
@@ -308,6 +318,11 @@ public ClusterState execute(ClusterState currentState) throws Exception {
                     MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
                     IndexMetaData.Builder imdBuilder = IndexMetaData.builder(followIndex);
 
+                    // Adding the leader index uuid for each shard as custom metadata:
+                    Map<String, String> metadata = new HashMap<>();
+                    metadata.put(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS, String.join(",", historyUUIDs));
+                    imdBuilder.putCustom(Ccr.CCR_CUSTOM_METADATA_KEY, metadata);
+
                     // Copy all settings, but overwrite a few settings.
                     Settings.Builder settingsBuilder = Settings.builder();
                     settingsBuilder.put(leaderIndexMetaData.getSettings());

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

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexingSlowLog;
 import org.elasticsearch.index.SearchSlowLog;
@@ -47,6 +48,7 @@
 import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.ccr.Ccr;
 import org.elasticsearch.xpack.ccr.CcrLicenseChecker;
 import org.elasticsearch.xpack.ccr.CcrSettings;
 
@@ -352,11 +354,17 @@ private void followLocalIndex(final Request request,
             final IndexMetaData followerIndexMetadata = state.getMetaData().index(request.getFollowerIndex());
             // following an index in local cluster, so use local cluster state to fetch leader index metadata
             final IndexMetaData leaderIndexMetadata = state.getMetaData().index(request.getLeaderIndex());
-            try {
-                start(request, null, leaderIndexMetadata, followerIndexMetadata, listener);
-            } catch (final IOException e) {
-                listener.onFailure(e);
+            if (leaderIndexMetadata == null) {
+                throw new IndexNotFoundException(request.getFollowerIndex());
             }
+
+            ccrLicenseChecker.fetchLeaderHistoryUUIDs(client, leaderIndexMetadata, listener::onFailure, historyUUIDs -> {
+                try {
+                    start(request, null, leaderIndexMetadata, followerIndexMetadata, historyUUIDs, listener);
+                } catch (final IOException e) {
+                    listener.onFailure(e);
+                }
+            });
         }
 
         private void followRemoteIndex(
@@ -366,14 +374,14 @@ private void followRemoteIndex(
                 final ActionListener<AcknowledgedResponse> listener) {
             final ClusterState state = clusterService.state();
             final IndexMetaData followerIndexMetadata = state.getMetaData().index(request.getFollowerIndex());
-            ccrLicenseChecker.checkRemoteClusterLicenseAndFetchLeaderIndexMetadata(
+            ccrLicenseChecker.checkRemoteClusterLicenseAndFetchLeaderIndexMetadataAndHistoryUUIDs(
                     client,
                     clusterAlias,
                     leaderIndex,
                     listener::onFailure,
-                    leaderIndexMetadata -> {
+                    (leaderHistoryUUID, leaderIndexMetadata) -> {
                         try {
-                            start(request, clusterAlias, leaderIndexMetadata, followerIndexMetadata, listener);
+                            start(request, clusterAlias, leaderIndexMetadata, followerIndexMetadata, leaderHistoryUUID, listener);
                         } catch (final IOException e) {
                             listener.onFailure(e);
                         }
@@ -395,25 +403,37 @@ void start(
             String clusterNameAlias,
             IndexMetaData leaderIndexMetadata,
             IndexMetaData followIndexMetadata,
+            String[] leaderIndexHistoryUUIDs,
             ActionListener<AcknowledgedResponse> handler) throws IOException {
 
             MapperService mapperService = followIndexMetadata != null ? indicesService.createIndexMapperService(followIndexMetadata) : null;
-            validate(request, leaderIndexMetadata, followIndexMetadata, mapperService);
+            validate(request, leaderIndexMetadata, followIndexMetadata, leaderIndexHistoryUUIDs, mapperService);
             final int numShards = followIndexMetadata.getNumberOfShards();
             final AtomicInteger counter = new AtomicInteger(numShards);
             final AtomicReferenceArray<Object> responses = new AtomicReferenceArray<>(followIndexMetadata.getNumberOfShards());
             Map<String, String> filteredHeaders = threadPool.getThreadContext().getHeaders().entrySet().stream()
                 .filter(e -> ShardFollowTask.HEADER_FILTERS.contains(e.getKey()))
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));for (int i = 0; i < numShards; i++) {
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+            for (int i = 0; i < numShards; i++) {
                 final int shardId = i;
                 String taskId = followIndexMetadata.getIndexUUID() + "-" + shardId;
+                String[] recordedLeaderShardHistoryUUIDs = extractIndexShardHistoryUUIDs(followIndexMetadata);
+                String recordedLeaderShardHistoryUUID = recordedLeaderShardHistoryUUIDs[shardId];
 
                 ShardFollowTask shardFollowTask = new ShardFollowTask(clusterNameAlias,
-                        new ShardId(followIndexMetadata.getIndex(), shardId),
-                        new ShardId(leaderIndexMetadata.getIndex(), shardId),
-                        request.maxBatchOperationCount, request.maxConcurrentReadBatches, request.maxOperationSizeInBytes,
-                        request.maxConcurrentWriteBatches, request.maxWriteBufferSize, request.retryTimeout,
-                        request.idleShardRetryDelay, filteredHeaders);
+                    new ShardId(followIndexMetadata.getIndex(), shardId),
+                    new ShardId(leaderIndexMetadata.getIndex(), shardId),
+                    request.maxBatchOperationCount,
+                    request.maxConcurrentReadBatches,
+                    request.maxOperationSizeInBytes,
+                    request.maxConcurrentWriteBatches,
+                    request.maxWriteBufferSize,
+                    request.retryTimeout,
+                    request.idleShardRetryDelay,
+                    recordedLeaderShardHistoryUUID,
+                    filteredHeaders
+                );
                 persistentTasksService.sendStartRequest(taskId, ShardFollowTask.NAME, shardFollowTask,
                         new ActionListener<PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask>>() {
                             @Override
@@ -510,13 +530,28 @@ void finalizeResponse() {
 
     static void validate(Request request,
                          IndexMetaData leaderIndex,
-                         IndexMetaData followIndex, MapperService followerMapperService) {
+                         IndexMetaData followIndex,
+                         String[] leaderIndexHistoryUUID,
+                         MapperService followerMapperService) {
         if (leaderIndex == null) {
             throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not exist");
         }
         if (followIndex == null) {
             throw new IllegalArgumentException("follow index [" + request.followerIndex + "] does not exist");
         }
+
+        String[] recordedHistoryUUIDs = extractIndexShardHistoryUUIDs(followIndex);
+        assert recordedHistoryUUIDs.length == leaderIndexHistoryUUID.length;
+        for (int i = 0; i < leaderIndexHistoryUUID.length; i++) {
+            String recordedLeaderIndexHistoryUUID = recordedHistoryUUIDs[i];
+            String actualLeaderIndexHistoryUUID = leaderIndexHistoryUUID[i];
+            if (recordedLeaderIndexHistoryUUID.equals(actualLeaderIndexHistoryUUID) == false) {
+                throw new IllegalArgumentException("follow index [" + request.followerIndex + "] should reference [" +
+                    recordedLeaderIndexHistoryUUID + "] as history uuid but instead reference [" +
+                    actualLeaderIndexHistoryUUID + "] as history uuid");
+            }
+        }
+
         if (leaderIndex.getSettings().getAsBoolean(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), false) == false) {
             throw new IllegalArgumentException("leader index [" + request.leaderIndex + "] does not have soft deletes enabled");
         }
@@ -568,4 +603,10 @@ private static Settings filter(Settings originalSettings) {
         return settings.build();
     }
 
+    private static String[] extractIndexShardHistoryUUIDs(IndexMetaData followIndexMetadata) {
+        String historyUUIDs = followIndexMetadata.getCustomData(Ccr.CCR_CUSTOM_METADATA_KEY)
+            .get(Ccr.CCR_CUSTOM_METADATA_LEADER_INDEX_SHARD_HISTORY_UUIDS);
+        return historyUUIDs.split(",");
+    }
+
 }