Fix inefficient (worst case exponential) loading of snapshot repository

core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/TransportGetSnapshotsAction.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.repositories.RepositoryData;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.snapshots.SnapshotMissingException;
@@ -82,13 +83,14 @@ protected void masterOperation(final GetSnapshotsRequest request, ClusterState s
             List<SnapshotInfo> snapshotInfoBuilder = new ArrayList<>();
             final Map<String, SnapshotId> allSnapshotIds = new HashMap<>();
             final List<SnapshotId> currentSnapshotIds = new ArrayList<>();
+            final RepositoryData repositoryData  = snapshotsService.getRepositoryData(repository);
             for (SnapshotInfo snapshotInfo : snapshotsService.currentSnapshots(repository)) {
                 SnapshotId snapshotId = snapshotInfo.snapshotId();
                 allSnapshotIds.put(snapshotId.getName(), snapshotId);
                 currentSnapshotIds.add(snapshotId);
             }
             if (isCurrentSnapshotsOnly(request.snapshots()) == false) {
-                for (SnapshotId snapshotId : snapshotsService.getRepositoryData(repository).getAllSnapshotIds()) {
+                for (SnapshotId snapshotId : repositoryData.getAllSnapshotIds()) {
                     allSnapshotIds.put(snapshotId.getName(), snapshotId);
                 }
             }
@@ -119,7 +121,8 @@ protected void masterOperation(final GetSnapshotsRequest request, ClusterState s
                 }
             }
 
-            snapshotInfoBuilder.addAll(snapshotsService.snapshots(repository, new ArrayList<>(toResolve), request.ignoreUnavailable()));
+            snapshotInfoBuilder.addAll(snapshotsService.snapshots(
+                repository, new ArrayList<>(toResolve), repositoryData.getIncompatibleSnapshotIds(), request.ignoreUnavailable()));
             listener.onResponse(new GetSnapshotsResponse(snapshotInfoBuilder));
         } catch (Exception e) {
             listener.onFailure(e);

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

@@ -490,11 +490,6 @@ public MetaData getSnapshotMetaData(SnapshotInfo snapshot, List<IndexId> indices
 
     @Override
     public SnapshotInfo getSnapshotInfo(final SnapshotId snapshotId) {
-        if (getRepositoryData().getIncompatibleSnapshotIds().contains(snapshotId)) {
-            // an incompatible snapshot - cannot read its snapshot metadata file, just return
-            // a SnapshotInfo indicating its incompatible
-            return SnapshotInfo.incompatible(snapshotId);
-        }
         return getSnapshotInfoInternal(snapshotId);
     }
 

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

@@ -164,11 +164,15 @@ public SnapshotInfo snapshot(final String repositoryName, final SnapshotId snaps
      *
      * @param repositoryName repository name
      * @param snapshotIds       snapshots for which to fetch snapshot information
+     * @param incompatibleSnapshotIds   snapshots for which not to fetch snapshot information
      * @param ignoreUnavailable if true, snapshots that could not be read will only be logged with a warning,
      *                          if false, they will throw an error
      * @return list of snapshots
      */
-    public List<SnapshotInfo> snapshots(final String repositoryName, List<SnapshotId> snapshotIds, final boolean ignoreUnavailable) {
+    public List<SnapshotInfo> snapshots(final String repositoryName,
+                                        final List<SnapshotId> snapshotIds,
+                                        final List<SnapshotId> incompatibleSnapshotIds,
+                                        final boolean ignoreUnavailable) {
         final Set<SnapshotInfo> snapshotSet = new HashSet<>();
         final Set<SnapshotId> snapshotIdsToIterate = new HashSet<>(snapshotIds);
         // first, look at the snapshots in progress
@@ -182,7 +186,13 @@ public List<SnapshotInfo> snapshots(final String repositoryName, List<SnapshotId
         final Repository repository = repositoriesService.repository(repositoryName);
         for (SnapshotId snapshotId : snapshotIdsToIterate) {
             try {
-                snapshotSet.add(repository.getSnapshotInfo(snapshotId));
+                if (incompatibleSnapshotIds.contains(snapshotId)) {
+                    // an incompatible snapshot - cannot read its snapshot metadata file, just return
+                    // a SnapshotInfo indicating its incompatible
+                    snapshotSet.add(SnapshotInfo.incompatible(snapshotId));
+                } else {
+                    snapshotSet.add(repository.getSnapshotInfo(snapshotId));
+                }
             } catch (Exception ex) {
                 if (ignoreUnavailable) {
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("failed to get snapshot [{}]", snapshotId), ex);
@@ -196,6 +206,7 @@ public List<SnapshotInfo> snapshots(final String repositoryName, List<SnapshotId
         return Collections.unmodifiableList(snapshotList);
     }
 
+
     /**
      * Returns a list of currently running snapshots from repository sorted by snapshot creation date
      *