Recursively Delete Unreferenced Index Directories

modules/repository-url/src/main/java/org/elasticsearch/common/blobstore/url/URLBlobContainer.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.common.blobstore.url;
 
 import org.elasticsearch.common.SuppressForbidden;
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
@@ -74,6 +75,11 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
         throw new UnsupportedOperationException("URL repository doesn't support this operation");
     }
 
+    @Override
+    public Map<String, BlobContainer> children() throws IOException {
+        throw new UnsupportedOperationException("URL repository doesn't support this operation");
+    }
+
     /**
      * This operation is not supported by URLBlobContainer
      */

plugins/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobContainer.java

@@ -24,6 +24,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
@@ -135,6 +136,15 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
         return listBlobsByPrefix(null);
     }
 
+    @Override
+    public Map<String, BlobContainer> children() throws IOException {
+        try {
+            return blobStore.children(path());
+        } catch (URISyntaxException | StorageException e) {
+            throw new IOException("Failed to list children.", e);
+        }
+    }
+
     protected String buildKey(String blobName) {
         return keyPath + (blobName == null ? "" : blobName);
     }

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

@@ -21,7 +21,6 @@
 
 import com.microsoft.azure.storage.LocationMode;
 import com.microsoft.azure.storage.StorageException;
-
 import org.elasticsearch.cluster.metadata.RepositoryMetaData;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
@@ -33,6 +32,8 @@
 import java.io.InputStream;
 import java.net.URISyntaxException;
 import java.nio.file.FileAlreadyExistsException;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 
 import static java.util.Collections.emptyMap;
@@ -103,6 +104,13 @@ public Map<String, BlobMetaData> listBlobsByPrefix(String keyPath, String prefix
         return service.listBlobsByPrefix(clientName, container, keyPath, prefix);
     }
 
+    public Map<String, BlobContainer> children(BlobPath path) throws URISyntaxException, StorageException {
+        return Collections.unmodifiableMap(service.children(clientName, container, path).stream().collect(
+            HashMap::new,
+            (existing, name) -> existing.put(name, new AzureBlobContainer(path.add(name), this)),
+            HashMap::putAll));
+    }
+
     public void writeBlob(String blobName, InputStream inputStream, long blobSize, boolean failIfAlreadyExists)
         throws URISyntaxException, StorageException, FileAlreadyExistsException {
         service.writeBlob(this.clientName, container, blobName, inputStream, blobSize, failIfAlreadyExists);

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

@@ -38,6 +38,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.common.blobstore.BlobMetaData;
+import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.support.PlainBlobMetaData;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
@@ -54,7 +55,9 @@
 import java.security.InvalidKeyException;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Supplier;
 
 import static java.util.Collections.emptyMap;
@@ -241,6 +244,30 @@ public Map<String, BlobMetaData> listBlobsByPrefix(String account, String contai
         return Map.copyOf(blobsBuilder);
     }
 
+    public Set<String> children(String account, String container, BlobPath path) throws URISyntaxException, StorageException {
+        final var blobsBuilder = new HashSet<String>();
+        final EnumSet<BlobListingDetails> enumBlobListingDetails = EnumSet.of(BlobListingDetails.METADATA);
+        final Tuple<CloudBlobClient, Supplier<OperationContext>> client = client(account);
+        final CloudBlobContainer blobContainer = client.v1().getContainerReference(container);
+        final String keyPath = path.buildAsString();
+        SocketAccess.doPrivilegedVoidException(() -> {
+            for (final ListBlobItem blobItem : blobContainer.listBlobs(keyPath, false,
+                enumBlobListingDetails, null, client.v2().get())) {
+
+                final URI uri = blobItem.getUri();
+                logger.trace(() -> new ParameterizedMessage("blob url [{}]", uri));
+                // uri.getPath is of the form /container/keyPath.* and we want to strip off the /container/
+                // this requires 1 + container.length() + 1, with each 1 corresponding to one of the /
+                final String blobPath = uri.getPath().substring(1 + container.length() + 1);
+                final BlobProperties properties = ((CloudBlockBlob) blobItem).getProperties();
+                final String name = blobPath.substring(keyPath.length());
+                logger.trace(() -> new ParameterizedMessage("blob url [{}], name [{}], size [{}]", uri, name, properties.getLength()));
+                blobsBuilder.add(name);
+            }
+        });
+        return Set.copyOf(blobsBuilder);
+    }
+
     public void writeBlob(String account, String container, String blobName, InputStream inputStream, long blobSize,
                           boolean failIfAlreadyExists)
         throws URISyntaxException, StorageException, FileAlreadyExistsException {

plugins/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobContainer.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.repositories.gcs;
 
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStoreException;
@@ -55,6 +56,11 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
         return blobStore.listBlobs(path);
     }
 
+    @Override
+    public Map<String, BlobContainer> children() throws IOException {
+        return blobStore.listChildren(path());
+    }
+
     @Override
     public Map<String, BlobMetaData> listBlobsByPrefix(String prefix) throws IOException {
         return blobStore.listBlobsByPrefix(path, prefix);

plugins/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStore.java

@@ -30,7 +30,6 @@
 import com.google.cloud.storage.Storage.BlobListOption;
 import com.google.cloud.storage.StorageBatch;
 import com.google.cloud.storage.StorageException;
-
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
@@ -50,11 +49,11 @@
 import java.nio.channels.WritableByteChannel;
 import java.nio.file.FileAlreadyExistsException;
 import java.nio.file.NoSuchFileException;
-import java.util.Map;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
@@ -132,13 +131,26 @@ Map<String, BlobMetaData> listBlobs(String path) throws IOException {
     Map<String, BlobMetaData> listBlobsByPrefix(String path, String prefix) throws IOException {
         final String pathPrefix = buildKey(path, prefix);
         final MapBuilder<String, BlobMetaData> mapBuilder = MapBuilder.newMapBuilder();
-        SocketAccess.doPrivilegedVoidIOException(() -> {
-            client().get(bucketName).list(BlobListOption.prefix(pathPrefix)).iterateAll().forEach(blob -> {
+        SocketAccess.doPrivilegedVoidIOException(
+            () -> client().get(bucketName).list(BlobListOption.prefix(pathPrefix)).iterateAll().forEach(blob -> {
                 assert blob.getName().startsWith(path);
                 final String suffixName = blob.getName().substring(path.length());
                 mapBuilder.put(suffixName, new PlainBlobMetaData(suffixName, blob.getSize()));
-            });
-        });
+            }));
+        return mapBuilder.immutableMap();
+    }
+
+    Map<String, BlobContainer> listChildren(BlobPath path) throws IOException {
+        final String pathStr = path.buildAsString();
+        final MapBuilder<String, BlobContainer> mapBuilder = MapBuilder.newMapBuilder();
+        SocketAccess.doPrivilegedVoidIOException
+            (() -> client().get(bucketName).list(BlobListOption.currentDirectory()).iterateAll().forEach(blob -> {
+                if (blob.isDirectory()) {
+                    assert blob.getName().startsWith(pathStr);
+                    final String suffixName = blob.getName().substring(pathStr.length());
+                    mapBuilder.put(suffixName, new GoogleCloudStorageBlobContainer(path.add(suffixName), this));
+                }
+            }));
         return mapBuilder.immutableMap();
     }
 

plugins/repository-hdfs/src/main/java/org/elasticsearch/repositories/hdfs/HdfsBlobContainer.java

@@ -25,6 +25,7 @@
 import org.apache.hadoop.fs.Options.CreateOpts;
 import org.apache.hadoop.fs.Path;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.fs.FsBlobContainer;
@@ -137,9 +138,9 @@ public void writeBlobAtomic(String blobName, InputStream inputStream, long blobS
 
     @Override
     public Map<String, BlobMetaData> listBlobsByPrefix(@Nullable final String prefix) throws IOException {
-        FileStatus[] files = store.execute(fileContext -> (fileContext.util().listStatus(path,
-            path -> prefix == null || path.getName().startsWith(prefix))));
-        Map<String, BlobMetaData> map = new LinkedHashMap<String, BlobMetaData>();
+        FileStatus[] files = store.execute(fileContext -> fileContext.util().listStatus(path,
+            path -> prefix == null || path.getName().startsWith(prefix)));
+        Map<String, BlobMetaData> map = new LinkedHashMap<>();
         for (FileStatus file : files) {
             map.put(file.getPath().getName(), new PlainBlobMetaData(file.getPath().getName(), file.getLen()));
         }
@@ -151,6 +152,20 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
         return listBlobsByPrefix(null);
     }
 
+    @Override
+    public Map<String, BlobContainer> children() throws IOException {
+        FileStatus[] files = store.execute(fileContext -> fileContext.util().listStatus(path));
+        Map<String, BlobContainer> map = new LinkedHashMap<>();
+        for (FileStatus file : files) {
+            if (file.isDirectory()) {
+                final String name = file.getPath().getName();
+                map.put(file.getPath().getName(),
+                    new HdfsBlobContainer(path().add(name), store, new Path(path, name), bufferSize, securityContext));
+            }
+        }
+        return Collections.unmodifiableMap(map);
+    }
+
     /**
      * Exists to wrap underlying InputStream methods that might make socket connections in
      * doPrivileged blocks. This is due to the way that hdfs client libraries might open

plugins/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobContainer.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStoreException;
@@ -231,6 +232,36 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
         return listBlobsByPrefix(null);
     }
 
+    @Override
+    public Map<String, BlobContainer> children() throws IOException {
+        try (AmazonS3Reference clientReference = blobStore.clientReference()) {
+            ObjectListing prevListing = null;
+            final var entries = new ArrayList<Map.Entry<String, BlobContainer>>();
+            while (true) {
+                ObjectListing list;
+                if (prevListing != null) {
+                    final ObjectListing finalPrevListing = prevListing;
+                    list = SocketAccess.doPrivileged(() -> clientReference.client().listNextBatchOfObjects(finalPrevListing));
+                } else {
+                    list = SocketAccess.doPrivileged(() -> clientReference.client().listObjects(blobStore.bucket(), keyPath));
+                }
+                for (final String summary : list.getCommonPrefixes()) {
+                    final String name = summary.substring(keyPath.length());
+                    final BlobPath path = path().add(name);
+                    entries.add(entry(name, blobStore.blobContainer(path)));
+                }
+                if (list.isTruncated()) {
+                    prevListing = list;
+                } else {
+                    break;
+                }
+            }
+            return Maps.ofEntries(entries);
+        } catch (final AmazonClientException e) {
+            throw new IOException("Exception when listing children", e);
+        }
+    }
+
     private String buildKey(String blobName) {
         return keyPath + blobName;
     }

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

@@ -158,6 +158,16 @@ default void deleteBlobIgnoringIfNotExists(String blobName) throws IOException {
      */
     Map<String, BlobMetaData> listBlobs() throws IOException;
 
+    /**
+     * Lists all child containers under this container. A child container is defined as a container whose {@link #path()} method returns
+     * a path that has the return  of this container's {@link #path()} method as a prefix with a single path component added to it.
+     * The path component by which a container and its child differ is referred to as the name of the child container below.
+     *
+     * @return Map of name of the child container to child container
+     * @throws IOException on failure to list child containers
+     */
+    Map<String, BlobContainer> children() throws IOException;
+
     /**
      * Lists all blobs in the container that match the specified prefix.
      *

server/src/main/java/org/elasticsearch/common/blobstore/fs/FsBlobContainer.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.common.blobstore.fs;
 
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.support.AbstractBlobContainer;
@@ -73,6 +74,22 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
         return listBlobsByPrefix(null);
     }
 
+    @Override
+    public Map<String, BlobContainer> children() throws IOException {
+        // If we get duplicate files we should just take the last entry
+        Map<String, BlobContainer> builder = new HashMap<>();
+        try (DirectoryStream<Path> stream = Files.newDirectoryStream(path)) {
+            for (Path file : stream) {
+                final BasicFileAttributes attrs = Files.readAttributes(file, BasicFileAttributes.class);
+                if (attrs.isDirectory()) {
+                    final String name = file.getFileName().toString();
+                    builder.put(name, new FsBlobContainer(blobStore, path().add(name), file));
+                }
+            }
+        }
+        return unmodifiableMap(builder);
+    }
+
     @Override
     public Map<String, BlobMetaData> listBlobsByPrefix(String blobNamePrefix) throws IOException {
         // If we get duplicate files we should just take the last entry