Adding repository index generational files

Before, a repository would maintain an index file (named 'index') per
repository, that contained the current snapshots in the repository.
This file was not atomically written, so repositories had to depend on
listing the blobs in the repository to determine what the current
snapshots are, and only rely on the index file if the repository does
not support the listBlobs operation.  This could cause an incorrect view
of the current snapshots in the repository if any prior snapshot delete
operations failed to delete snapshot metadata files.

This commit introduces the atomic writing of the index file, and because
atomic writes are not guaranteed if the file already exists, we write to
a generational index file (index-N, where N is the current generation).
We also maintain an index-latest file that contains the current
generation, for those repositories that cannot list blobs.

Closes #19002
Relates #18156

Author: Ali Beyad

core/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -366,7 +366,6 @@ public void apply(Settings value, Settings current, Settings previous) {
                     Node.NODE_INGEST_SETTING,
                     Node.NODE_ATTRIBUTES,
                     URLRepository.ALLOWED_URLS_SETTING,
-                    URLRepository.REPOSITORIES_LIST_DIRECTORIES_SETTING,
                     URLRepository.REPOSITORIES_URL_SETTING,
                     URLRepository.SUPPORTED_PROTOCOLS_SETTING,
                     TransportMasterNodeReadAction.FORCE_LOCAL_SETTING,

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

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.repositories.uri;
 
-import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStore;
 import org.elasticsearch.common.blobstore.url.URLBlobStore;
@@ -42,7 +41,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Function;
-import java.util.function.Predicate;
 
 /**
  * Read-only URL-based implementation of the BlobStoreRepository
@@ -69,11 +67,6 @@ public class URLRepository extends BlobStoreRepository {
         new Setting<>("repositories.url.url", (s) -> s.get("repositories.uri.url", "http:"), URLRepository::parseURL,
             Property.NodeScope);
 
-    public static final Setting<Boolean> LIST_DIRECTORIES_SETTING =
-        Setting.boolSetting("list_directories", true, Property.NodeScope);
-    public static final Setting<Boolean> REPOSITORIES_LIST_DIRECTORIES_SETTING =
-        Setting.boolSetting("repositories.uri.list_directories", true, Property.NodeScope);
-
     private final List<String> supportedProtocols;
 
     private final URIPattern[] urlWhiteList;
@@ -84,8 +77,6 @@ public class URLRepository extends BlobStoreRepository {
 
     private final BlobPath basePath;
 
-    private boolean listDirectories;
-
     /**
      * Constructs new read-only URL-based repository
      *
@@ -103,7 +94,6 @@ public URLRepository(RepositoryName name, RepositorySettings repositorySettings,
         supportedProtocols = SUPPORTED_PROTOCOLS_SETTING.get(settings);
         urlWhiteList = ALLOWED_URLS_SETTING.get(settings).toArray(new URIPattern[]{});
         this.environment = environment;
-        listDirectories = LIST_DIRECTORIES_SETTING.exists(repositorySettings.settings()) ? LIST_DIRECTORIES_SETTING.get(repositorySettings.settings()) : REPOSITORIES_LIST_DIRECTORIES_SETTING.get(settings);
 
         URL url = URL_SETTING.exists(repositorySettings.settings()) ? URL_SETTING.get(repositorySettings.settings()) : REPOSITORIES_URL_SETTING.get(settings);
         URL normalizedURL = checkURL(url);
@@ -124,19 +114,6 @@ protected BlobPath basePath() {
         return basePath;
     }
 
-    @Override
-    public List<SnapshotId> snapshots() {
-        if (listDirectories) {
-            return super.snapshots();
-        } else {
-            try {
-                return readSnapshotList();
-            } catch (IOException ex) {
-                throw new RepositoryException(repositoryName, "failed to get snapshot list in repository", ex);
-            }
-        }
-    }
-
     /**
      * Makes sure that the url is white listed or if it points to the local file system it matches one on of the root path in path.repo
      */

core/src/main/java/org/elasticsearch/repositories/uri/URLRepository.java

@@ -24,7 +24,6 @@
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.OutputStreamStreamOutput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -46,7 +45,6 @@
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.repositories.blobstore.BlobStoreRepository.blobId;
-import static org.elasticsearch.repositories.blobstore.BlobStoreRepository.parseNameUUIDFromBlobName;
 import static org.hamcrest.Matchers.equalTo;
 
 /**
@@ -108,53 +106,57 @@ public void testRetrieveSnapshots() throws Exception {
         assertThat(snapshotIds, equalTo(originalSnapshots));
     }
 
-    public void testSnapshotIndexFile() throws Exception {
-        final Client client = client();
-        final Path location = ESIntegTestCase.randomRepoPath(node().settings());
-        final String repositoryName = "test-repo";
-
-        PutRepositoryResponse putRepositoryResponse =
-            client.admin().cluster().preparePutRepository(repositoryName)
-                                    .setType("fs")
-                                    .setSettings(Settings.builder().put(node().settings()).put("location", location))
-                                    .get();
-        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
-
-        final RepositoriesService repositoriesService = getInstanceFromNode(RepositoriesService.class);
-        @SuppressWarnings("unchecked") final BlobStoreRepository repository =
-            (BlobStoreRepository) repositoriesService.repository(repositoryName);
+    public void testReadAndWriteSnapshotsThroughIndexFile() throws Exception {
+        final BlobStoreRepository repository = setupRepo();
 
         // write to and read from a snapshot file with no entries
-        repository.writeSnapshotList(Collections.emptyList());
-        List<SnapshotId> readSnapshotIds = repository.readSnapshotList();
-        assertThat(readSnapshotIds.size(), equalTo(0));
+        assertThat(repository.snapshots().size(), equalTo(0));
+        repository.writeSnapshotsToIndexGen(Collections.emptyList());
+        assertThat(repository.snapshots().size(), equalTo(0));
 
         // write to and read from a snapshot file with a random number of entries
         final int numSnapshots = randomIntBetween(1, 1000);
         final List<SnapshotId> snapshotIds = new ArrayList<>(numSnapshots);
         for (int i = 0; i < numSnapshots; i++) {
             snapshotIds.add(new SnapshotId(randomAsciiOfLength(8), UUIDs.randomBase64UUID()));
         }
-        repository.writeSnapshotList(snapshotIds);
-        readSnapshotIds = repository.readSnapshotList();
-        assertThat(readSnapshotIds, equalTo(snapshotIds));
+        repository.writeSnapshotsToIndexGen(snapshotIds);
+        assertThat(repository.snapshots(), equalTo(snapshotIds));
     }
 
-    public void testOldIndexFileFormat() throws Exception {
-        final Client client = client();
-        final Path location = ESIntegTestCase.randomRepoPath(node().settings());
-        final String repositoryName = "test-repo";
+    public void testIndexGenerationalFiles() throws Exception {
+        final BlobStoreRepository repository = setupRepo();
 
-        PutRepositoryResponse putRepositoryResponse =
-            client.admin().cluster().preparePutRepository(repositoryName)
-                .setType("fs")
-                .setSettings(Settings.builder().put(node().settings()).put("location", location))
-                .get();
-        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+        // write to index generational file
+        final int numSnapshots = randomIntBetween(1, 1000);
+        final List<SnapshotId> snapshotIds = new ArrayList<>(numSnapshots);
+        for (int i = 0; i < numSnapshots; i++) {
+            snapshotIds.add(new SnapshotId(randomAsciiOfLength(8), UUIDs.randomBase64UUID()));
+        }
+        repository.writeSnapshotsToIndexGen(snapshotIds);
+        assertThat(Sets.newHashSet(repository.readSnapshotsFromIndex()), equalTo(Sets.newHashSet(snapshotIds)));
+        assertThat(repository.latestIndexBlobId(), equalTo(0L));
+        assertThat(repository.readSnapshotIndexLatestBlob(), equalTo(0L));
 
-        final RepositoriesService repositoriesService = getInstanceFromNode(RepositoriesService.class);
-        @SuppressWarnings("unchecked") final BlobStoreRepository repository =
-            (BlobStoreRepository) repositoriesService.repository(repositoryName);
+        // adding more and writing to a new index generational file
+        for (int i = 0; i < 10; i++) {
+            snapshotIds.add(new SnapshotId(randomAsciiOfLength(8), UUIDs.randomBase64UUID()));
+        }
+        repository.writeSnapshotsToIndexGen(snapshotIds);
+        assertThat(Sets.newHashSet(repository.readSnapshotsFromIndex()), equalTo(Sets.newHashSet(snapshotIds)));
+        assertThat(repository.latestIndexBlobId(), equalTo(1L));
+        assertThat(repository.readSnapshotIndexLatestBlob(), equalTo(1L));
+
+        // removing a snapshot adn writing to a new index generational file
+        snapshotIds.remove(0);
+        repository.writeSnapshotsToIndexGen(snapshotIds);
+        assertThat(Sets.newHashSet(repository.readSnapshotsFromIndex()), equalTo(Sets.newHashSet(snapshotIds)));
+        assertThat(repository.latestIndexBlobId(), equalTo(2L));
+        assertThat(repository.readSnapshotIndexLatestBlob(), equalTo(2L));
+    }
+
+    public void testOldIndexFileFormat() throws Exception {
+        final BlobStoreRepository repository = setupRepo();
 
         // write old index file format
         final int numOldSnapshots = randomIntBetween(1, 50);
@@ -163,36 +165,15 @@ public void testOldIndexFileFormat() throws Exception {
             snapshotIds.add(new SnapshotId(randomAsciiOfLength(8), SnapshotId.UNASSIGNED_UUID));
         }
         writeOldFormat(repository, snapshotIds.stream().map(SnapshotId::getName).collect(Collectors.toList()));
-        List<SnapshotId> readSnapshotIds = repository.readSnapshotList();
-        assertThat(Sets.newHashSet(readSnapshotIds), equalTo(Sets.newHashSet(snapshotIds)));
+        assertThat(Sets.newHashSet(repository.snapshots()), equalTo(Sets.newHashSet(snapshotIds)));
 
         // write to and read from a snapshot file with a random number of new entries added
         final int numSnapshots = randomIntBetween(1, 1000);
         for (int i = 0; i < numSnapshots; i++) {
             snapshotIds.add(new SnapshotId(randomAsciiOfLength(8), UUIDs.randomBase64UUID()));
         }
-        repository.writeSnapshotList(snapshotIds);
-        readSnapshotIds = repository.readSnapshotList();
-        assertThat(Sets.newHashSet(readSnapshotIds), equalTo(Sets.newHashSet(snapshotIds)));
-    }
-
-    public void testParseUUIDFromBlobName() {
-        String blobStr = "abc123";
-        Tuple<String, String> pair = parseNameUUIDFromBlobName(blobStr);
-        assertThat(pair.v1(), equalTo(blobStr)); // snapshot name
-        assertThat(pair.v2(), equalTo(SnapshotId.UNASSIGNED_UUID)); // snapshot uuid
-        blobStr = "abcefghijklmnopqrstuvwxyz";
-        pair = parseNameUUIDFromBlobName(blobStr);
-        assertThat(pair.v1(), equalTo(blobStr));
-        assertThat(pair.v2(), equalTo(SnapshotId.UNASSIGNED_UUID));
-        blobStr = "abc123-xyz"; // not enough characters after '-' to have a uuid
-        pair = parseNameUUIDFromBlobName(blobStr);
-        assertThat(pair.v1(), equalTo(blobStr));
-        assertThat(pair.v2(), equalTo(SnapshotId.UNASSIGNED_UUID));
-        blobStr = "abc123-a1b2c3d4e5f6g7h8i9j0k1";
-        pair = parseNameUUIDFromBlobName(blobStr);
-        assertThat(pair.v1(), equalTo("abc123"));
-        assertThat(pair.v2(), equalTo("a1b2c3d4e5f6g7h8i9j0k1"));
+        repository.writeSnapshotsToIndexGen(snapshotIds);
+        assertThat(Sets.newHashSet(repository.snapshots()), equalTo(Sets.newHashSet(snapshotIds)));
     }
 
     public void testBlobId() {
@@ -208,6 +189,24 @@ public void testBlobId() {
         assertThat(blobId(snapshotId), equalTo("abc-123-" + uuid)); // snapshot name + '-' + uuid
     }
 
+    private BlobStoreRepository setupRepo() {
+        final Client client = client();
+        final Path location = ESIntegTestCase.randomRepoPath(node().settings());
+        final String repositoryName = "test-repo";
+
+        PutRepositoryResponse putRepositoryResponse =
+            client.admin().cluster().preparePutRepository(repositoryName)
+                                    .setType("fs")
+                                    .setSettings(Settings.builder().put(node().settings()).put("location", location))
+                                    .get();
+        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+
+        final RepositoriesService repositoriesService = getInstanceFromNode(RepositoriesService.class);
+        @SuppressWarnings("unchecked") final BlobStoreRepository repository =
+            (BlobStoreRepository) repositoriesService.repository(repositoryName);
+        return repository;
+    }
+
     private void writeOldFormat(final BlobStoreRepository repository, final List<String> snapshotNames) throws Exception {
         final BytesReference bRef;
         try (BytesStreamOutput bStream = new BytesStreamOutput()) {

core/src/test/java/org/elasticsearch/repositories/blobstore/BlobStoreRepositoryTests.java

@@ -284,8 +284,8 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
         countDownLatch.await();
     }
 
-    private static interface ClusterStateUpdater {
-        public ClusterState execute(ClusterState currentState) throws Exception;
+    private interface ClusterStateUpdater {
+        ClusterState execute(ClusterState currentState) throws Exception;
     }
 
     public void testSnapshotDuringNodeShutdown() throws Exception {
@@ -392,8 +392,11 @@ public void testSnapshotWithStuckNode() throws Exception {
 
         logger.info("--> making sure that snapshot no longer exists");
         assertThrows(client().admin().cluster().prepareGetSnapshots("test-repo").setSnapshots("test-snap").execute(), SnapshotMissingException.class);
-        // Subtract index file from the count
-        assertThat("not all files were deleted during snapshot cancellation", numberOfFilesBeforeSnapshot, equalTo(numberOfFiles(repo) - 1));
+        // Subtract three files that will remain in the repository:
+        //   (1) index-1
+        //   (2) index-0 (because we keep the previous version) and
+        //   (3) index-latest
+        assertThat("not all files were deleted during snapshot cancellation", numberOfFilesBeforeSnapshot, equalTo(numberOfFiles(repo) - 3));
         logger.info("--> done");
     }
 

core/src/test/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java

@@ -818,8 +818,9 @@ public void testDeleteSnapshot() throws Exception {
 
         logger.info("--> delete the last snapshot");
         client.admin().cluster().prepareDeleteSnapshot("test-repo", lastSnapshot).get();
-        logger.info("--> make sure that number of files is back to what it was when the first snapshot was made");
-        assertThat(numberOfFiles(repo), equalTo(numberOfFiles[0]));
+        logger.info("--> make sure that number of files is back to what it was when the first snapshot was made, " +
+                    "plus one because one backup index-N file should remain");
+        assertThat(numberOfFiles(repo), equalTo(numberOfFiles[0] + 1));
     }
 
     public void testDeleteSnapshotWithMissingIndexAndShardMetadata() throws Exception {

core/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.repositories.RepositoryName;
 import org.elasticsearch.repositories.RepositorySettings;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.URISyntaxException;
@@ -127,7 +128,7 @@ public void deleteBlob(String container, String blob) throws URISyntaxException,
         this.client.deleteBlob(this.accountName, this.locMode, container, blob);
     }
 
-    public InputStream getInputStream(String container, String blob) throws URISyntaxException, StorageException
+    public InputStream getInputStream(String container, String blob) throws URISyntaxException, StorageException, IOException
     {
         return this.client.getInputStream(this.accountName, this.locMode, container, blob);
     }

plugins/repository-azure/src/main/java/org/elasticsearch/cloud/azure/blobstore/AzureBlobStore.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.URISyntaxException;
@@ -75,7 +76,7 @@ final class Storage {
     void deleteBlob(String account, LocationMode mode, String container, String blob) throws URISyntaxException, StorageException;
 
     InputStream getInputStream(String account, LocationMode mode, String container, String blob)
-        throws URISyntaxException, StorageException;
+        throws URISyntaxException, StorageException, IOException;
 
     OutputStream getOutputStream(String account, LocationMode mode, String container, String blob)
         throws URISyntaxException, StorageException;

plugins/repository-azure/src/main/java/org/elasticsearch/cloud/azure/storage/AzureStorageService.java

@@ -31,6 +31,8 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.URISyntaxException;
@@ -79,7 +81,10 @@ public void deleteBlob(String account, LocationMode mode, String container, Stri
     }
 
     @Override
-    public InputStream getInputStream(String account, LocationMode mode, String container, String blob) {
+    public InputStream getInputStream(String account, LocationMode mode, String container, String blob) throws IOException {
+        if (!blobExists(account, mode, container, blob)) {
+            throw new FileNotFoundException("missing blob [" + blob + "]");
+        }
         return new ByteArrayInputStream(blobs.get(blob).toByteArray());
     }