HDDS-8778. Support recursive volume delete using Ozone sh command.

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/shell/TestOzoneShellHA.java

@@ -55,6 +55,7 @@
 import org.apache.hadoop.ozone.ha.ConfUtils;
 import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
 import org.apache.hadoop.ozone.shell.s3.S3Shell;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -1416,7 +1417,8 @@ public void testVolumeListKeys()
         BucketLayout.FILE_SYSTEM_OPTIMIZED.toString());
 
     // Create OBS bucket in volx
-    String[] args = new String[]{"bucket", "create", volume1 + "/bucketobs"};
+    String[] args = new String[]{"bucket", "create", "--layout",
+        BucketLayout.OBJECT_STORE.toString(), volume1 + "/bucketobs"};
     execute(ozoneShell, args);
     out.reset();
 
@@ -1432,7 +1434,8 @@ public void testVolumeListKeys()
     out.reset();
 
     // Create Legacy bucket in volx
-    args = new String[]{"bucket", "create", volume1 + "/bucketlegacy"};
+    args = new String[]{"bucket", "create", "--layout",
+        BucketLayout.LEGACY.toString(), volume1 + "/bucketlegacy"};
     execute(ozoneShell, args);
     out.reset();
 
@@ -1461,4 +1464,83 @@ public void testVolumeListKeys()
     LambdaTestUtils.intercept(ExecutionException.class,
         "VOLUME_NOT_FOUND", () -> execute(ozoneShell, args1));
   }
+
+  @Test
+  public void testRecursiveVolumeDelete()
+      throws Exception {
+    String volume1 = "volume10";
+    String volume2 = "volume20";
+
+    // Create volume volume1
+    // Create bucket bucket1 with layout FILE_SYSTEM_OPTIMIZED
+    // Insert some keys into it
+    generateKeys(OZONE_URI_DELIMITER + volume1,
+        "/bucketfso",
+        BucketLayout.FILE_SYSTEM_OPTIMIZED.toString());
+
+    // Create another volume  volume2 with bucket and some keys into it.
+    generateKeys(OZONE_URI_DELIMITER + volume2,
+        "/bucket2",
+        BucketLayout.FILE_SYSTEM_OPTIMIZED.toString());
+
+    // Create OBS bucket in volume1
+    String[] args = new String[] {"bucket", "create", "--layout",
+        BucketLayout.OBJECT_STORE.toString(), volume1 + "/bucketobs"};
+    execute(ozoneShell, args);
+    out.reset();
+
+    // Insert few keys into OBS bucket
+    String keyName = OZONE_URI_DELIMITER + volume1 + "/bucketobs" +
+        OZONE_URI_DELIMITER + "key";
+    for (int i = 0; i < 5; i++) {
+      args = new String[] {
+          "key", "put", "o3://" + omServiceId + keyName + i,
+          testFile.getPath()};
+      execute(ozoneShell, args);
+    }
+    out.reset();
+
+    // Create Legacy bucket in volume1
+    args = new String[] {"bucket", "create", "--layout",
+        BucketLayout.LEGACY.toString(), volume1 + "/bucketlegacy"};
+    execute(ozoneShell, args);
+    out.reset();
+
+    // Insert few keys into legacy bucket
+    keyName = OZONE_URI_DELIMITER + volume1 + "/bucketlegacy" +
+        OZONE_URI_DELIMITER + "key";
+    for (int i = 0; i < 5; i++) {
+      args = new String[] {
+          "key", "put", "o3://" + omServiceId + keyName + i,
+          testFile.getPath()};
+      execute(ozoneShell, args);
+    }
+    out.reset();
+
+    // Try volume delete without recursive
+    // It should fail as volume is not empty
+    final String[] args1 = new String[] {"volume", "delete", volume1};
+    LambdaTestUtils.intercept(ExecutionException.class,
+        "VOLUME_NOT_EMPTY", () -> execute(ozoneShell, args1));
+    out.reset();
+
+    // volume1 should still exist
+    Assert.assertEquals(client.getObjectStore().getVolume(volume1)
+        .getName(), volume1);
+
+    // Delete volume1(containing OBS, FSO and Legacy buckets) recursively
+    args =
+        new String[] {"volume", "delete", volume1, "-r", "--yes",
+            "-id", omServiceId};
+
+    execute(ozoneShell, args);
+    out.reset();
+    // volume2 should still exist
+    Assert.assertEquals(client.getObjectStore().getVolume(volume2)
+        .getName(), volume2);
+
+    // volume1 should not exist
+    LambdaTestUtils.intercept(OMException.class,
+        "VOLUME_NOT_FOUND", () -> client.getObjectStore().getVolume(volume1));
+  }
 }

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/shell/volume/DeleteVolumeHandler.java

@@ -18,27 +18,247 @@
 
 package org.apache.hadoop.ozone.shell.volume;
 
+import com.google.common.base.Strings;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.OzoneClient;
+import org.apache.hadoop.ozone.client.OzoneKey;
+import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.shell.OzoneAddress;
 
+import picocli.CommandLine;
 import picocli.CommandLine.Command;
 
 import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Scanner;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY;
+import static org.apache.hadoop.hdds.scm.net.NetConstants.PATH_SEPARATOR_STR;
+import static org.apache.hadoop.ozone.OzoneConsts.OZONE_OFS_URI_SCHEME;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SERVICE_IDS_KEY;
 
 /**
  * Executes deleteVolume call for the shell.
  */
 @Command(name = "delete",
-    description = "deletes a volume if it is empty")
+    description = "deletes a volume")
 public class DeleteVolumeHandler extends VolumeHandler {
+  @CommandLine.Option(
+      names = {"-r"},
+      description = "Delete volume recursively"
+  )
+  private boolean bRecursive;
+
+  @CommandLine.Option(
+      names = {"-id", "--om-service-id"},
+      description = "Ozone Manager Service ID"
+  )
+  private String omServiceId;
+
+  @CommandLine.Option(names = {"-t", "--threads", "--thread"},
+      description = "Number of threads used to execute recursive delete")
+  private int threadNo = 10;
+
+  @CommandLine.Option(names = {"-y", "--yes"},
+      description = "Continue without interactive user confirmation")
+  private boolean yes;
+  private ExecutorService executor;
+  private List<String> bucketIdList = new ArrayList<>();
+  private AtomicInteger cleanedBucketCounter =
+      new AtomicInteger();
+  private int totalBucketCount;
+  private OzoneVolume vol;
+  private AtomicInteger numberOfBucketsCleaned = new AtomicInteger(0);
+  private volatile Throwable exception;
+  private static final int MAX_KEY_DELETE_BATCH_SIZE = 1000;
 
   @Override
   protected void execute(OzoneClient client, OzoneAddress address)
       throws IOException {
 
     String volumeName = address.getVolumeName();
-
+    try {
+      if (bRecursive) {
+        Collection<String> serviceIds = getConf().getTrimmedStringCollection(
+            OZONE_OM_SERVICE_IDS_KEY);
+        if (Strings.isNullOrEmpty(omServiceId)) {
+          if (serviceIds.size() > 1) {
+            out().printf("OmServiceID not provided, provide using " +
+                "-id <OM_SERVICE_ID>%n");
+            return;
+          } else if (serviceIds.size() == 1) {
+            // Only one OM service ID configured, we can use that
+            omServiceId = serviceIds.iterator().next();
+          }
+        }
+        if (!yes) {
+          // Ask for user confirmation
+          out().print("This command will delete volume recursively." +
+              "\nThere is no recovery option after using this command, " +
+              "and no trash for FSO buckets." +
+              "\nDelay is expected running this command." +
+              "\nEnter 'yes' to proceed': ");
+          out().flush();
+          Scanner scanner = new Scanner(new InputStreamReader(
+              System.in, StandardCharsets.UTF_8));
+          String confirmation = scanner.next().trim().toLowerCase();
+          if (!confirmation.equals("yes")) {
+            out().println("Operation cancelled.");
+            return;
+          }
+        }
+        vol = client.getObjectStore().getVolume(volumeName);
+        deleteVolumeRecursive();
+      }
+    } catch (InterruptedException e) {
+      out().printf("Exception while deleting volume recursively%n");
+      return;
+    }
     client.getObjectStore().deleteVolume(volumeName);
     out().printf("Volume %s is deleted%n", volumeName);
   }
+
+  private void deleteVolumeRecursive()
+      throws InterruptedException {
+    // Get all the buckets for given volume
+    Iterator<? extends OzoneBucket> bucketIterator =
+        vol.listBuckets(null);
+
+    while (bucketIterator.hasNext()) {
+      OzoneBucket bucket = bucketIterator.next();
+      bucketIdList.add(bucket.getName());
+      totalBucketCount++;
+    }
+    doCleanBuckets();
+  }
+
+  /**
+   * Clean OBS bucket recursively.
+   *
+   * @param  bucket OzoneBucket
+   * @return boolean
+   */
+  private boolean cleanOBSBucket(OzoneBucket bucket) {
+    ArrayList<String> keys = new ArrayList<>();
+    try {
+      if (!bucket.isLink()) {
+        Iterator<? extends OzoneKey> iterator = bucket.listKeys(null);
+        while (iterator.hasNext()) {
+          keys.add(iterator.next().getName());
+          if (MAX_KEY_DELETE_BATCH_SIZE == keys.size()) {
+            bucket.deleteKeys(keys);
+            keys.clear();
+          }
+        }
+        // delete if any remaining keys left
+        if (keys.size() > 0) {
+          bucket.deleteKeys(keys);
+        }
+      }
+      vol.deleteBucket(bucket.getName());
+      numberOfBucketsCleaned.getAndIncrement();
+      return true;
+    } catch (Exception e) {
+      LOG.error("Could not clean bucket ", e);
+      return false;
+    }
+  }
+
+  /**
+   * Clean Legacy/FSO bucket recursively.
+   *
+   * @param  bucket OzoneBucket
+   * @return boolean
+   */
+  private boolean cleanFSBucket(OzoneBucket bucket) {
+    try {
+      String hostPrefix = OZONE_OFS_URI_SCHEME + "://";
+      if (!Strings.isNullOrEmpty(omServiceId)) {
+        hostPrefix += omServiceId + PATH_SEPARATOR_STR;
+      } else {
+        hostPrefix += getConf().get(OZONE_OM_ADDRESS_KEY) +
+            PATH_SEPARATOR_STR;
+      }
+      String ofsPrefix = hostPrefix + vol.getName() + PATH_SEPARATOR_STR +
+          bucket.getName();
+      final Path path = new Path(ofsPrefix);
+      OzoneConfiguration clientConf = new OzoneConfiguration(getConf());
+      clientConf.set(FS_DEFAULT_NAME_KEY, hostPrefix);
+      FileSystem fs = FileSystem.get(clientConf);
+      if (!fs.delete(path, true)) {
+        throw new IOException("Failed to delete bucket");
+      }
+      numberOfBucketsCleaned.getAndIncrement();
+      return true;
+    } catch (Exception e) {
+      exception = e;
+      LOG.error("Could not clean bucket ", e);
+      return false;
+    }
+  }
+
+  private class BucketCleaner implements Runnable {
+    @Override
+    public void run() {
+      int i;
+      while ((i = cleanedBucketCounter.getAndIncrement()) < totalBucketCount) {
+        try {
+          OzoneBucket bucket = vol.getBucket(bucketIdList.get(i));
+          switch (bucket.getBucketLayout()) {
+          case FILE_SYSTEM_OPTIMIZED:
+          case LEGACY:
+            if (!cleanFSBucket(bucket)) {
+              throw new RuntimeException("Failed to clean bucket");
+            }
+            break;
+          case OBJECT_STORE:
+            if (!cleanOBSBucket(bucket)) {
+              throw new RuntimeException("Failed to clean bucket");
+            }
+          default:
+            throw new RuntimeException("Invalid bucket layout");
+          }
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+  }
+
+  private void doCleanBuckets() throws InterruptedException {
+    executor = Executors.newFixedThreadPool(threadNo);
+    for (int i = 0; i < threadNo; i++) {
+      executor.execute(new BucketCleaner());
+    }
+
+    try {
+      // wait until all Buckets are cleaned or exception occurred.
+      while (numberOfBucketsCleaned.get() != totalBucketCount
+          && exception == null) {
+        try {
+          Thread.sleep(100);
+        } catch (InterruptedException e) {
+          throw e;
+        }
+      }
+    } catch (InterruptedException e) {
+      LOG.error("Failed to wait until all Buckets are cleaned", e);
+      Thread.currentThread().interrupt();
+    }
+    executor.shutdown();
+    executor.awaitTermination(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
+  }
 }