Implement Eventually Consistent Mock Repository for SnapshotResiliencyTests

server/src/main/java/org/elasticsearch/common/blobstore/BlobPath.java

@@ -76,4 +76,20 @@ public String toString() {
         }
         return sb.toString();
     }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        return paths.equals(((BlobPath) o).paths);
+    }
+
+    @Override
+    public int hashCode() {
+        return paths.hashCode();
+    }
 }

server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java

@@ -105,6 +105,7 @@
 import org.elasticsearch.cluster.service.ClusterApplierService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.MasterService;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
@@ -155,6 +156,7 @@
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.fetch.FetchPhase;
+import org.elasticsearch.snapshots.mockstore.MockEventuallyConsistentRepository;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.disruption.DisruptableMockTransport;
 import org.elasticsearch.test.disruption.NetworkDisruption;
@@ -167,10 +169,16 @@
 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;
@@ -208,9 +216,18 @@ public class SnapshotResiliencyTests extends ESTestCase {
 
     private Path tempDir;
 
+    /**
+     * Context shared by all the node's {@link Repository} instances if the eventually consistent blobstore is to be used.
+     * {@code null} if not using the eventually consistent blobstore.
+     */
+    @Nullable private MockEventuallyConsistentRepository.Context blobStoreContext;
+
     @Before
     public void createServices() {
         tempDir = createTempDir();
+        if (randomBoolean()) {
+            blobStoreContext = new MockEventuallyConsistentRepository.Context();
+        }
         deterministicTaskQueue =
             new DeterministicTaskQueue(Settings.builder().put(NODE_NAME_SETTING.getKey(), "shared").build(), random());
     }
@@ -523,10 +540,11 @@ public void run() {
     private void assertNoStaleRepositoryData() throws IOException {
         final Path repoPath = tempDir.resolve("repo").toAbsolutePath();
         final List<Path> repos;
-        try (Stream<Path> reposDir = Files.list(repoPath)) {
+        try (Stream<Path> 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);
@@ -542,8 +560,37 @@ private void assertNoStaleRepositoryData() throws IOException {
         }
     }
 
+    // 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 in this test.
+    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<Path> repoRootBlobs = Files.list(repoRoot)) {
+        try (Stream<Path> 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();
@@ -555,7 +602,7 @@ private static void assertIndexGenerations(Path repoRoot, long latestGen) throws
     private static void assertIndexUUIDs(Path repoRoot, RepositoryData repositoryData) throws IOException {
         final List<String> expectedIndexUUIDs =
             repositoryData.getIndices().values().stream().map(IndexId::getId).collect(Collectors.toList());
-        try (Stream<Path> indexRoots = Files.list(repoRoot.resolve("indices"))) {
+        try (Stream<Path> indexRoots = repoFilesByPrefix(repoRoot.resolve("indices"))) {
             final List<String> 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)));
@@ -566,7 +613,7 @@ private static void assertSnapshotUUIDs(Path repoRoot, RepositoryData repository
         final List<String> expectedSnapshotUUIDs =
             repositoryData.getSnapshotIds().stream().map(SnapshotId::getUUID).collect(Collectors.toList());
         for (String prefix : new String[]{"snap-", "meta-"}) {
-            try (Stream<Path> repoRootBlobs = Files.list(repoRoot)) {
+            try (Stream<Path> repoRootBlobs = repoFilesByPrefix(repoRoot)) {
                 final Collection<String> foundSnapshotUUIDs = repoRootBlobs.filter(p -> p.getFileName().toString().startsWith(prefix))
                     .map(p -> p.getFileName().toString().replace(prefix, "").replace(".dat", ""))
                     .collect(Collectors.toSet());
@@ -575,6 +622,20 @@ private static void assertSnapshotUUIDs(Path repoRoot, RepositoryData repository
         }
     }
 
+    /**
+     * 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<Path> repoFilesByPrefix(Path prefix) throws IOException {
+        try {
+            return Files.list(prefix);
+        } catch (FileNotFoundException | NoSuchFileException e) {
+            return Stream.empty();
+        }
+    }
+
     private void clearDisruptionsAndAwaitSync() {
         testClusterNodes.clearNetworkDisruptions();
         runUntil(() -> {
@@ -959,19 +1020,7 @@ public void onFailure(final Exception e) {
             final IndexNameExpressionResolver indexNameExpressionResolver = new IndexNameExpressionResolver();
             repositoriesService = new RepositoriesService(
                 settings, clusterService, transportService,
-                Collections.singletonMap(FsRepository.TYPE, metaData -> {
-                        final Repository repository = new FsRepository(metaData, environment, xContentRegistry(), threadPool) {
-                            @Override
-                            protected void assertSnapshotOrGenericThread() {
-                                // eliminate thread name check as we create repo in the test thread
-                            }
-                        };
-                        repository.start();
-                        return repository;
-                    }
-                ),
-                emptyMap(),
-                threadPool
+                Collections.singletonMap(FsRepository.TYPE, getRepoFactory(environment)), emptyMap(), threadPool
             );
             snapshotsService =
                 new SnapshotsService(settings, clusterService, indexNameExpressionResolver, repositoriesService, threadPool);
@@ -1152,6 +1201,28 @@ searchTransportService, new SearchPhaseController(searchService::createReduceCon
             client.initialize(actions, () -> clusterService.localNode().getId(), transportService.getRemoteClusterService());
         }
 
+        private Repository.Factory getRepoFactory(Environment environment) {
+            // Run half the tests with the eventually consistent repository
+            if (blobStoreContext == null) {
+                return metaData -> {
+                    final Repository repository = new FsRepository(metaData, environment, xContentRegistry(), threadPool) {
+                        @Override
+                        protected void assertSnapshotOrGenericThread() {
+                            // eliminate thread name check as we create repo in the test thread
+                        }
+                    };
+                    repository.start();
+                    return repository;
+                };
+            } else {
+                return metaData -> {
+                    final Repository repository = new MockEventuallyConsistentRepository(
+                        metaData, environment, xContentRegistry(), deterministicTaskQueue, blobStoreContext);
+                    repository.start();
+                    return repository;
+                };
+            }
+        }
         public void restart() {
             testClusterNodes.disconnectNode(this);
             final ClusterState oldState = this.clusterService.state();