Skip to content

Commit

Permalink
Tighten mapping syncing in ccr remote restore (#38071)
Browse files Browse the repository at this point in the history 
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
  • Loading branch information
@dnhatn
dnhatn authored Feb 4, 2019
1 parent 5a33816 commit cecfa5b
Show file tree
Hide file tree
Showing 5 changed files with 79 additions and 70 deletions.
38 changes: 38 additions & 0 deletions x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrRequests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,18 +6,23 @@
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;
import org.elasticsearch.xpack.ccr.CcrSettings;

import java.util.Arrays;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;

public final class CcrRequests {
Expand All @@ -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<TimeValue> timeoutSupplier, ActionListener<IndexMetaData> 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.
Expand Down
39 changes: 4 additions & 35 deletions ...plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -59,6 +58,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;
Expand Down Expand Up @@ -111,7 +111,9 @@ protected AllocatedPersistentTask createTask(long id, String type, String action
@Override
protected void innerUpdateMapping(long minRequiredMappingVersion, LongConsumer handler, Consumer<Exception> errorHandler) {
final Index followerIndex = params.getFollowShardId().getIndex();
getIndexMetadata(minRequiredMappingVersion, 0L, params, ActionListener.wrap(
final Index leaderIndex = params.getLeaderShardId().getIndex();
final Supplier<TimeValue> 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;
Expand Down Expand Up @@ -246,39 +248,6 @@ private Client remoteClient(ShardFollowTask params) {
return wrapClient(client.getRemoteClusterClient(params.getRemoteCluster()), params.getHeaders());
}

private void getIndexMetadata(long minRequiredMappingVersion, long minRequiredMetadataVersion,
ShardFollowTask params, ActionListener<IndexMetaData> 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);
}
Expand Down
14 changes: 12 additions & 2 deletions ...main/java/org/elasticsearch/xpack/ccr/action/repositories/PutCcrRestoreSessionAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,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
Expand All @@ -97,33 +98,38 @@ 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
public void readFrom(StreamInput in) throws IOException {
super.readFrom(in);
node = new DiscoveryNode(in);
storeFileMetaData = new Store.MetadataSnapshot(in);
mappingVersion = in.readVLong();
}

@Override
public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
node.writeTo(out);
storeFileMetaData.writeTo(out);
out.writeVLong(mappingVersion);
}

public DiscoveryNode getNode() {
Expand All @@ -133,5 +139,9 @@ public DiscoveryNode getNode() {
public Store.MetadataSnapshot getStoreFileMetaData() {
return storeFileMetaData;
}

public long getMappingVersion() {
return mappingVersion;
}
}
}
41 changes: 23 additions & 18 deletions x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -288,30 +290,31 @@ 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
public IndexShardSnapshotStatus getShardSnapshotStatus(SnapshotId snapshotId, Version version, IndexId indexId, ShardId leaderShardId) {
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();
MappingMetaData mappingMetaData = leaderIndexMetadata.mapping();
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<IndexMetaData> indexMetadataFuture = new PlainActionFuture<>();
final long startTimeInNanos = System.nanoTime();
final Supplier<TimeValue> 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());
final MappingMetaData mappingMetaData = leaderIndexMetadata.mapping();
if (mappingMetaData != null) {
final PutMappingRequest putMappingRequest = CcrRequests.putMappingRequest(followerIndex.getName(), mappingMetaData);
followerClient.admin().indices().putMapping(putMappingRequest).actionGet(ccrSettings.getRecoveryActionTimeout());
}
}

Expand All @@ -321,7 +324,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 {
Expand All @@ -332,17 +335,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;
}
Expand Down
17 changes: 2 additions & 15 deletions x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/CcrRepositoryIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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<RestoreInfo> future = PlainActionFuture.newFuture();
restoreService.restoreSnapshot(restoreRequest, waitForRestore(clusterService, future));
Expand All @@ -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"));
Expand Down

0 comments on commit cecfa5b

Please sign in to comment.