Ensure every repository has an incompatible-snapshots blob (#24403)

In #22267, we introduced the notion of incompatible snapshots in a
repository, and they were stored in a root-level blob named
`incompatible-snapshots`.  If there were no incompatible snapshots in
the repository, then there was no `incompatible-snapshots` blob.

However, this causes a problem for some cloud-based repositories,
because if the blob does not exist, the cloud-based repositories may
attempt to keep retrying the read of a non-existent blob with
expontential backoff until giving up.  This causes performance issues
(and potential timeouts) on snapshot operations because getting the
`incompatible-snapshots` is part of getting the repository data (see
RepositoryData#getRepositoryData()).

This commit fixes the issue by creating an empty
`incompatible-snapshots` blob in the repository if one does not exist.

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

@@ -635,8 +635,17 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
                     repositoryData = repositoryData.incompatibleSnapshotsFromXContent(parser);
                 }
             } catch (NoSuchFileException e) {
-                logger.debug("[{}] Incompatible snapshots blob [{}] does not exist, the likely reason is that " +
-                             "there are no incompatible snapshots in the repository", metadata.name(), INCOMPATIBLE_SNAPSHOTS_BLOB);
+                if (isReadOnly()) {
+                    logger.debug("[{}] Incompatible snapshots blob [{}] does not exist, the likely " +
+                                 "reason is that there are no incompatible snapshots in the repository",
+                                 metadata.name(), INCOMPATIBLE_SNAPSHOTS_BLOB);
+                } else {
+                    // write an empty incompatible-snapshots blob - we do this so that there
+                    // is a blob present, which helps speed up some cloud-based repositories
+                    // (e.g. S3), which retry if a blob is missing with exponential backoff,
+                    // delaying the read of repository data and sometimes causing a timeout
+                    writeIncompatibleSnapshots(RepositoryData.EMPTY);
+                }
             }
             return repositoryData;
         } catch (NoSuchFileException ex) {
@@ -802,8 +811,6 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
         }
     }
 
-
-
     @Override
     public void snapshotShard(IndexShard shard, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit, IndexShardSnapshotStatus snapshotStatus) {
         SnapshotContext snapshotContext = new SnapshotContext(shard, snapshotId, indexId, snapshotStatus);

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

@@ -189,6 +189,18 @@ public class BlobStoreRepositoryTests extends ESSingleNodeTestCase {
         assertEquals(repositoryData.getIncompatibleSnapshotIds(), readData.getIncompatibleSnapshotIds());
     }
 
+    public void testIncompatibleSnapshotsBlobExists() throws Exception {
+        final BlobStoreRepository repository = setupRepo();
+        RepositoryData emptyData = RepositoryData.EMPTY;
+        repository.writeIndexGen(emptyData, emptyData.getGenId());
+        RepositoryData repoData = repository.getRepositoryData();
+        assertEquals(emptyData, repoData);
+        assertTrue(repository.blobContainer().blobExists("incompatible-snapshots"));
+        repoData = addRandomSnapshotsToRepoData(repository.getRepositoryData(), true);
+        repository.writeIndexGen(repoData, repoData.getGenId());
+        assertEquals(0, repository.getRepositoryData().getIncompatibleSnapshotIds().size());
+    }
+
     private BlobStoreRepository setupRepo() {
         final Client client = client();
         final Path location = ESIntegTestCase.randomRepoPath(node().settings());

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

@@ -429,11 +429,13 @@ public class DedicatedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTest
 
         logger.info("--> making sure that snapshot no longer exists");
         assertThrows(client().admin().cluster().prepareGetSnapshots("test-repo").setSnapshots("test-snap").execute(), SnapshotMissingException.class);
-        // Subtract three files that will remain in the repository:
+        // Subtract four 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));
+        //   (4) incompatible-snapshots
+        assertThat("not all files were deleted during snapshot cancellation",
+            numberOfFilesBeforeSnapshot, equalTo(numberOfFiles(repo) - 4));
         logger.info("--> done");
     }
 

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

@@ -989,8 +989,8 @@ public class SharedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTestCas
         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, " +
-                    "plus one because one backup index-N file should remain");
-        assertThat(numberOfFiles(repo), equalTo(numberOfFiles[0] + 1));
+                    "plus two because one backup index-N file should remain and incompatible-snapshots");
+        assertThat(numberOfFiles(repo), equalTo(numberOfFiles[0] + 2));
     }
 
     public void testDeleteSnapshotWithMissingIndexAndShardMetadata() throws Exception {