HDDS-5891. OFS mkdir -p does not work as expected for bucket creation when volume exists due to volume create ACL check

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestRootedOzoneFileSystem.java

@@ -38,12 +38,12 @@
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.TestDataUtil;
+import org.apache.hadoop.ozone.client.BucketArgs;
 import org.apache.hadoop.ozone.client.ObjectStore;
 import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.OzoneKeyDetails;
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.VolumeArgs;
-import org.apache.hadoop.ozone.client.BucketArgs;
 import org.apache.hadoop.ozone.client.protocol.ClientProtocol;
 import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.om.OMMetrics;
@@ -59,6 +59,7 @@
 import org.apache.ozone.test.LambdaTestUtils;
 import org.junit.AfterClass;
 import org.junit.Assert;
+import org.junit.Assume;
 import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Rule;
@@ -71,6 +72,7 @@
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -161,6 +163,12 @@ public static Path getBucketPath() {
   private static String rootPath;
   private static BucketLayout bucketLayout;
 
+  private static final String USER1 = "regularuser1";
+  private static final UserGroupInformation UGI_USER1 = UserGroupInformation
+      .createUserForTesting(USER1,  new String[] {"usergroup"});
+  // Non-privileged OFS instance
+  private static RootedOzoneFileSystem userOfs;
+
   @BeforeClass
   public static void init() throws Exception {
     conf = new OzoneConfiguration();
@@ -178,12 +186,17 @@ public static void init() throws Exception {
           enabledFileSystemPaths);
     }
     conf.setBoolean(OzoneConfigKeys.OZONE_ACL_ENABLED, enableAcl);
+    // Set ACL authorizer class to OzoneNativeAuthorizer. The default
+    // OzoneAccessAuthorizer always returns true for all ACL checks which
+    // doesn't work for the test.
+    conf.set(OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS,
+        OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS_NATIVE);
     cluster = MiniOzoneCluster.newBuilder(conf)
         .setNumDatanodes(3)
         .build();
     cluster.waitForClusterToBeReady();
     objectStore = cluster.getClient().getObjectStore();
-    
+
     // create a volume and a bucket to be used by RootedOzoneFileSystem (OFS)
     OzoneBucket bucket =
         TestDataUtil.createVolumeAndBucket(cluster, bucketLayout);
@@ -206,6 +219,10 @@ public static void init() throws Exception {
     trash = new Trash(conf);
     ofs = (RootedOzoneFileSystem) fs;
     adapter = (BasicRootedOzoneClientAdapterImpl) ofs.getAdapter();
+
+    userOfs = UGI_USER1.doAs(
+        (PrivilegedExceptionAction<RootedOzoneFileSystem>)()
+            -> (RootedOzoneFileSystem) FileSystem.get(conf));
   }
 
   @AfterClass
@@ -870,7 +887,7 @@ public void testListStatusRootAndVolumeContinuation() throws IOException {
    * OFS: Test /tmp mount behavior.
    */
   @Test
-  public void testTempMount() throws Exception {
+  public void testTempMount() throws IOException {
     // Prep
     // Use ClientProtocol to pass in volume ACL, ObjectStore won't do it
     ClientProtocol proxy = objectStore.getClientProxy();
@@ -1513,4 +1530,62 @@ public void testRenameDestinationParentDoesntExist() throws Exception {
     }
   }
 
+  @Test
+  public void testNonPrivilegedUserMkdirCreateBucket() throws IOException {
+    // This test is only meaningful when ACL is enabled
+    Assume.assumeTrue("ACL is not enabled. Skipping this test as it requires " +
+            "ACL to be enabled to be meaningful.", enableAcl);
+
+    // This unit test does the correct check. However, the parameterized
+    // test is not initialized correctly since HDDS-4998 (or HDDS-4040).
+
+    // The cluster is started with enableAcl = false, but across different set
+    // of test parameters, the OM is not restarted. So OM is actually stuck with
+    // enableAcl = false. That's why ACL checks are skipped in the tests.
+    // OM should be restarted when enableAcl is changed.
+
+    // For now, in order to properly run this specific test, on line 147,
+    // explicitly assign:
+    //     private static boolean enableAcl = true;
+    // When unset, it defaults to false due to Java primitive defaults.
+
+    // TODO: Fix the parameterized test to properly initialize the cluster.
+    //  When the parameterized test is initialized correctly, the line
+    //  below can be uncommented.
+//    Assert.assertTrue(cluster.getOzoneManager().getAclsEnabled());
+
+    final String volume = "volume-for-test-get-bucket";
+    // Create a volume as admin
+    // Create volume "tmp" with world access. allow non-admin to create buckets
+    ClientProtocol proxy = objectStore.getClientProxy();
+
+    // Get default acl rights for user
+    OzoneAclConfig aclConfig = conf.getObject(OzoneAclConfig.class);
+    ACLType userRights = aclConfig.getUserDefaultRights();
+    // Construct ACL for world access
+    OzoneAcl aclWorldAccess = new OzoneAcl(ACLIdentityType.WORLD, "",
+        userRights, ACCESS);
+    // Construct VolumeArgs, set ACL to world access
+    VolumeArgs volumeArgs = new VolumeArgs.Builder()
+        .setAcls(Collections.singletonList(aclWorldAccess))
+        .build();
+    proxy.createVolume(volume, volumeArgs);
+
+    // Create a bucket as non-admin, should succeed
+    final String bucket = "test-bucket-1";
+    try {
+      final Path myBucketPath = new Path(volume, bucket);
+      // Have to prepend the root to bucket path here.
+      // Otherwise, FS will automatically prepend user home directory path
+      // which is not we want here.
+      Assert.assertTrue(userOfs.mkdirs(new Path("/", myBucketPath)));
+    } catch (IOException e) {
+      Assert.fail("Should not have thrown exception when creating bucket as" +
+          " a regular user here");
+    }
+
+    // Clean up
+    proxy.deleteBucket(volume, bucket);
+    proxy.deleteVolume(volume);
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/BucketManagerImpl.java

@@ -45,6 +45,7 @@
 
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.BUCKET_NOT_FOUND;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INTERNAL_ERROR;
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.VOLUME_NOT_FOUND;
 import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.BUCKET_LOCK;
 import static org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource.VOLUME_LOCK;
 
@@ -127,7 +128,7 @@ public void createBucket(OmBucketInfo bucketInfo) throws IOException {
       if (volumeArgs == null) {
         LOG.debug("volume: {} not found ", volumeName);
         throw new OMException("Volume doesn't exist",
-            OMException.ResultCodes.VOLUME_NOT_FOUND);
+            VOLUME_NOT_FOUND);
       }
       //Check if bucket already exists
       if (metadataManager.getBucketTable().get(bucketKey) != null) {
@@ -236,6 +237,12 @@ private void commitBucketInfoToDB(OmBucketInfo omBucketInfo)
    *
    * @param volumeName - Name of the Volume.
    * @param bucketName - Name of the Bucket.
+   * @return OmBucketInfo
+   * @throws IOException The exception thrown could be:
+   * 1. OMException VOLUME_NOT_FOUND when the parent volume doesn't exist.
+   * 2. OMException BUCKET_NOT_FOUND when the parent volume exists, but bucket
+   * doesn't.
+   * 3. Other exceptions, e.g. IOException thrown from getBucketTable().get().
    */
   @Override
   public OmBucketInfo getBucketInfo(String volumeName, String bucketName)
@@ -250,8 +257,16 @@ public OmBucketInfo getBucketInfo(String volumeName, String bucketName)
       if (value == null) {
         LOG.debug("bucket: {} not found in volume: {}.", bucketName,
             volumeName);
-        throw new OMException("Bucket not found",
-            BUCKET_NOT_FOUND);
+        // Check parent volume existence
+        final String dbVolumeKey = metadataManager.getVolumeKey(volumeName);
+        if (metadataManager.getVolumeTable().get(dbVolumeKey) == null) {
+          // Parent volume doesn't exist, throw VOLUME_NOT_FOUND
+          throw new OMException("Volume not found when getting bucket info",
+              VOLUME_NOT_FOUND);
+        } else {
+          // Parent volume exists, throw BUCKET_NOT_FOUND
+          throw new OMException("Bucket not found", BUCKET_NOT_FOUND);
+        }
       }
       return value;
     } catch (IOException ex) {

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestBucketManagerImpl.java

@@ -45,7 +45,6 @@
  * Tests BucketManagerImpl, mocks OMMetadataManager for testing.
  */
 @RunWith(MockitoJUnitRunner.class)
-@Ignore("Bucket Manager does not use cache, Disable it for now.")
 public class TestBucketManagerImpl {
   @Rule
   public ExpectedException thrown = ExpectedException.none();
@@ -102,6 +101,7 @@ public void testCreateBucketWithoutVolume() throws Exception {
   }
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testCreateBucket() throws Exception {
     OmMetadataManagerImpl metaMgr = createSampleVol();
 
@@ -136,6 +136,7 @@ public void testCreateBucket() throws Exception {
 
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testCreateEncryptedBucket() throws Exception {
     OmMetadataManagerImpl metaMgr = createSampleVol();
 
@@ -151,6 +152,7 @@ public void testCreateEncryptedBucket() throws Exception {
   }
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testCreateAlreadyExistingBucket() throws Exception {
     thrown.expectMessage("Bucket already exist");
     OmMetadataManagerImpl metaMgr = createSampleVol();
@@ -173,6 +175,7 @@ public void testCreateAlreadyExistingBucket() throws Exception {
   }
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testGetBucketInfoForInvalidBucket() throws Exception {
     thrown.expectMessage("Bucket not found");
     OmMetadataManagerImpl metaMgr = createSampleVol();
@@ -192,23 +195,51 @@ public void testGetBucketInfoForInvalidBucket() throws Exception {
 
   @Test
   public void testGetBucketInfo() throws Exception {
+    final String volumeName = "sampleVol";
+    final String bucketName = "bucketOne";
     OmMetadataManagerImpl metaMgr = createSampleVol();
 
     BucketManager bucketManager = new BucketManagerImpl(metaMgr);
+    // Check exception thrown when volume does not exist
+    try {
+      bucketManager.getBucketInfo(volumeName, bucketName);
+      Assert.fail("Should have thrown OMException");
+    } catch (OMException omEx) {
+      Assert.assertEquals("getBucketInfo() should have thrown " +
+              "VOLUME_NOT_FOUND as the parent volume is not created!",
+          ResultCodes.VOLUME_NOT_FOUND, omEx.getResult());
+    }
     OmBucketInfo bucketInfo = OmBucketInfo.newBuilder()
-        .setVolumeName("sampleVol")
-        .setBucketName("bucketOne")
+        .setVolumeName(volumeName)
+        .setBucketName(bucketName)
         .setStorageType(StorageType.DISK)
         .setIsVersionEnabled(false)
         .build();
+    // Note: the helper method createBucket() in this scope won't create the
+    // parent volume DB entry. In order to verify getBucketInfo's behavior, we
+    // need to create the volume entry in DB manually.
+    OmVolumeArgs args = OmVolumeArgs.newBuilder()
+            .setVolume(volumeName)
+            .setAdminName("bilbo")
+            .setOwnerName("bilbo")
+            .build();
+    TestOMRequestUtils.addVolumeToOM(metaMgr, args);
+    // Create bucket
     createBucket(metaMgr, bucketInfo);
-    OmBucketInfo result = bucketManager.getBucketInfo(
-        "sampleVol", "bucketOne");
-    Assert.assertEquals("sampleVol", result.getVolumeName());
-    Assert.assertEquals("bucketOne", result.getBucketName());
-    Assert.assertEquals(StorageType.DISK,
-        result.getStorageType());
-    Assert.assertEquals(false, result.getIsVersionEnabled());
+    // Check exception thrown when bucket does not exist
+    try {
+      bucketManager.getBucketInfo(volumeName, "bucketNotExist");
+      Assert.fail("Should have thrown OMException");
+    } catch (OMException omEx) {
+      Assert.assertEquals("getBucketInfo() should have thrown " +
+              "BUCKET_NOT_FOUND as the parent volume exists but bucket " +
+              "doesn't!", ResultCodes.BUCKET_NOT_FOUND, omEx.getResult());
+    }
+    OmBucketInfo result = bucketManager.getBucketInfo(volumeName, bucketName);
+    Assert.assertEquals(volumeName, result.getVolumeName());
+    Assert.assertEquals(bucketName, result.getBucketName());
+    Assert.assertEquals(StorageType.DISK, result.getStorageType());
+    Assert.assertFalse(result.getIsVersionEnabled());
     metaMgr.getStore().close();
   }
 
@@ -218,6 +249,7 @@ private void createBucket(OMMetadataManager metadataManager,
   }
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testSetBucketPropertyChangeStorageType() throws Exception {
     OmMetadataManagerImpl metaMgr = createSampleVol();
 
@@ -246,6 +278,7 @@ public void testSetBucketPropertyChangeStorageType() throws Exception {
   }
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testSetBucketPropertyChangeVersioning() throws Exception {
     OmMetadataManagerImpl metaMgr = createSampleVol();
 
@@ -272,6 +305,7 @@ public void testSetBucketPropertyChangeVersioning() throws Exception {
   }
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testDeleteBucket() throws Exception {
     thrown.expectMessage("Bucket not found");
     OmMetadataManagerImpl metaMgr = createSampleVol();
@@ -306,6 +340,7 @@ public void testDeleteBucket() throws Exception {
   }
 
   @Test
+  @Ignore("Bucket Manager does not use cache, Disable it for now.")
   public void testDeleteNonEmptyBucket() throws Exception {
     thrown.expectMessage("Bucket is not empty");
     OmMetadataManagerImpl metaMgr = createSampleVol();

hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneClientAdapterImpl.java

@@ -229,7 +229,7 @@ private OzoneBucket getBucket(String volumeStr, String bucketStr,
         // when ACL is disabled. Both exceptions need to be handled.
         switch (ex.getResult()) {
         case VOLUME_NOT_FOUND:
-        case BUCKET_NOT_FOUND:
+          // Create the volume first when the volume doesn't exist
           try {
             objectStore.createVolume(volumeStr);
           } catch (OMException newVolEx) {
@@ -238,7 +238,11 @@ private OzoneBucket getBucket(String volumeStr, String bucketStr,
               throw newVolEx;
             }
           }
-
+          // No break here. Proceed to create the bucket
+        case BUCKET_NOT_FOUND:
+          // When BUCKET_NOT_FOUND is thrown, we expect the parent volume
+          // exists, so that we don't call create volume and incur
+          // unnecessary ACL checks which could lead to unwanted behavior.
           OzoneVolume volume = proxy.getVolumeDetails(volumeStr);
           // Create the bucket
           try {