From 0ed62dbe1ccc184fade564c4157f6f1cd83b9e4c Mon Sep 17 00:00:00 2001 From: Nhat Nguyen Date: Mon, 4 Feb 2019 17:53:41 -0500 Subject: [PATCH] Tighten mapping syncing in ccr remote restore (#38071) There are two issues regarding the way that we sync mapping from leader to follower when a ccr restore is completed: 1. The returned mapping from a cluster service might not be up to date as the mapping of the restored index commit. 2. We should not compare the mapping version of the follower and the leader. They are not related to one another. Moreover, I think we should only ensure that once the restore is done, the mapping on the follower should be at least the mapping of the copied index commit. We don't have to sync the mapping which is updated after we have opened a session. Relates #36879 Closes #37887 --- .../xpack/ccr/action/CcrRequests.java | 38 ++++++++++++++++ .../ccr/action/ShardFollowTasksExecutor.java | 39 ++-------------- .../PutCcrRestoreSessionAction.java | 14 +++++- .../xpack/ccr/repository/CcrRepository.java | 45 ++++++++++--------- .../xpack/ccr/CcrRepositoryIT.java | 17 +------ 5 files changed, 81 insertions(+), 72 deletions(-) diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrRequests.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrRequests.java index 87d913c337642..f039810ed940c 100644 --- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrRequests.java +++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrRequests.java @@ -6,11 +6,15 @@ package org.elasticsearch.xpack.ccr.action; import org.elasticsearch.ElasticsearchStatusException; +import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest; import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest; import org.elasticsearch.action.admin.indices.mapping.put.MappingRequestValidator; +import org.elasticsearch.client.Client; import org.elasticsearch.cluster.metadata.IndexMetaData; import org.elasticsearch.cluster.metadata.MappingMetaData; +import org.elasticsearch.cluster.metadata.MetaData; +import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.index.Index; import org.elasticsearch.rest.RestStatus; @@ -18,6 +22,7 @@ import java.util.Arrays; import java.util.List; +import java.util.function.Supplier; import java.util.stream.Collectors; public final class CcrRequests { @@ -40,6 +45,39 @@ public static PutMappingRequest putMappingRequest(String followerIndex, MappingM return putMappingRequest; } + /** + * Gets an {@link IndexMetaData} of the given index. The mapping version and metadata version of the returned {@link IndexMetaData} + * must be at least the provided {@code mappingVersion} and {@code metadataVersion} respectively. + */ + public static void getIndexMetadata(Client client, Index index, long mappingVersion, long metadataVersion, + Supplier timeoutSupplier, ActionListener listener) { + final ClusterStateRequest request = CcrRequests.metaDataRequest(index.getName()); + if (metadataVersion > 0) { + request.waitForMetaDataVersion(metadataVersion).waitForTimeout(timeoutSupplier.get()); + } + client.admin().cluster().state(request, ActionListener.wrap( + response -> { + if (response.getState() == null) { + assert metadataVersion > 0 : metadataVersion; + throw new IllegalStateException("timeout to get cluster state with" + + " metadata version [" + metadataVersion + "], mapping version [" + mappingVersion + "]"); + } + final MetaData metaData = response.getState().metaData(); + final IndexMetaData indexMetaData = metaData.getIndexSafe(index); + if (indexMetaData.getMappingVersion() >= mappingVersion) { + listener.onResponse(indexMetaData); + return; + } + if (timeoutSupplier.get().nanos() < 0) { + throw new IllegalStateException("timeout to get cluster state with mapping version [" + mappingVersion + "]"); + } + // ask for the next version. + getIndexMetadata(client, index, mappingVersion, metaData.version() + 1, timeoutSupplier, listener); + }, + listener::onFailure + )); + } + public static final MappingRequestValidator CCR_PUT_MAPPING_REQUEST_VALIDATOR = (request, state, indices) -> { if (request.origin() == null) { return null; // a put-mapping-request on old versions does not have origin. diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java index 7e632a3f35782..abcabeb5a82be 100644 --- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java +++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java @@ -24,7 +24,6 @@ import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.cluster.metadata.IndexMetaData; import org.elasticsearch.cluster.metadata.MappingMetaData; -import org.elasticsearch.cluster.metadata.MetaData; import org.elasticsearch.cluster.routing.IndexRoutingTable; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.common.CheckedConsumer; @@ -60,6 +59,7 @@ import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.LongConsumer; +import java.util.function.Supplier; import static org.elasticsearch.xpack.ccr.CcrLicenseChecker.wrapClient; import static org.elasticsearch.xpack.ccr.action.TransportResumeFollowAction.extractLeaderShardHistoryUUIDs; @@ -121,7 +121,9 @@ protected AllocatedPersistentTask createTask(long id, String type, String action @Override protected void innerUpdateMapping(long minRequiredMappingVersion, LongConsumer handler, Consumer errorHandler) { final Index followerIndex = params.getFollowShardId().getIndex(); - getIndexMetadata(minRequiredMappingVersion, 0L, params, ActionListener.wrap( + final Index leaderIndex = params.getLeaderShardId().getIndex(); + final Supplier timeout = () -> isStopped() ? TimeValue.MINUS_ONE : waitForMetadataTimeOut; + CcrRequests.getIndexMetadata(remoteClient(params), leaderIndex, minRequiredMappingVersion, 0L, timeout, ActionListener.wrap( indexMetaData -> { if (indexMetaData.getMappings().isEmpty()) { assert indexMetaData.getMappingVersion() == 1; @@ -256,39 +258,6 @@ private Client remoteClient(ShardFollowTask params) { return wrapClient(client.getRemoteClusterClient(params.getRemoteCluster()), params.getHeaders()); } - private void getIndexMetadata(long minRequiredMappingVersion, long minRequiredMetadataVersion, - ShardFollowTask params, ActionListener listener) { - final Index leaderIndex = params.getLeaderShardId().getIndex(); - final ClusterStateRequest clusterStateRequest = CcrRequests.metaDataRequest(leaderIndex.getName()); - if (minRequiredMetadataVersion > 0) { - clusterStateRequest.waitForMetaDataVersion(minRequiredMetadataVersion).waitForTimeout(waitForMetadataTimeOut); - } - try { - remoteClient(params).admin().cluster().state(clusterStateRequest, ActionListener.wrap( - r -> { - // if wait_for_metadata_version timeout, the response is empty - if (r.getState() == null) { - assert minRequiredMetadataVersion > 0; - getIndexMetadata(minRequiredMappingVersion, minRequiredMetadataVersion, params, listener); - return; - } - final MetaData metaData = r.getState().metaData(); - final IndexMetaData indexMetaData = metaData.getIndexSafe(leaderIndex); - if (indexMetaData.getMappingVersion() < minRequiredMappingVersion) { - // ask for the next version. - getIndexMetadata(minRequiredMappingVersion, metaData.version() + 1, params, listener); - } else { - assert metaData.version() >= minRequiredMetadataVersion : metaData.version() + " < " + minRequiredMetadataVersion; - listener.onResponse(indexMetaData); - } - }, - listener::onFailure - )); - } catch (Exception e) { - listener.onFailure(e); - } - } - interface FollowerStatsInfoHandler { void accept(String followerHistoryUUID, long globalCheckpoint, long maxSeqNo); } diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/repositories/PutCcrRestoreSessionAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/repositories/PutCcrRestoreSessionAction.java index be116b523b3a0..fa38be0537262 100644 --- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/repositories/PutCcrRestoreSessionAction.java +++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/repositories/PutCcrRestoreSessionAction.java @@ -81,7 +81,8 @@ protected PutCcrRestoreSessionResponse shardOperation(PutCcrRestoreSessionReques throw new ShardNotFoundException(shardId); } Store.MetadataSnapshot storeFileMetaData = ccrRestoreService.openSession(request.getSessionUUID(), indexShard); - return new PutCcrRestoreSessionResponse(clusterService.localNode(), storeFileMetaData); + long mappingVersion = indexShard.indexSettings().getIndexMetaData().getMappingVersion(); + return new PutCcrRestoreSessionResponse(clusterService.localNode(), storeFileMetaData, mappingVersion); } @Override @@ -106,19 +107,22 @@ public static class PutCcrRestoreSessionResponse extends ActionResponse { private DiscoveryNode node; private Store.MetadataSnapshot storeFileMetaData; + private long mappingVersion; PutCcrRestoreSessionResponse() { } - PutCcrRestoreSessionResponse(DiscoveryNode node, Store.MetadataSnapshot storeFileMetaData) { + PutCcrRestoreSessionResponse(DiscoveryNode node, Store.MetadataSnapshot storeFileMetaData, long mappingVersion) { this.node = node; this.storeFileMetaData = storeFileMetaData; + this.mappingVersion = mappingVersion; } PutCcrRestoreSessionResponse(StreamInput in) throws IOException { super(in); node = new DiscoveryNode(in); storeFileMetaData = new Store.MetadataSnapshot(in); + mappingVersion = in.readVLong(); } @Override @@ -126,6 +130,7 @@ public void readFrom(StreamInput in) throws IOException { super.readFrom(in); node = new DiscoveryNode(in); storeFileMetaData = new Store.MetadataSnapshot(in); + mappingVersion = in.readVLong(); } @Override @@ -133,6 +138,7 @@ public void writeTo(StreamOutput out) throws IOException { super.writeTo(out); node.writeTo(out); storeFileMetaData.writeTo(out); + out.writeVLong(mappingVersion); } public DiscoveryNode getNode() { @@ -142,5 +148,9 @@ public DiscoveryNode getNode() { public Store.MetadataSnapshot getStoreFileMetaData() { return storeFileMetaData; } + + public long getMappingVersion() { + return mappingVersion; + } } } diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java index a1b601f352583..e8279b8394a51 100644 --- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java +++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java @@ -29,9 +29,9 @@ import org.elasticsearch.common.metrics.CounterMetric; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.ByteSizeValue; +import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.util.CombinedRateLimiter; import org.elasticsearch.index.Index; -import org.elasticsearch.index.IndexSettings; import org.elasticsearch.index.engine.EngineException; import org.elasticsearch.index.shard.IndexShard; import org.elasticsearch.index.shard.IndexShardRecoveryException; @@ -72,6 +72,8 @@ import java.util.Map; import java.util.Set; import java.util.function.LongConsumer; +import java.util.function.Supplier; + /** * This repository relies on a remote cluster for Ccr restores. It is read-only so it can only be used to @@ -288,11 +290,10 @@ public void restoreShard(IndexShard indexShard, SnapshotId snapshotId, Version v String name = metadata.name(); try (RestoreSession restoreSession = openSession(name, remoteClient, leaderShardId, indexShard, recoveryState)) { restoreSession.restoreFiles(); + updateMappings(remoteClient, leaderIndex, restoreSession.mappingVersion, client, indexShard.routingEntry().index()); } catch (Exception e) { throw new IndexShardRestoreFailedException(indexShard.shardId(), "failed to restore snapshot [" + snapshotId + "]", e); } - - maybeUpdateMappings(client, remoteClient, leaderIndex, indexShard.indexSettings()); } @Override @@ -300,20 +301,22 @@ public IndexShardSnapshotStatus getShardSnapshotStatus(SnapshotId snapshotId, Ve throw new UnsupportedOperationException("Unsupported for repository of type: " + TYPE); } - private void maybeUpdateMappings(Client localClient, Client remoteClient, Index leaderIndex, IndexSettings followerIndexSettings) { - ClusterStateRequest clusterStateRequest = CcrRequests.metaDataRequest(leaderIndex.getName()); - ClusterStateResponse clusterState = remoteClient.admin().cluster().state(clusterStateRequest) - .actionGet(ccrSettings.getRecoveryActionTimeout()); - IndexMetaData leaderIndexMetadata = clusterState.getState().metaData().getIndexSafe(leaderIndex); - long leaderMappingVersion = leaderIndexMetadata.getMappingVersion(); - - if (leaderMappingVersion > followerIndexSettings.getIndexMetaData().getMappingVersion()) { - Index followerIndex = followerIndexSettings.getIndex(); - assert leaderIndexMetadata.getMappings().size() == 1 : "expected exactly one mapping, but got [" + - leaderIndexMetadata.getMappings().size() + "]"; - MappingMetaData mappingMetaData = leaderIndexMetadata.getMappings().iterator().next().value; - PutMappingRequest putMappingRequest = CcrRequests.putMappingRequest(followerIndex.getName(), mappingMetaData); - localClient.admin().indices().putMapping(putMappingRequest).actionGet(ccrSettings.getRecoveryActionTimeout()); + private void updateMappings(Client leaderClient, Index leaderIndex, long leaderMappingVersion, + Client followerClient, Index followerIndex) { + final PlainActionFuture indexMetadataFuture = new PlainActionFuture<>(); + final long startTimeInNanos = System.nanoTime(); + final Supplier timeout = () -> { + final long elapsedInNanos = System.nanoTime() - startTimeInNanos; + return TimeValue.timeValueNanos(ccrSettings.getRecoveryActionTimeout().nanos() - elapsedInNanos); + }; + CcrRequests.getIndexMetadata(leaderClient, leaderIndex, leaderMappingVersion, 0L, timeout, indexMetadataFuture); + final IndexMetaData leaderIndexMetadata = indexMetadataFuture.actionGet(ccrSettings.getRecoveryActionTimeout()); + assert leaderIndexMetadata.getMappings().size() == 1 : "expected exactly one mapping, but got [" + + leaderIndexMetadata.getMappings().size() + "]"; + MappingMetaData mappingMetaData = leaderIndexMetadata.getMappings().iterator().next().value; + if (mappingMetaData != null) { + final PutMappingRequest putMappingRequest = CcrRequests.putMappingRequest(followerIndex.getName(), mappingMetaData); + followerClient.admin().indices().putMapping(putMappingRequest).actionGet(ccrSettings.getRecoveryActionTimeout()); } } @@ -323,7 +326,7 @@ private RestoreSession openSession(String repositoryName, Client remoteClient, S PutCcrRestoreSessionAction.PutCcrRestoreSessionResponse response = remoteClient.execute(PutCcrRestoreSessionAction.INSTANCE, new PutCcrRestoreSessionRequest(sessionUUID, leaderShardId)).actionGet(ccrSettings.getRecoveryActionTimeout()); return new RestoreSession(repositoryName, remoteClient, sessionUUID, response.getNode(), indexShard, recoveryState, - response.getStoreFileMetaData(), ccrSettings, throttledTime::inc); + response.getStoreFileMetaData(), response.getMappingVersion(), ccrSettings, throttledTime::inc); } private static class RestoreSession extends FileRestoreContext implements Closeable { @@ -334,17 +337,19 @@ private static class RestoreSession extends FileRestoreContext implements Closea private final String sessionUUID; private final DiscoveryNode node; private final Store.MetadataSnapshot sourceMetaData; + private final long mappingVersion; private final CcrSettings ccrSettings; private final LongConsumer throttleListener; RestoreSession(String repositoryName, Client remoteClient, String sessionUUID, DiscoveryNode node, IndexShard indexShard, - RecoveryState recoveryState, Store.MetadataSnapshot sourceMetaData, CcrSettings ccrSettings, - LongConsumer throttleListener) { + RecoveryState recoveryState, Store.MetadataSnapshot sourceMetaData, long mappingVersion, + CcrSettings ccrSettings, LongConsumer throttleListener) { super(repositoryName, indexShard, SNAPSHOT_ID, recoveryState, BUFFER_SIZE); this.remoteClient = remoteClient; this.sessionUUID = sessionUUID; this.node = node; this.sourceMetaData = sourceMetaData; + this.mappingVersion = mappingVersion; this.ccrSettings = ccrSettings; this.throttleListener = throttleListener; } diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRepositoryIT.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRepositoryIT.java index 170930ef7d67a..b3e4d23115fc8 100644 --- a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRepositoryIT.java +++ b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRepositoryIT.java @@ -390,7 +390,6 @@ public void testIndividualActionsTimeout() throws Exception { assertAcked(followerClient().admin().cluster().updateSettings(settingsRequest).actionGet()); } - @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/37887") public void testFollowerMappingIsUpdated() throws IOException { String leaderClusterRepoName = CcrRepository.NAME_PREFIX + "leader_cluster"; String leaderIndex = "index1"; @@ -413,16 +412,8 @@ public void testFollowerMappingIsUpdated() throws IOException { .renameReplacement(followerIndex).masterNodeTimeout(new TimeValue(1L, TimeUnit.HOURS)) .indexSettings(settingsBuilder); - // TODO: Eventually when the file recovery work is complete, we should test updated mappings by - // indexing to the leader while the recovery is happening. However, into order to that test mappings - // are updated prior to that work, we index documents in the clear session callback. This will - // ensure a mapping change prior to the final mapping check on the follower side. - for (CcrRestoreSourceService restoreSourceService : getLeaderCluster().getDataNodeInstances(CcrRestoreSourceService.class)) { - restoreSourceService.addCloseSessionListener(s -> { - final String source = String.format(Locale.ROOT, "{\"k\":%d}", 1); - leaderClient().prepareIndex("index1", "doc", Long.toString(1)).setSource(source, XContentType.JSON).get(); - }); - } + final String source = String.format(Locale.ROOT, "{\"k\":%d}", 1); + leaderClient().prepareIndex("index1", "doc", Long.toString(1)).setSource(source, XContentType.JSON).get(); PlainActionFuture future = PlainActionFuture.newFuture(); restoreService.restoreSnapshot(restoreRequest, waitForRestore(clusterService, future)); @@ -435,10 +426,6 @@ public void testFollowerMappingIsUpdated() throws IOException { clusterStateRequest.clear(); clusterStateRequest.metaData(true); clusterStateRequest.indices(followerIndex); - ClusterStateResponse clusterState = followerClient().admin().cluster().state(clusterStateRequest).actionGet(); - IndexMetaData followerIndexMetadata = clusterState.getState().metaData().index(followerIndex); - assertEquals(2, followerIndexMetadata.getMappingVersion()); - MappingMetaData mappingMetaData = followerClient().admin().indices().prepareGetMappings("index2").get().getMappings() .get("index2").get("doc"); assertThat(XContentMapValues.extractValue("properties.k.type", mappingMetaData.sourceAsMap()), equalTo("long"));