Skip to content

Tighten mapping syncing in ccr remote restore #38071

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Feb 4, 2019
Merged

Tighten mapping syncing in ccr remote restore #38071

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4a570fa
Tighten mapping syncing in ccr remote restore
dnhatn Jan 30, 2019
7715d00
Merge branch 'master' into remote-recovery-mapping
dnhatn Feb 4, 2019
90709b6
wording
dnhatn Feb 4, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrRequests.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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());
Copy link
Contributor

@ywelsch ywelsch Jan 31, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why remove this?

Copy link
Member Author

@dnhatn dnhatn Feb 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Previously we changed mapping twice in the leader: one before opening session and one after. These two events are propagated separately; then the mapping version on the follower is 2. Now, these two events happen before we open session, then the mapping version on the follower is 1 instead of 2. I think verifying the content of the mapping is good enough.

Copy link
Contributor

@ywelsch ywelsch Feb 4, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok


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