diff --git a/plugins/repository-hdfs/src/test/java/org/elasticsearch/repositories/hdfs/HdfsTests.java b/plugins/repository-hdfs/src/test/java/org/elasticsearch/repositories/hdfs/HdfsTests.java index 8c211a6cabb43..41423d25b9f9f 100644 --- a/plugins/repository-hdfs/src/test/java/org/elasticsearch/repositories/hdfs/HdfsTests.java +++ b/plugins/repository-hdfs/src/test/java/org/elasticsearch/repositories/hdfs/HdfsTests.java @@ -27,9 +27,13 @@ import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.plugins.Plugin; +import org.elasticsearch.repositories.RepositoriesService; import org.elasticsearch.repositories.RepositoryException; +import org.elasticsearch.repositories.blobstore.BlobStoreRepository; +import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil; import org.elasticsearch.snapshots.SnapshotState; import org.elasticsearch.test.ESSingleNodeTestCase; +import org.elasticsearch.threadpool.ThreadPool; import java.util.Collection; @@ -145,6 +149,9 @@ public void testSimpleWorkflow() { ClusterState clusterState = client.admin().cluster().prepareState().get().getState(); assertThat(clusterState.getMetaData().hasIndex("test-idx-1"), equalTo(true)); assertThat(clusterState.getMetaData().hasIndex("test-idx-2"), equalTo(false)); + final BlobStoreRepository repo = + (BlobStoreRepository) getInstanceFromNode(RepositoriesService.class).repository("test-repo"); + BlobStoreTestUtil.assertConsistency(repo, repo.threadPool().executor(ThreadPool.Names.GENERIC)); } public void testMissingUri() { diff --git a/plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryThirdPartyTests.java b/plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryThirdPartyTests.java index bdaace00f80ab..7e7ac8d430062 100644 --- a/plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryThirdPartyTests.java +++ b/plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryThirdPartyTests.java @@ -26,10 +26,12 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.plugins.Plugin; import org.elasticsearch.repositories.AbstractThirdPartyRepositoryTestCase; +import org.elasticsearch.repositories.blobstore.BlobStoreRepository; import org.elasticsearch.test.StreamsUtils; import java.io.IOException; import java.util.Collection; +import java.util.concurrent.Executor; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -76,6 +78,20 @@ protected void createRepository(String repoName) { } @Override + protected boolean assertCorruptionVisible(BlobStoreRepository repo, Executor genericExec) throws Exception { + // S3 is only eventually consistent for the list operations used by this assertions so we retry for 10 minutes assuming that + // listing operations will become consistent within these 10 minutes. + assertBusy(() -> assertTrue(super.assertCorruptionVisible(repo, genericExec)), 10L, TimeUnit.MINUTES); + return true; + } + + @Override + protected void assertConsistentRepository(BlobStoreRepository repo, Executor executor) throws Exception { + // S3 is only eventually consistent for the list operations used by this assertions so we retry for 10 minutes assuming that + // listing operations will become consistent within these 10 minutes. + assertBusy(() -> super.assertConsistentRepository(repo, executor), 10L, TimeUnit.MINUTES); + } + protected void assertBlobsByPrefix(BlobPath path, String prefix, Map blobs) throws Exception { // AWS S3 is eventually consistent so we retry for 10 minutes assuming a list operation will never take longer than that // to become consistent. diff --git a/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java b/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java index 9151c512dd325..1aa74776118e6 100644 --- a/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java +++ b/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java @@ -58,7 +58,6 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.ByteSizeUnit; import org.elasticsearch.common.unit.ByteSizeValue; -import org.elasticsearch.common.util.set.Sets; import org.elasticsearch.common.xcontent.LoggingDeprecationHandler; import org.elasticsearch.common.xcontent.NamedXContentRegistry; import org.elasticsearch.common.xcontent.XContentFactory; @@ -393,46 +392,68 @@ public void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId, Action logger.warn(() -> new ParameterizedMessage("cannot read snapshot file [{}]", snapshotId), ex); } // Delete snapshot from the index file, since it is the maintainer of truth of active snapshots - final RepositoryData repositoryData; final RepositoryData updatedRepositoryData; + final Map foundIndices; try { - repositoryData = getRepositoryData(); + final RepositoryData repositoryData = getRepositoryData(); updatedRepositoryData = repositoryData.removeSnapshot(snapshotId); + // Cache the indices that were found before writing out the new index-N blob so that a stuck master will never + // delete an index that was created by another master node after writing this index-N blob. + foundIndices = blobStore().blobContainer(basePath().add("indices")).children(); writeIndexGen(updatedRepositoryData, repositoryStateId); } catch (Exception ex) { listener.onFailure(new RepositoryException(metadata.name(), "failed to delete snapshot [" + snapshotId + "]", ex)); return; } final SnapshotInfo finalSnapshotInfo = snapshot; - final Collection unreferencedIndices = Sets.newHashSet(repositoryData.getIndices().values()); - unreferencedIndices.removeAll(updatedRepositoryData.getIndices().values()); try { blobContainer().deleteBlobsIgnoringIfNotExists( Arrays.asList(snapshotFormat.blobName(snapshotId.getUUID()), globalMetaDataFormat.blobName(snapshotId.getUUID()))); } catch (IOException e) { logger.warn(() -> new ParameterizedMessage("[{}] Unable to delete global metadata files", snapshotId), e); } + final var survivingIndices = updatedRepositoryData.getIndices(); deleteIndices( Optional.ofNullable(finalSnapshotInfo) - .map(info -> info.indices().stream().map(repositoryData::resolveIndexId).collect(Collectors.toList())) + .map(info -> info.indices().stream().filter(survivingIndices::containsKey) + .map(updatedRepositoryData::resolveIndexId).collect(Collectors.toList())) .orElse(Collections.emptyList()), snapshotId, ActionListener.map(listener, v -> { - try { - blobStore().blobContainer(indicesPath()).deleteBlobsIgnoringIfNotExists( - unreferencedIndices.stream().map(IndexId::getId).collect(Collectors.toList())); - } catch (IOException e) { - logger.warn(() -> - new ParameterizedMessage( - "[{}] indices {} are no longer part of any snapshots in the repository, " + - "but failed to clean up their index folders.", metadata.name(), unreferencedIndices), e); - } + cleanupStaleIndices(foundIndices, survivingIndices); return null; }) ); } } + private void cleanupStaleIndices(Map foundIndices, Map survivingIndices) { + try { + final Set survivingIndexIds = survivingIndices.values().stream() + .map(IndexId::getId).collect(Collectors.toSet()); + for (Map.Entry indexEntry : foundIndices.entrySet()) { + final String indexSnId = indexEntry.getKey(); + try { + if (survivingIndexIds.contains(indexSnId) == false) { + logger.debug("[{}] Found stale index [{}]. Cleaning it up", metadata.name(), indexSnId); + indexEntry.getValue().delete(); + logger.debug("[{}] Cleaned up stale index [{}]", metadata.name(), indexSnId); + } + } catch (IOException e) { + logger.warn(() -> new ParameterizedMessage( + "[{}] index {} is no longer part of any snapshots in the repository, " + + "but failed to clean up their index folders", metadata.name(), indexSnId), e); + } + } + } catch (Exception e) { + // TODO: We shouldn't be blanket catching and suppressing all exceptions here and instead handle them safely upstream. + // Currently this catch exists as a stop gap solution to tackle unexpected runtime exceptions from implementations + // bubbling up and breaking the snapshot functionality. + assert false : e; + logger.warn(new ParameterizedMessage("[{}] Exception during cleanup of stale indices", metadata.name()), e); + } + } + private void deleteIndices(List indices, SnapshotId snapshotId, ActionListener listener) { if (indices.isEmpty()) { listener.onResponse(null); @@ -494,9 +515,9 @@ public SnapshotInfo finalizeSnapshot(final SnapshotId snapshotId, startTime, failure, System.currentTimeMillis(), totalShards, shardFailures, includeGlobalState, userMetadata); try { + final RepositoryData updatedRepositoryData = getRepositoryData().addSnapshot(snapshotId, blobStoreSnapshot.state(), indices); snapshotFormat.write(blobStoreSnapshot, blobContainer(), snapshotId.getUUID()); - final RepositoryData repositoryData = getRepositoryData(); - writeIndexGen(repositoryData.addSnapshot(snapshotId, blobStoreSnapshot.state(), indices), repositoryStateId); + writeIndexGen(updatedRepositoryData, repositoryStateId); } catch (FileAlreadyExistsException ex) { // if another master was elected and took over finalizing the snapshot, it is possible // that both nodes try to finalize the snapshot and write to the same blobs, so we just diff --git a/server/src/test/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java b/server/src/test/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java index 032e8a982ed4b..9f680f11f2117 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java +++ b/server/src/test/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java @@ -20,11 +20,13 @@ import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse; import org.elasticsearch.cluster.SnapshotsInProgress; +import org.elasticsearch.cluster.metadata.RepositoryMetaData; import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.plugins.Plugin; import org.elasticsearch.repositories.RepositoriesService; +import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil; import org.elasticsearch.snapshots.mockstore.MockRepository; import org.elasticsearch.test.ESIntegTestCase; import org.junit.After; @@ -65,6 +67,32 @@ public void assertConsistentHistoryInLuceneIndex() throws Exception { internalCluster().assertConsistentHistoryBetweenTranslogAndLuceneIndex(); } + private String skipRepoConsistencyCheckReason; + + @After + public void assertRepoConsistency() { + if (skipRepoConsistencyCheckReason == null) { + client().admin().cluster().prepareGetRepositories().get().repositories() + .stream() + .map(RepositoryMetaData::name) + .forEach(name -> { + final List snapshots = client().admin().cluster().prepareGetSnapshots(name).get().getSnapshots(name); + // Delete one random snapshot to trigger repository cleanup. + if (snapshots.isEmpty() == false) { + client().admin().cluster().prepareDeleteSnapshot(name, randomFrom(snapshots).snapshotId().getName()).get(); + } + BlobStoreTestUtil.assertRepoConsistency(internalCluster(), name); + }); + } else { + logger.info("--> skipped repo consistency checks because [{}]", skipRepoConsistencyCheckReason); + } + } + + protected void disableRepoConsistencyCheck(String reason) { + assertNotNull(reason); + skipRepoConsistencyCheckReason = reason; + } + public static long getFailureCount(String repository) { long failureCount = 0; for (RepositoriesService repositoriesService : diff --git a/server/src/test/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java b/server/src/test/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java index fb9b295e8e643..302dc5a57730f 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java +++ b/server/src/test/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java @@ -722,6 +722,7 @@ public boolean clearData(String nodeName) { } public void testRegistrationFailure() { + disableRepoConsistencyCheck("This test does not create any data in the repository"); logger.info("--> start first node"); internalCluster().startNode(); logger.info("--> start second node"); @@ -741,6 +742,7 @@ public void testRegistrationFailure() { } public void testThatSensitiveRepositorySettingsAreNotExposed() throws Exception { + disableRepoConsistencyCheck("This test does not create any data in the repository"); Settings nodeSettings = Settings.EMPTY; logger.info("--> start two nodes"); internalCluster().startNodes(2, nodeSettings); diff --git a/server/src/test/java/org/elasticsearch/snapshots/MetadataLoadingDuringSnapshotRestoreIT.java b/server/src/test/java/org/elasticsearch/snapshots/MetadataLoadingDuringSnapshotRestoreIT.java index 367dc44a9a686..99cf9d14069b0 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/MetadataLoadingDuringSnapshotRestoreIT.java +++ b/server/src/test/java/org/elasticsearch/snapshots/MetadataLoadingDuringSnapshotRestoreIT.java @@ -141,8 +141,8 @@ public void testWhenMetadataAreLoaded() throws Exception { // Deleting a snapshot does not load the global metadata state but loads each index metadata assertAcked(client().admin().cluster().prepareDeleteSnapshot("repository", "snap").get()); assertGlobalMetadataLoads("snap", 1); - assertIndexMetadataLoads("snap", "docs", 5); - assertIndexMetadataLoads("snap", "others", 4); + assertIndexMetadataLoads("snap", "docs", 4); + assertIndexMetadataLoads("snap", "others", 3); } private void assertGlobalMetadataLoads(final String snapshot, final int times) { diff --git a/server/src/test/java/org/elasticsearch/snapshots/RepositoriesIT.java b/server/src/test/java/org/elasticsearch/snapshots/RepositoriesIT.java index 63e2c31ea3e96..455a3f18e5472 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/RepositoriesIT.java +++ b/server/src/test/java/org/elasticsearch/snapshots/RepositoriesIT.java @@ -184,6 +184,8 @@ public void testRepositoryAckTimeout() throws Exception { } public void testRepositoryVerification() throws Exception { + disableRepoConsistencyCheck("This test does not create any data in the repository."); + Client client = client(); Settings settings = Settings.builder() diff --git a/server/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java b/server/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java index 00f416e5ce52a..7168efb6b55e0 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java +++ b/server/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java @@ -362,7 +362,7 @@ public void testSingleGetAfterRestore() throws Exception { assertThat(client.prepareGet(restoredIndexName, typeName, docId).get().isExists(), equalTo(true)); } - public void testFreshIndexUUID() { + public void testFreshIndexUUID() throws InterruptedException { Client client = client(); logger.info("--> creating repository"); @@ -540,7 +540,6 @@ public void testRestoreAliases() throws Exception { logger.info("--> check that aliases are not restored and existing aliases still exist"); assertAliasesMissing(client.admin().indices().prepareAliasesExist("alias-123", "alias-1").get()); assertAliasesExist(client.admin().indices().prepareAliasesExist("alias-3").get()); - } public void testRestoreTemplates() throws Exception { @@ -593,7 +592,6 @@ public void testRestoreTemplates() throws Exception { logger.info("--> check that template is restored"); getIndexTemplatesResponse = client().admin().indices().prepareGetTemplates().get(); assertIndexTemplateExists(getIndexTemplatesResponse, "test-template"); - } public void testIncludeGlobalState() throws Exception { @@ -780,10 +778,10 @@ public void testIncludeGlobalState() throws Exception { assertFalse(client().admin().cluster().prepareGetPipeline("barbaz").get().isFound()); assertNull(client().admin().cluster().prepareGetStoredScript("foobar").get().getSource()); assertThat(client.prepareSearch("test-idx").setSize(0).get().getHits().getTotalHits().value, equalTo(100L)); - } - public void testSnapshotFileFailureDuringSnapshot() { + public void testSnapshotFileFailureDuringSnapshot() throws InterruptedException { + disableRepoConsistencyCheck("This test uses a purposely broken repository so it would fail consistency checks"); Client client = client(); logger.info("--> creating repository"); @@ -910,6 +908,8 @@ public void testDataFileFailureDuringSnapshot() throws Exception { } public void testDataFileFailureDuringRestore() throws Exception { + disableRepoConsistencyCheck("This test intentionally leaves a broken repository"); + Path repositoryLocation = randomRepoPath(); Client client = client(); logger.info("--> creating repository"); @@ -973,6 +973,8 @@ public void testDataFileFailureDuringRestore() throws Exception { } public void testDataFileCorruptionDuringRestore() throws Exception { + disableRepoConsistencyCheck("This test intentionally leaves a broken repository"); + Path repositoryLocation = randomRepoPath(); Client client = client(); logger.info("--> creating repository"); @@ -1237,7 +1239,6 @@ public void testDeletionOfFailingToRecoverIndexShouldStopRestore() throws Except assertThat(restoreSnapshotResponse.getRestoreInfo().failedShards(), equalTo(0)); SearchResponse countResponse = client.prepareSearch("test-idx").setSize(0).get(); assertThat(countResponse.getHits().getTotalHits().value, equalTo(100L)); - } public void testUnallocatedShards() throws Exception { @@ -1786,8 +1787,6 @@ public void testRenameOnRestore() throws Exception { .setIndices("test-idx-1").setRenamePattern("test-idx").setRenameReplacement("alias") .setWaitForCompletion(true).setIncludeAliases(false).execute().actionGet(); assertThat(restoreSnapshotResponse.getRestoreInfo().totalShards(), greaterThan(0)); - - } public void testMoveShardWhileSnapshotting() throws Exception { @@ -1854,6 +1853,7 @@ public void testMoveShardWhileSnapshotting() throws Exception { } public void testDeleteRepositoryWhileSnapshotting() throws Exception { + disableRepoConsistencyCheck("This test uses a purposely broken repository so it would fail consistency checks"); Client client = client(); Path repositoryLocation = randomRepoPath(); logger.info("--> creating repository"); @@ -2412,7 +2412,6 @@ public void testChangeSettingsOnRestore() throws Exception { assertHitCount(client.prepareSearch("test-idx").setSize(0).setQuery(matchQuery("field1", "Foo")).get(), numdocs); assertHitCount(client.prepareSearch("test-idx").setSize(0).setQuery(matchQuery("field1", "bar")).get(), numdocs); - } public void testRecreateBlocksOnRestore() throws Exception { @@ -2506,6 +2505,8 @@ public void testRecreateBlocksOnRestore() throws Exception { } public void testCloseOrDeleteIndexDuringSnapshot() throws Exception { + disableRepoConsistencyCheck("This test intentionally leaves a broken repository"); + Client client = client(); boolean allowPartial = randomBoolean(); @@ -2827,6 +2828,8 @@ private boolean waitForRelocationsToStart(final String index, TimeValue timeout) } public void testSnapshotName() throws Exception { + disableRepoConsistencyCheck("This test does not create any data in the repository"); + final Client client = client(); logger.info("--> creating repository"); @@ -2848,6 +2851,8 @@ public void testSnapshotName() throws Exception { } public void testListCorruptedSnapshot() throws Exception { + disableRepoConsistencyCheck("This test intentionally leaves a broken repository"); + Client client = client(); Path repo = randomRepoPath(); logger.info("--> creating repository at {}", repo.toAbsolutePath()); @@ -3418,6 +3423,9 @@ public void testSnapshotCanceledOnRemovedShard() throws Exception { } public void testSnapshotSucceedsAfterSnapshotFailure() throws Exception { + // TODO: Fix repo cleanup logic to handle these leaked snap-file and only exclude test-repo (the mock repo) here. + disableRepoConsistencyCheck( + "This test uses a purposely broken repository implementation that results in leaking snap-{uuid}.dat files"); logger.info("--> creating repository"); final Path repoPath = randomRepoPath(); final Client client = client(); diff --git a/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java b/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java index 912c274e8dbd7..86573ffc01f9e 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java +++ b/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java @@ -107,8 +107,6 @@ import org.elasticsearch.cluster.service.ClusterApplierService; import org.elasticsearch.cluster.service.ClusterService; import org.elasticsearch.cluster.service.MasterService; -import org.elasticsearch.common.Strings; -import org.elasticsearch.common.bytes.BytesArray; import org.elasticsearch.common.io.stream.NamedWriteableRegistry; import org.elasticsearch.common.network.NetworkModule; import org.elasticsearch.common.settings.ClusterSettings; @@ -119,11 +117,7 @@ import org.elasticsearch.common.util.PageCacheRecycler; import org.elasticsearch.common.util.concurrent.AbstractRunnable; import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor; -import org.elasticsearch.common.xcontent.LoggingDeprecationHandler; import org.elasticsearch.common.xcontent.NamedXContentRegistry; -import org.elasticsearch.common.xcontent.XContentHelper; -import org.elasticsearch.common.xcontent.XContentParser; -import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.env.Environment; import org.elasticsearch.env.NodeEnvironment; import org.elasticsearch.env.TestEnvironment; @@ -148,10 +142,10 @@ import org.elasticsearch.ingest.IngestService; import org.elasticsearch.node.ResponseCollectorService; import org.elasticsearch.plugins.PluginsService; -import org.elasticsearch.repositories.IndexId; import org.elasticsearch.repositories.RepositoriesService; import org.elasticsearch.repositories.Repository; -import org.elasticsearch.repositories.RepositoryData; +import org.elasticsearch.repositories.blobstore.BlobStoreRepository; +import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil; import org.elasticsearch.repositories.fs.FsRepository; import org.elasticsearch.script.ScriptService; import org.elasticsearch.search.SearchService; @@ -169,16 +163,8 @@ import org.junit.After; import org.junit.Before; -import java.io.FileNotFoundException; import java.io.IOException; -import java.nio.ByteBuffer; -import java.nio.file.DirectoryNotEmptyException; -import java.nio.file.FileVisitResult; -import java.nio.file.Files; -import java.nio.file.NoSuchFileException; import java.nio.file.Path; -import java.nio.file.SimpleFileVisitor; -import java.nio.file.attribute.BasicFileAttributes; import java.util.Collection; import java.util.Collections; import java.util.Comparator; @@ -224,9 +210,11 @@ public void createServices() { } @After - public void verifyReposThenStopServices() throws IOException { + public void verifyReposThenStopServices() { try { - assertNoStaleRepositoryData(); + BlobStoreTestUtil.assertConsistency( + (BlobStoreRepository) testClusterNodes.randomMasterNodeSafe().repositoriesService.repository("repo"), + Runnable::run); } finally { testClusterNodes.nodes.values().forEach(TestClusterNode::stop); } @@ -524,109 +512,6 @@ public void run() { assertThat(snapshotIds, either(hasSize(1)).or(hasSize(0))); } - /** - * Assert that there are no unreferenced indices or unreferenced root-level metadata blobs in any repository. - * TODO: Expand the logic here to also check for unreferenced segment blobs and shard level metadata - */ - private void assertNoStaleRepositoryData() throws IOException { - final Path repoPath = tempDir.resolve("repo").toAbsolutePath(); - final List repos; - try (Stream reposDir = repoFilesByPrefix(repoPath)) { - repos = reposDir.filter(s -> s.getFileName().toString().startsWith("extra") == false).collect(Collectors.toList()); - } - for (Path repoRoot : repos) { - cleanupEmptyTrees(repoRoot); - final Path latestIndexGenBlob = repoRoot.resolve("index.latest"); - assertTrue("Could not find index.latest blob for repo at [" + repoRoot + ']', Files.exists(latestIndexGenBlob)); - final long latestGen = ByteBuffer.wrap(Files.readAllBytes(latestIndexGenBlob)).getLong(0); - assertIndexGenerations(repoRoot, latestGen); - final RepositoryData repositoryData; - try (XContentParser parser = - XContentHelper.createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, - new BytesArray(Files.readAllBytes(repoRoot.resolve("index-" + latestGen))), XContentType.JSON)) { - repositoryData = RepositoryData.snapshotsFromXContent(parser, latestGen); - } - assertIndexUUIDs(repoRoot, repositoryData); - assertSnapshotUUIDs(repoRoot, repositoryData); - } - } - - // Lucene's mock file system randomly generates empty `extra0` files that break the deletion of blob-store directories. - // We clean those up here before checking a blob-store for stale files. - private void cleanupEmptyTrees(Path repoPath) { - try { - Files.walkFileTree(repoPath, new SimpleFileVisitor<>() { - - @Override - public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException { - if (file.getFileName().toString().startsWith("extra")) { - Files.delete(file); - } - return FileVisitResult.CONTINUE; - } - - @Override - public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException { - try { - Files.delete(dir); - } catch (DirectoryNotEmptyException e) { - // We're only interested in deleting empty trees here, just ignore directories with content - } - return FileVisitResult.CONTINUE; - } - }); - } catch (IOException e) { - throw new AssertionError(e); - } - } - - private static void assertIndexGenerations(Path repoRoot, long latestGen) throws IOException { - try (Stream repoRootBlobs = repoFilesByPrefix(repoRoot)) { - final long[] indexGenerations = repoRootBlobs.filter(p -> p.getFileName().toString().startsWith("index-")) - .map(p -> p.getFileName().toString().replace("index-", "")) - .mapToLong(Long::parseLong).sorted().toArray(); - assertEquals(latestGen, indexGenerations[indexGenerations.length - 1]); - assertTrue(indexGenerations.length <= 2); - } - } - - private static void assertIndexUUIDs(Path repoRoot, RepositoryData repositoryData) throws IOException { - final List expectedIndexUUIDs = - repositoryData.getIndices().values().stream().map(IndexId::getId).collect(Collectors.toList()); - try (Stream indexRoots = repoFilesByPrefix(repoRoot.resolve("indices"))) { - final List foundIndexUUIDs = indexRoots.filter(s -> s.getFileName().toString().startsWith("extra") == false) - .map(p -> p.getFileName().toString()).collect(Collectors.toList()); - assertThat(foundIndexUUIDs, containsInAnyOrder(expectedIndexUUIDs.toArray(Strings.EMPTY_ARRAY))); - } - } - - private static void assertSnapshotUUIDs(Path repoRoot, RepositoryData repositoryData) throws IOException { - final List expectedSnapshotUUIDs = - repositoryData.getSnapshotIds().stream().map(SnapshotId::getUUID).collect(Collectors.toList()); - for (String prefix : new String[]{"snap-", "meta-"}) { - try (Stream repoRootBlobs = repoFilesByPrefix(repoRoot)) { - final Collection foundSnapshotUUIDs = repoRootBlobs.filter(p -> p.getFileName().toString().startsWith(prefix)) - .map(p -> p.getFileName().toString().replace(prefix, "").replace(".dat", "")) - .collect(Collectors.toSet()); - assertThat(foundSnapshotUUIDs, containsInAnyOrder(expectedSnapshotUUIDs.toArray(Strings.EMPTY_ARRAY))); - } - } - } - - /** - * List contents of a blob path and return an empty stream if the path doesn't exist. - * @param prefix Path to find children for - * @return stream of child paths - * @throws IOException on failure - */ - private static Stream repoFilesByPrefix(Path prefix) throws IOException { - try { - return Files.list(prefix); - } catch (FileNotFoundException | NoSuchFileException e) { - return Stream.empty(); - } - } - private void clearDisruptionsAndAwaitSync() { testClusterNodes.clearNetworkDisruptions(); runUntil(() -> { diff --git a/server/src/test/java/org/elasticsearch/snapshots/mockstore/MockRepository.java b/server/src/test/java/org/elasticsearch/snapshots/mockstore/MockRepository.java index 6879406d13a40..21ade636f53df 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/mockstore/MockRepository.java +++ b/server/src/test/java/org/elasticsearch/snapshots/mockstore/MockRepository.java @@ -59,6 +59,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; public class MockRepository extends FsRepository { private static final Logger logger = LogManager.getLogger(MockRepository.class); @@ -333,6 +334,17 @@ public void deleteBlobIgnoringIfNotExists(String blobName) throws IOException { super.deleteBlobIgnoringIfNotExists(blobName); } + @Override + public void delete() throws IOException { + for (BlobContainer child : children().values()) { + child.delete(); + } + for (String blob : listBlobs().values().stream().map(BlobMetaData::name).collect(Collectors.toList())) { + deleteBlobIgnoringIfNotExists(blob); + } + blobStore().blobContainer(path().parent()).deleteBlob(path().toArray()[path().toArray().length - 1]); + } + @Override public Map listBlobs() throws IOException { maybeIOExceptionOrBlock(""); diff --git a/test/framework/src/main/java/org/elasticsearch/repositories/AbstractThirdPartyRepositoryTestCase.java b/test/framework/src/main/java/org/elasticsearch/repositories/AbstractThirdPartyRepositoryTestCase.java index 52ed98ac811ee..beab182176694 100644 --- a/test/framework/src/main/java/org/elasticsearch/repositories/AbstractThirdPartyRepositoryTestCase.java +++ b/test/framework/src/main/java/org/elasticsearch/repositories/AbstractThirdPartyRepositoryTestCase.java @@ -20,6 +20,7 @@ import org.elasticsearch.action.ActionRunnable; import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse; +import org.elasticsearch.action.admin.cluster.snapshots.delete.DeleteSnapshotRequest; import org.elasticsearch.action.support.PlainActionFuture; import org.elasticsearch.common.Strings; import org.elasticsearch.common.blobstore.BlobMetaData; @@ -29,8 +30,10 @@ import org.elasticsearch.common.settings.SecureSettings; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.repositories.blobstore.BlobStoreRepository; +import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil; import org.elasticsearch.snapshots.SnapshotState; import org.elasticsearch.test.ESSingleNodeTestCase; +import org.elasticsearch.threadpool.ThreadPool; import java.io.ByteArrayInputStream; import java.util.Collection; @@ -184,6 +187,86 @@ protected void doRun() throws Exception { } } + public void testCleanup() throws Exception { + createRepository("test-repo"); + + createIndex("test-idx-1"); + createIndex("test-idx-2"); + createIndex("test-idx-3"); + ensureGreen(); + + logger.info("--> indexing some data"); + for (int i = 0; i < 100; i++) { + client().prepareIndex("test-idx-1", "doc", Integer.toString(i)).setSource("foo", "bar" + i).get(); + client().prepareIndex("test-idx-2", "doc", Integer.toString(i)).setSource("foo", "bar" + i).get(); + client().prepareIndex("test-idx-3", "doc", Integer.toString(i)).setSource("foo", "bar" + i).get(); + } + client().admin().indices().prepareRefresh().get(); + + final String snapshotName = "test-snap-" + System.currentTimeMillis(); + + logger.info("--> snapshot"); + CreateSnapshotResponse createSnapshotResponse = client().admin() + .cluster() + .prepareCreateSnapshot("test-repo", snapshotName) + .setWaitForCompletion(true) + .setIndices("test-idx-*", "-test-idx-3") + .get(); + assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), greaterThan(0)); + assertThat(createSnapshotResponse.getSnapshotInfo().successfulShards(), + equalTo(createSnapshotResponse.getSnapshotInfo().totalShards())); + + assertThat(client().admin() + .cluster() + .prepareGetSnapshots("test-repo") + .setSnapshots(snapshotName) + .get() + .getSnapshots("test-repo") + .get(0) + .state(), + equalTo(SnapshotState.SUCCESS)); + + logger.info("--> creating a dangling index folder"); + final BlobStoreRepository repo = + (BlobStoreRepository) getInstanceFromNode(RepositoriesService.class).repository("test-repo"); + final PlainActionFuture future = PlainActionFuture.newFuture(); + final Executor genericExec = repo.threadPool().executor(ThreadPool.Names.GENERIC); + genericExec.execute(new ActionRunnable<>(future) { + @Override + protected void doRun() throws Exception { + final BlobStore blobStore = repo.blobStore(); + blobStore.blobContainer(BlobPath.cleanPath().add("indices").add("foo")) + .writeBlob("bar", new ByteArrayInputStream(new byte[0]), 0, false); + future.onResponse(null); + } + }); + future.actionGet(); + assertTrue(assertCorruptionVisible(repo, genericExec)); + logger.info("--> deleting a snapshot to trigger repository cleanup"); + client().admin().cluster().deleteSnapshot(new DeleteSnapshotRequest("test-repo", snapshotName)).actionGet(); + + assertConsistentRepository(repo, genericExec); + } + + protected boolean assertCorruptionVisible(BlobStoreRepository repo, Executor executor) throws Exception { + final PlainActionFuture future = PlainActionFuture.newFuture(); + executor.execute(new ActionRunnable<>(future) { + @Override + protected void doRun() throws Exception { + final BlobStore blobStore = repo.blobStore(); + future.onResponse( + blobStore.blobContainer(BlobPath.cleanPath().add("indices")).children().containsKey("foo") + && blobStore.blobContainer(BlobPath.cleanPath().add("indices").add("foo")).blobExists("bar") + ); + } + }); + return future.actionGet(); + } + + protected void assertConsistentRepository(BlobStoreRepository repo, Executor executor) throws Exception { + BlobStoreTestUtil.assertConsistency(repo, executor); + } + protected void assertDeleted(BlobPath path, String name) throws Exception { assertThat(listChildren(path), not(contains(name))); } diff --git a/test/framework/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreTestUtil.java b/test/framework/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreTestUtil.java new file mode 100644 index 0000000000000..8bdb18d1b3dfa --- /dev/null +++ b/test/framework/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreTestUtil.java @@ -0,0 +1,133 @@ +/* + * Licensed to Elasticsearch under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch licenses this file to you under + * the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.elasticsearch.repositories.blobstore; + +import org.elasticsearch.action.ActionRunnable; +import org.elasticsearch.action.support.PlainActionFuture; +import org.elasticsearch.common.Strings; +import org.elasticsearch.common.blobstore.BlobContainer; +import org.elasticsearch.common.io.stream.BytesStreamOutput; +import org.elasticsearch.common.unit.TimeValue; +import org.elasticsearch.common.xcontent.LoggingDeprecationHandler; +import org.elasticsearch.common.xcontent.NamedXContentRegistry; +import org.elasticsearch.common.xcontent.XContentHelper; +import org.elasticsearch.common.xcontent.XContentParser; +import org.elasticsearch.common.xcontent.XContentType; +import org.elasticsearch.core.internal.io.Streams; +import org.elasticsearch.repositories.IndexId; +import org.elasticsearch.repositories.RepositoriesService; +import org.elasticsearch.repositories.RepositoryData; +import org.elasticsearch.snapshots.SnapshotId; +import org.elasticsearch.test.InternalTestCluster; +import org.elasticsearch.threadpool.ThreadPool; + +import java.io.DataInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.stream.Collectors; + +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +public final class BlobStoreTestUtil { + + public static void assertRepoConsistency(InternalTestCluster testCluster, String repoName) { + final BlobStoreRepository repo = + (BlobStoreRepository) testCluster.getCurrentMasterNodeInstance(RepositoriesService.class).repository(repoName); + BlobStoreTestUtil.assertConsistency(repo, repo.threadPool().executor(ThreadPool.Names.GENERIC)); + } + + /** + * Assert that there are no unreferenced indices or unreferenced root-level metadata blobs in any repository. + * TODO: Expand the logic here to also check for unreferenced segment blobs and shard level metadata + * @param repository BlobStoreRepository to check + * @param executor Executor to run all repository calls on. This is needed since the production {@link BlobStoreRepository} + * implementations assert that all IO inducing calls happen on the generic or snapshot thread-pools and hence callers + * of this assertion must pass an executor on those when using such an implementation. + */ + public static void assertConsistency(BlobStoreRepository repository, Executor executor) { + final PlainActionFuture listener = PlainActionFuture.newFuture(); + executor.execute(new ActionRunnable<>(listener) { + @Override + protected void doRun() throws Exception { + final BlobContainer blobContainer = repository.blobContainer(); + assertTrue( + "Could not find index.latest blob for repo [" + repository + "]", blobContainer.blobExists("index.latest")); + final long latestGen; + try (DataInputStream inputStream = new DataInputStream(blobContainer.readBlob("index.latest"))) { + latestGen = inputStream.readLong(); + } + assertIndexGenerations(blobContainer, latestGen); + final RepositoryData repositoryData; + try (InputStream inputStream = blobContainer.readBlob("index-" + latestGen); + BytesStreamOutput out = new BytesStreamOutput()) { + Streams.copy(inputStream, out); + try (XContentParser parser = + XContentHelper.createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, + out.bytes(), XContentType.JSON)) { + repositoryData = RepositoryData.snapshotsFromXContent(parser, latestGen); + } + } + assertIndexUUIDs(blobContainer, repositoryData); + assertSnapshotUUIDs(blobContainer, repositoryData); + listener.onResponse(null); + } + }); + listener.actionGet(TimeValue.timeValueMinutes(1L)); + } + + private static void assertIndexGenerations(BlobContainer repoRoot, long latestGen) throws IOException { + final long[] indexGenerations = repoRoot.listBlobsByPrefix("index-").keySet().stream() + .map(s -> s.replace("index-", "")) + .mapToLong(Long::parseLong).sorted().toArray(); + assertEquals(latestGen, indexGenerations[indexGenerations.length - 1]); + assertTrue(indexGenerations.length <= 2); + } + + private static void assertIndexUUIDs(BlobContainer repoRoot, RepositoryData repositoryData) throws IOException { + final List expectedIndexUUIDs = + repositoryData.getIndices().values().stream().map(IndexId::getId).collect(Collectors.toList()); + final BlobContainer indicesContainer = repoRoot.children().get("indices"); + final List foundIndexUUIDs; + if (indicesContainer == null) { + foundIndexUUIDs = Collections.emptyList(); + } else { + foundIndexUUIDs = indicesContainer.children().keySet().stream().filter( + s -> s.startsWith("extra") == false).collect(Collectors.toList()); + } + assertThat(foundIndexUUIDs, containsInAnyOrder(expectedIndexUUIDs.toArray(Strings.EMPTY_ARRAY))); + } + + private static void assertSnapshotUUIDs(BlobContainer repoRoot, RepositoryData repositoryData) throws IOException { + final List expectedSnapshotUUIDs = + repositoryData.getSnapshotIds().stream().map(SnapshotId::getUUID).collect(Collectors.toList()); + for (String prefix : new String[]{"snap-", "meta-"}) { + final Collection foundSnapshotUUIDs = repoRoot.listBlobs().keySet().stream().filter(p -> p.startsWith(prefix)) + .map(p -> p.replace(prefix, "").replace(".dat", "")) + .collect(Collectors.toSet()); + assertThat(foundSnapshotUUIDs, containsInAnyOrder(expectedSnapshotUUIDs.toArray(Strings.EMPTY_ARRAY))); + } + } +}