HBASE-28081 Snapshot working dir does not retain ACLs after snapshot commit phase (apache#5437)

Signed-off-by: Duo Zhang <zhangduo@apache.org>
Signed-off-by: Andrew Purtell <apurtell@apache.org>
Signed-off-by: Aman Poonia <aman.poonia.29@gmail.com>
(cherry picked from commit 0d4c756)
Change-Id: I8003de7b96ad7db6778a2187ad21557661030c9c

Author: virajjasani

hbase-server/src/main/java/org/apache/hadoop/hbase/master/snapshot/SnapshotManager.java

@@ -36,10 +36,13 @@
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonPathCapabilities;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.Stoppable;
@@ -486,6 +489,7 @@ private synchronized void prepareToTakeSnapshot(SnapshotDescription snapshot)
           "Couldn't create working directory (" + workingDir + ") for snapshot",
           ProtobufUtil.createSnapshotDesc(snapshot));
       }
+      updateWorkingDirAclsIfRequired(workingDir, workingDirFS);
     } catch (HBaseSnapshotException e) {
       throw e;
     } catch (IOException e) {
@@ -495,6 +499,42 @@ private synchronized void prepareToTakeSnapshot(SnapshotDescription snapshot)
     }
   }
 
+  /**
+   * If the parent dir of the snapshot working dir (e.g. /hbase/.hbase-snapshot) has non-empty ACLs,
+   * use them for the current working dir (e.g. /hbase/.hbase-snapshot/.tmp/{snapshot-name}) so that
+   * regardless of whether the snapshot commit phase performs atomic rename or non-atomic copy of
+   * the working dir to new snapshot dir, the ACLs are retained.
+   * @param workingDir   working dir to build the snapshot.
+   * @param workingDirFS working dir file system.
+   * @throws IOException If ACL read/modify operation fails.
+   */
+  private static void updateWorkingDirAclsIfRequired(Path workingDir, FileSystem workingDirFS)
+    throws IOException {
+    if (
+      !workingDirFS.hasPathCapability(workingDir, CommonPathCapabilities.FS_ACLS)
+        || workingDir.getParent() == null || workingDir.getParent().getParent() == null
+    ) {
+      return;
+    }
+    AclStatus snapshotWorkingParentDirStatus;
+    try {
+      snapshotWorkingParentDirStatus =
+        workingDirFS.getAclStatus(workingDir.getParent().getParent());
+    } catch (IOException e) {
+      LOG.warn("Unable to retrieve ACL status for path: {}, current working dir path: {}",
+        workingDir.getParent().getParent(), workingDir, e);
+      return;
+    }
+    List<AclEntry> snapshotWorkingParentDirAclStatusEntries =
+      snapshotWorkingParentDirStatus.getEntries();
+    if (
+      snapshotWorkingParentDirAclStatusEntries != null
+        && snapshotWorkingParentDirAclStatusEntries.size() > 0
+    ) {
+      workingDirFS.modifyAclEntries(workingDir, snapshotWorkingParentDirAclStatusEntries);
+    }
+  }
+
   /**
    * Take a snapshot of a disabled table.
    * @param snapshot description of the snapshot to take. Modified to be {@link Type#DISABLED}.

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/SnapshotScannerHDFSAclHelper.java

@@ -155,11 +155,12 @@ public boolean grantAcl(UserPermission userPermission, Set<String> skipNamespace
       long start = System.currentTimeMillis();
       handleGrantOrRevokeAcl(userPermission, HDFSAclOperation.OperationType.MODIFY, skipNamespaces,
         skipTables);
-      LOG.info("Set HDFS acl when grant {}, cost {} ms", userPermission,
-        System.currentTimeMillis() - start);
+      LOG.info("Set HDFS acl when grant {}, skipNamespaces: {}, skipTables: {}, cost {} ms",
+        userPermission, skipNamespaces, skipTables, System.currentTimeMillis() - start);
       return true;
     } catch (Exception e) {
-      LOG.error("Set HDFS acl error when grant: {}", userPermission, e);
+      LOG.error("Set HDFS acl error when grant: {}, skipNamespaces: {}, skipTables: {}",
+        userPermission, skipNamespaces, skipTables, e);
       return false;
     }
   }
@@ -177,11 +178,12 @@ public boolean revokeAcl(UserPermission userPermission, Set<String> skipNamespac
       long start = System.currentTimeMillis();
       handleGrantOrRevokeAcl(userPermission, HDFSAclOperation.OperationType.REMOVE, skipNamespaces,
         skipTables);
-      LOG.info("Set HDFS acl when revoke {}, cost {} ms", userPermission,
-        System.currentTimeMillis() - start);
+      LOG.info("Set HDFS acl when revoke {}, skipNamespaces: {}, skipTables: {}, cost {} ms",
+        userPermission, skipNamespaces, skipTables, System.currentTimeMillis() - start);
       return true;
     } catch (Exception e) {
-      LOG.error("Set HDFS acl error when revoke: {}", userPermission, e);
+      LOG.error("Set HDFS acl error when revoke: {}, skipNamespaces: {}, skipTables: {}",
+        userPermission, skipNamespaces, skipTables, e);
       return false;
     }
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestSnapshotScannerHDFSAclController.java

@@ -158,7 +158,6 @@ public void testGrantGlobal1() throws Exception {
 
     TestHDFSAclHelper.createTableAndPut(TEST_UTIL, table);
     snapshotAndWait(snapshot1, table);
-    snapshotAndWait(snapshot2, table);
     // grant G(R)
     SecureTestUtil.grantGlobal(TEST_UTIL, grantUserName, READ);
     TestHDFSAclHelper.canUserScanSnapshot(TEST_UTIL, grantUser, snapshot1, 6);
@@ -175,6 +174,8 @@ public void testGrantGlobal1() throws Exception {
     // grant G(R)
     SecureTestUtil.grantGlobal(TEST_UTIL, grantUserName, READ);
     TestHDFSAclHelper.canUserScanSnapshot(TEST_UTIL, grantUser, snapshot1, 6);
+    // take a snapshot and ACLs are inherited automatically
+    snapshotAndWait(snapshot2, table);
     TestHDFSAclHelper.canUserScanSnapshot(TEST_UTIL, grantUser, snapshot2, 6);
     assertTrue(hasUserGlobalHdfsAcl(aclTable, grantUserName));
     deleteTable(table);
@@ -196,10 +197,10 @@ public void testGrantGlobal2() throws Exception {
     // create table in namespace1 and snapshot
     TestHDFSAclHelper.createTableAndPut(TEST_UTIL, table1);
     snapshotAndWait(snapshot1, table1);
-    // grant G(W)
-    SecureTestUtil.grantGlobal(TEST_UTIL, grantUserName, WRITE);
     admin.grant(new UserPermission(grantUserName,
       Permission.newBuilder(namespace1).withActions(READ).build()), false);
+    // grant G(W)
+    SecureTestUtil.grantGlobal(TEST_UTIL, grantUserName, WRITE);
     // create table in namespace2 and snapshot
     TestHDFSAclHelper.createTableAndPut(TEST_UTIL, table2);
     snapshotAndWait(snapshot2, table2);
@@ -230,11 +231,11 @@ public void testGrantGlobal3() throws Exception {
     // grant table1(R)
     TestHDFSAclHelper.createTableAndPut(TEST_UTIL, table1);
     snapshotAndWait(snapshot1, table1);
-    TestHDFSAclHelper.createTableAndPut(TEST_UTIL, table2);
-    snapshotAndWait(snapshot2, table2);
+    TestHDFSAclHelper.grantOnTable(TEST_UTIL, grantUserName, table1, READ);
     // grant G(W)
     SecureTestUtil.grantGlobal(TEST_UTIL, grantUserName, WRITE);
-    TestHDFSAclHelper.grantOnTable(TEST_UTIL, grantUserName, table1, READ);
+    TestHDFSAclHelper.createTableAndPut(TEST_UTIL, table2);
+    snapshotAndWait(snapshot2, table2);
     // check scan snapshot
     TestHDFSAclHelper.canUserScanSnapshot(TEST_UTIL, grantUser, snapshot1, 6);
     TestHDFSAclHelper.canUserScanSnapshot(TEST_UTIL, grantUser, snapshot2, -1);