Skip to content

Commit 506449b

Browse files
committed
Delete shard store files before restoring a snapshot (#27476)
Pull request #20220 added a change where the store files that have the same name but are different from the ones in the snapshot are deleted first before the snapshot is restored. This logic was based on the `Store.RecoveryDiff.different` set of files which works by computing a diff between an existing store and a snapshot. This works well when the files on the filesystem form valid shard store, ie there's a `segments` file and store files are not corrupted. Otherwise, the existing store's snapshot metadata cannot be read (using Store#snapshotStoreMetadata()) and an exception is thrown (CorruptIndexException, IndexFormatTooOldException etc) which is later caught as the begining of the restore process (see RestoreContext#restore()) and is translated into an empty store metadata (Store.MetadataSnapshot.EMPTY). This will make the deletion of different files introduced in #20220 useless as the set of files will always be empty even when store files exist on the filesystem. And if some files are present within the store directory, then restoring a snapshot with files with same names will fail with a FileAlreadyExistException. This is part of the #26865 issue. There are various cases were some files could exist in the store directory before a snapshot is restored. One that Igor identified is a restore attempt that failed on a node and only first files were restored, then the shard is allocated again to the same node and the restore starts again (but fails because of existing files). Another one is when some files of a closed index are corrupted / deleted and the index is restored. This commit adds a test that uses the infrastructure provided by IndexShardTestCase in order to test that restoring a shard succeed even when files with same names exist on filesystem. Related to #26865
1 parent 3bc3d4d commit 506449b

File tree

5 files changed

+209
-18
lines changed

5 files changed

+209
-18
lines changed

core/src/main/java/org/elasticsearch/index/store/Store.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -731,7 +731,7 @@ public String toString() {
731731

732732
/**
733733
* Represents a snapshot of the current directory build from the latest Lucene commit.
734-
* Only files that are part of the last commit are considered in this datastrucutre.
734+
* Only files that are part of the last commit are considered in this datastructure.
735735
* For backwards compatibility the snapshot might include legacy checksums that
736736
* are derived from a dedicated checksum file written by older elasticsearch version pre 1.3
737737
* <p>

core/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

+27-16
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@
3535
import org.apache.lucene.store.RateLimiter;
3636
import org.apache.lucene.util.BytesRef;
3737
import org.apache.lucene.util.BytesRefBuilder;
38-
import org.apache.lucene.util.IOUtils;
3938
import org.elasticsearch.ElasticsearchParseException;
4039
import org.elasticsearch.ExceptionsHelper;
4140
import org.elasticsearch.ResourceNotFoundException;
@@ -110,6 +109,7 @@
110109
import java.nio.file.FileAlreadyExistsException;
111110
import java.nio.file.NoSuchFileException;
112111
import java.util.ArrayList;
112+
import java.util.Arrays;
113113
import java.util.Collection;
114114
import java.util.Collections;
115115
import java.util.HashMap;
@@ -1451,6 +1451,9 @@ public void restore() throws IOException {
14511451
SnapshotFiles snapshotFiles = new SnapshotFiles(snapshot.snapshot(), snapshot.indexFiles());
14521452
Store.MetadataSnapshot recoveryTargetMetadata;
14531453
try {
1454+
// this will throw an IOException if the store has no segments infos file. The
1455+
// store can still have existing files but they will be deleted just before being
1456+
// restored.
14541457
recoveryTargetMetadata = targetShard.snapshotStoreMetadata();
14551458
} catch (IndexNotFoundException e) {
14561459
// happens when restore to an empty shard, not a big deal
@@ -1478,7 +1481,14 @@ public void restore() throws IOException {
14781481
snapshotMetaData.put(fileInfo.metadata().name(), fileInfo.metadata());
14791482
fileInfos.put(fileInfo.metadata().name(), fileInfo);
14801483
}
1484+
14811485
final Store.MetadataSnapshot sourceMetaData = new Store.MetadataSnapshot(unmodifiableMap(snapshotMetaData), emptyMap(), 0);
1486+
1487+
final StoreFileMetaData restoredSegmentsFile = sourceMetaData.getSegmentsFile();
1488+
if (restoredSegmentsFile == null) {
1489+
throw new IndexShardRestoreFailedException(shardId, "Snapshot has no segments file");
1490+
}
1491+
14821492
final Store.RecoveryDiff diff = sourceMetaData.recoveryDiff(recoveryTargetMetadata);
14831493
for (StoreFileMetaData md : diff.identical) {
14841494
BlobStoreIndexShardSnapshot.FileInfo fileInfo = fileInfos.get(md.name());
@@ -1505,29 +1515,31 @@ public void restore() throws IOException {
15051515
logger.trace("no files to recover, all exists within the local store");
15061516
}
15071517

1508-
if (logger.isTraceEnabled()) {
1509-
logger.trace("[{}] [{}] recovering_files [{}] with total_size [{}], reusing_files [{}] with reused_size [{}]", shardId, snapshotId,
1510-
index.totalRecoverFiles(), new ByteSizeValue(index.totalRecoverBytes()), index.reusedFileCount(), new ByteSizeValue(index.reusedFileCount()));
1511-
}
15121518
try {
1513-
// first, delete pre-existing files in the store that have the same name but are
1514-
// different (i.e. different length/checksum) from those being restored in the snapshot
1515-
for (final StoreFileMetaData storeFileMetaData : diff.different) {
1516-
IOUtils.deleteFiles(store.directory(), storeFileMetaData.name());
1517-
}
1519+
// list of all existing store files
1520+
final List<String> deleteIfExistFiles = Arrays.asList(store.directory().listAll());
1521+
15181522
// restore the files from the snapshot to the Lucene store
15191523
for (final BlobStoreIndexShardSnapshot.FileInfo fileToRecover : filesToRecover) {
1524+
// if a file with a same physical name already exist in the store we need to delete it
1525+
// before restoring it from the snapshot. We could be lenient and try to reuse the existing
1526+
// store files (and compare their names/length/checksum again with the snapshot files) but to
1527+
// avoid extra complexity we simply delete them and restore them again like StoreRecovery
1528+
// does with dangling indices. Any existing store file that is not restored from the snapshot
1529+
// will be clean up by RecoveryTarget.cleanFiles().
1530+
final String physicalName = fileToRecover.physicalName();
1531+
if (deleteIfExistFiles.contains(physicalName)) {
1532+
logger.trace("[{}] [{}] deleting pre-existing file [{}]", shardId, snapshotId, physicalName);
1533+
store.directory().deleteFile(physicalName);
1534+
}
1535+
15201536
logger.trace("[{}] [{}] restoring file [{}]", shardId, snapshotId, fileToRecover.name());
15211537
restoreFile(fileToRecover, store);
15221538
}
15231539
} catch (IOException ex) {
15241540
throw new IndexShardRestoreFailedException(shardId, "Failed to recover index", ex);
15251541
}
1526-
final StoreFileMetaData restoredSegmentsFile = sourceMetaData.getSegmentsFile();
1527-
if (recoveryTargetMetadata == null) {
1528-
throw new IndexShardRestoreFailedException(shardId, "Snapshot has no segments file");
1529-
}
1530-
assert restoredSegmentsFile != null;
1542+
15311543
// read the snapshot data persisted
15321544
final SegmentInfos segmentCommitInfos;
15331545
try {
@@ -1602,5 +1614,4 @@ private void restoreFile(final BlobStoreIndexShardSnapshot.FileInfo fileInfo, fi
16021614
}
16031615
}
16041616
}
1605-
16061617
}

core/src/main/java/org/elasticsearch/snapshots/RestoreService.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -188,7 +188,7 @@ public void restoreSnapshot(final RestoreRequest request, final ActionListener<R
188188
final SnapshotInfo snapshotInfo = repository.getSnapshotInfo(snapshotId);
189189
final Snapshot snapshot = new Snapshot(request.repositoryName, snapshotId);
190190
List<String> filteredIndices = SnapshotUtils.filterIndices(snapshotInfo.indices(), request.indices(), request.indicesOptions());
191-
MetaData metaData = repository.getSnapshotMetaData(snapshotInfo, repositoryData.resolveIndices(filteredIndices));
191+
final MetaData metaData = repository.getSnapshotMetaData(snapshotInfo, repositoryData.resolveIndices(filteredIndices));
192192

193193
// Make sure that we can restore from this snapshot
194194
validateSnapshotRestorable(request.repositoryName, snapshotInfo);
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.elasticsearch.repositories.blobstore;
21+
22+
import org.apache.lucene.store.Directory;
23+
import org.apache.lucene.util.IOUtils;
24+
import org.apache.lucene.util.TestUtil;
25+
import org.elasticsearch.cluster.metadata.RepositoryMetaData;
26+
import org.elasticsearch.cluster.routing.ShardRouting;
27+
import org.elasticsearch.cluster.routing.ShardRoutingHelper;
28+
import org.elasticsearch.common.UUIDs;
29+
import org.elasticsearch.common.settings.Settings;
30+
import org.elasticsearch.env.Environment;
31+
import org.elasticsearch.index.shard.IndexShard;
32+
import org.elasticsearch.index.shard.IndexShardState;
33+
import org.elasticsearch.index.shard.IndexShardTestCase;
34+
import org.elasticsearch.index.shard.ShardId;
35+
import org.elasticsearch.index.store.Store;
36+
import org.elasticsearch.index.store.StoreFileMetaData;
37+
import org.elasticsearch.repositories.IndexId;
38+
import org.elasticsearch.repositories.Repository;
39+
import org.elasticsearch.repositories.fs.FsRepository;
40+
import org.elasticsearch.snapshots.Snapshot;
41+
import org.elasticsearch.snapshots.SnapshotId;
42+
43+
import java.io.IOException;
44+
import java.nio.file.Files;
45+
import java.nio.file.Path;
46+
import java.util.Arrays;
47+
import java.util.List;
48+
49+
import static org.elasticsearch.cluster.routing.RecoverySource.StoreRecoverySource.EXISTING_STORE_INSTANCE;
50+
51+
/**
52+
* This class tests the behavior of {@link BlobStoreRepository} when it
53+
* restores a shard from a snapshot but some files with same names already
54+
* exist on disc.
55+
*/
56+
public class BlobStoreRepositoryRestoreTests extends IndexShardTestCase {
57+
58+
/**
59+
* Restoring a snapshot that contains multiple files must succeed even when
60+
* some files already exist in the shard's store.
61+
*/
62+
public void testRestoreSnapshotWithExistingFiles() throws IOException {
63+
final IndexId indexId = new IndexId(randomAlphaOfLength(10), UUIDs.randomBase64UUID());
64+
final ShardId shardId = new ShardId(indexId.getName(), indexId.getId(), 0);
65+
66+
IndexShard shard = newShard(shardId, true);
67+
try {
68+
// index documents in the shards
69+
final int numDocs = scaledRandomIntBetween(1, 500);
70+
recoverShardFromStore(shard);
71+
for (int i = 0; i < numDocs; i++) {
72+
indexDoc(shard, "doc", Integer.toString(i));
73+
if (rarely()) {
74+
flushShard(shard, false);
75+
}
76+
}
77+
assertDocCount(shard, numDocs);
78+
79+
// snapshot the shard
80+
final Repository repository = createRepository();
81+
final Snapshot snapshot = new Snapshot(repository.getMetadata().name(), new SnapshotId(randomAlphaOfLength(10), "_uuid"));
82+
snapshotShard(shard, snapshot, repository);
83+
84+
// capture current store files
85+
final Store.MetadataSnapshot storeFiles = shard.snapshotStoreMetadata();
86+
assertFalse(storeFiles.asMap().isEmpty());
87+
88+
// close the shard
89+
closeShards(shard);
90+
91+
// delete some random files in the store
92+
List<String> deletedFiles = randomSubsetOf(randomIntBetween(1, storeFiles.size() - 1), storeFiles.asMap().keySet());
93+
for (String deletedFile : deletedFiles) {
94+
Files.delete(shard.shardPath().resolveIndex().resolve(deletedFile));
95+
}
96+
97+
// build a new shard using the same store directory as the closed shard
98+
ShardRouting shardRouting = ShardRoutingHelper.initWithSameId(shard.routingEntry(), EXISTING_STORE_INSTANCE);
99+
shard = newShard(shardRouting, shard.shardPath(), shard.indexSettings().getIndexMetaData(), null, null, () -> {});
100+
101+
// restore the shard
102+
recoverShardFromSnapshot(shard, snapshot, repository);
103+
104+
// check that the shard is not corrupted
105+
TestUtil.checkIndex(shard.store().directory());
106+
107+
// check that all files have been restored
108+
final Directory directory = shard.store().directory();
109+
final List<String> directoryFiles = Arrays.asList(directory.listAll());
110+
111+
for (StoreFileMetaData storeFile : storeFiles) {
112+
String fileName = storeFile.name();
113+
assertTrue("File [" + fileName + "] does not exist in store directory", directoryFiles.contains(fileName));
114+
assertEquals(storeFile.length(), shard.store().directory().fileLength(fileName));
115+
}
116+
} finally {
117+
if (shard != null && shard.state() != IndexShardState.CLOSED) {
118+
try {
119+
shard.close("test", false);
120+
} finally {
121+
IOUtils.close(shard.store());
122+
}
123+
}
124+
}
125+
}
126+
127+
/** Create a {@link Repository} with a random name **/
128+
private Repository createRepository() throws IOException {
129+
Settings settings = Settings.builder().put("location", randomAlphaOfLength(10)).build();
130+
RepositoryMetaData repositoryMetaData = new RepositoryMetaData(randomAlphaOfLength(10), FsRepository.TYPE, settings);
131+
return new FsRepository(repositoryMetaData, createEnvironment(), xContentRegistry());
132+
}
133+
134+
/** Create a {@link Environment} with random path.home and path.repo **/
135+
private Environment createEnvironment() {
136+
Path home = createTempDir();
137+
return new Environment(Settings.builder()
138+
.put(Environment.PATH_HOME_SETTING.getKey(), home.toAbsolutePath())
139+
.put(Environment.PATH_REPO_SETTING.getKey(), home.resolve("repo").toAbsolutePath())
140+
.build());
141+
}
142+
}

test/framework/src/main/java/org/elasticsearch/index/shard/IndexShardTestCase.java

+38
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
import org.elasticsearch.common.util.BigArrays;
4747
import org.elasticsearch.common.xcontent.XContentType;
4848
import org.elasticsearch.env.NodeEnvironment;
49+
import org.elasticsearch.index.Index;
4950
import org.elasticsearch.index.IndexSettings;
5051
import org.elasticsearch.index.MapperTestUtils;
5152
import org.elasticsearch.index.VersionType;
@@ -60,6 +61,7 @@
6061
import org.elasticsearch.index.mapper.Uid;
6162
import org.elasticsearch.index.seqno.SequenceNumbers;
6263
import org.elasticsearch.index.similarity.SimilarityService;
64+
import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
6365
import org.elasticsearch.index.store.DirectoryService;
6466
import org.elasticsearch.index.store.Store;
6567
import org.elasticsearch.indices.recovery.PeerRecoveryTargetService;
@@ -69,6 +71,9 @@
6971
import org.elasticsearch.indices.recovery.RecoveryTarget;
7072
import org.elasticsearch.indices.recovery.StartRecoveryRequest;
7173
import org.elasticsearch.node.Node;
74+
import org.elasticsearch.repositories.IndexId;
75+
import org.elasticsearch.repositories.Repository;
76+
import org.elasticsearch.snapshots.Snapshot;
7277
import org.elasticsearch.test.DummyShardLock;
7378
import org.elasticsearch.test.ESTestCase;
7479
import org.elasticsearch.threadpool.TestThreadPool;
@@ -85,6 +90,7 @@
8590
import java.util.function.BiFunction;
8691
import java.util.function.Consumer;
8792

93+
import static org.elasticsearch.cluster.routing.TestShardRouting.newShardRouting;
8894
import static org.hamcrest.Matchers.contains;
8995
import static org.hamcrest.Matchers.hasSize;
9096

@@ -583,6 +589,38 @@ protected void flushShard(IndexShard shard, boolean force) {
583589
shard.flush(new FlushRequest(shard.shardId().getIndexName()).force(force));
584590
}
585591

592+
/** Recover a shard from a snapshot using a given repository **/
593+
protected void recoverShardFromSnapshot(final IndexShard shard,
594+
final Snapshot snapshot,
595+
final Repository repository) throws IOException {
596+
final Version version = Version.CURRENT;
597+
final ShardId shardId = shard.shardId();
598+
final String index = shardId.getIndexName();
599+
final IndexId indexId = new IndexId(shardId.getIndex().getName(), shardId.getIndex().getUUID());
600+
final DiscoveryNode node = getFakeDiscoNode(shard.routingEntry().currentNodeId());
601+
final RecoverySource.SnapshotRecoverySource recoverySource = new RecoverySource.SnapshotRecoverySource(snapshot, version, index);
602+
final ShardRouting shardRouting = newShardRouting(shardId, node.getId(), true, ShardRoutingState.INITIALIZING, recoverySource);
603+
604+
shard.markAsRecovering("from snapshot", new RecoveryState(shardRouting, node, null));
605+
repository.restoreShard(shard, snapshot.getSnapshotId(), version, indexId, shard.shardId(), shard.recoveryState());
606+
}
607+
608+
/** Snapshot a shard using a given repository **/
609+
protected void snapshotShard(final IndexShard shard,
610+
final Snapshot snapshot,
611+
final Repository repository) throws IOException {
612+
final IndexShardSnapshotStatus snapshotStatus = new IndexShardSnapshotStatus();
613+
try (Engine.IndexCommitRef indexCommitRef = shard.acquireIndexCommit(true)) {
614+
Index index = shard.shardId().getIndex();
615+
IndexId indexId = new IndexId(index.getName(), index.getUUID());
616+
617+
repository.snapshotShard(shard, snapshot.getSnapshotId(), indexId, indexCommitRef.getIndexCommit(), snapshotStatus);
618+
}
619+
assertEquals(IndexShardSnapshotStatus.Stage.DONE, snapshotStatus.stage());
620+
assertEquals(shard.snapshotStoreMetadata().size(), snapshotStatus.numberOfFiles());
621+
assertNull(snapshotStatus.failure());
622+
}
623+
586624
/**
587625
* Helper method to access (package-protected) engine from tests
588626
*/

0 commit comments

Comments
 (0)