HBASE-21995 Add a coprocessor to set HDFS ACL for hbase granted user

hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MasterObserver.java

@@ -750,6 +750,16 @@ default void postSnapshot(final ObserverContext<MasterCoprocessorEnvironment> ct
       final SnapshotDescription snapshot, final TableDescriptor tableDescriptor)
       throws IOException {}
 
+  /**
+   * Called after the snapshot operation has been completed.
+   * @param ctx the environment to interact with the framework and master
+   * @param snapshot the SnapshotDescriptor for the snapshot
+   * @param tableDescriptor the TableDescriptor of the table to snapshot
+   */
+  default void postCompletedSnapshotAction(ObserverContext<MasterCoprocessorEnvironment> ctx,
+      SnapshotDescription snapshot, TableDescriptor tableDescriptor) throws IOException {
+  }
+
   /**
    * Called before listSnapshots request has been processed.
    * @param ctx the environment to interact with the framework and master

hbase-server/src/main/java/org/apache/hadoop/hbase/io/FileLink.java

@@ -26,9 +26,6 @@
 import java.io.FileNotFoundException;
 import java.util.List;
 
-import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.apache.hadoop.fs.CanSetDropBehind;
 import org.apache.hadoop.fs.CanSetReadahead;
 import org.apache.hadoop.fs.CanUnbuffer;
@@ -40,6 +37,10 @@
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.hbase.util.FSUtils;
 import org.apache.hadoop.ipc.RemoteException;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * The FileLink is a sort of hardlink, that allows access to a file given a set of locations.
@@ -297,6 +298,7 @@ public void unbuffer() {
      * @throws IOException on unexpected error, or file not found.
      */
     private FSDataInputStream tryOpen() throws IOException {
+      IOException exception = null;
       for (Path path: fileLink.getLocations()) {
         if (path.equals(currentPath)) continue;
         try {
@@ -312,14 +314,11 @@ private FSDataInputStream tryOpen() throws IOException {
           }
           currentPath = path;
           return(in);
-        } catch (FileNotFoundException e) {
-          // Try another file location
-        } catch (RemoteException re) {
-          IOException ioe = re.unwrapRemoteException(FileNotFoundException.class);
-          if (!(ioe instanceof FileNotFoundException)) throw re;
+        } catch (FileNotFoundException | AccessControlException | RemoteException e) {
+          exception = FileLink.handleAccessLocationException(fileLink, e, exception);
         }
       }
-      throw new FileNotFoundException("Unable to open link: " + fileLink);
+      throw exception;
     }
 
     @Override
@@ -405,14 +404,47 @@ public Path getAvailablePath(FileSystem fs) throws IOException {
    * @throws IOException on unexpected error.
    */
   public FileStatus getFileStatus(FileSystem fs) throws IOException {
+    IOException exception = null;
     for (int i = 0; i < locations.length; ++i) {
       try {
         return fs.getFileStatus(locations[i]);
-      } catch (FileNotFoundException e) {
-        // Try another file location
+      } catch (FileNotFoundException | AccessControlException e) {
+        exception = handleAccessLocationException(this, e, exception);
       }
     }
-    throw new FileNotFoundException("Unable to open link: " + this);
+    throw exception;
+  }
+
+  /**
+   * Handle exceptions which are threw when access locations of file link
+   * @param fileLink the file link
+   * @param newException the exception caught by access the current location
+   * @param previousException the previous exception caught by access the other locations
+   * @return return AccessControlException if access one of the locations caught, otherwise return
+   *         FileNotFoundException. The AccessControlException is threw if user scan snapshot
+   *         feature is enabled, see
+   *         {@link org.apache.hadoop.hbase.security.access.SnapshotScannerHDFSAclController}.
+   * @throws IOException if the exception is neither AccessControlException nor
+   *           FileNotFoundException
+   */
+  private static IOException handleAccessLocationException(FileLink fileLink,
+      IOException newException, IOException previousException) throws IOException {
+    if (newException instanceof RemoteException) {
+      newException = ((RemoteException) newException)
+          .unwrapRemoteException(FileNotFoundException.class, AccessControlException.class);
+    }
+    if (newException instanceof FileNotFoundException) {
+      // Try another file location
+      if (previousException == null) {
+        previousException = new FileNotFoundException("Unable to open link: " + fileLink);
+      }
+    } else if (newException instanceof AccessControlException) {
+      // Try another file location
+      previousException = newException;
+    } else {
+      throw newException;
+    }
+    return previousException;
   }
 
   /**

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

@@ -1067,6 +1067,16 @@ public void call(MasterObserver observer) throws IOException {
     });
   }
 
+  public void postCompletedSnapshotAction(SnapshotDescription snapshot,
+      TableDescriptor hTableDescriptor) throws IOException {
+    execOperation(coprocEnvironments.isEmpty() ? null : new MasterObserverOperation() {
+      @Override
+      public void call(MasterObserver observer) throws IOException {
+        observer.postCompletedSnapshotAction(this, snapshot, hTableDescriptor);
+      }
+    });
+  }
+
   public void preListSnapshot(final SnapshotDescription snapshot) throws IOException {
     execOperation(coprocEnvironments.isEmpty() ? null : new MasterObserverOperation() {
       @Override

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

@@ -59,7 +59,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
-
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos.SnapshotDescription;
 
 /**
@@ -228,6 +228,10 @@ public void process() {
       status.markComplete(msg);
       LOG.info(msg);
       metricsSnapshot.addSnapshot(status.getCompletionTimestamp() - status.getStartTime());
+      if (master.getMasterCoprocessorHost() != null) {
+        master.getMasterCoprocessorHost()
+            .postCompletedSnapshotAction(ProtobufUtil.createSnapshotDesc(snapshot), this.htd);
+      }
     } catch (Exception e) { // FindBugs: REC_CATCH_EXCEPTION
       status.abort("Failed to complete snapshot " + snapshot.getName() + " on table " +
           snapshotTable + " because " + e.getMessage());

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

@@ -263,6 +263,7 @@ private static void removePermissionRecord(Configuration conf, UserPermission us
   static void removeTablePermissions(Configuration conf, TableName tableName, Table t)
       throws IOException{
     Delete d = new Delete(tableName.getName());
+    d.addFamily(ACL_LIST_FAMILY);
 
     if (LOG.isDebugEnabled()) {
       LOG.debug("Removing permissions of removed table "+ tableName);
@@ -280,7 +281,7 @@ static void removeTablePermissions(Configuration conf, TableName tableName, Tabl
   static void removeNamespacePermissions(Configuration conf, String namespace, Table t)
       throws IOException{
     Delete d = new Delete(Bytes.toBytes(toNamespaceEntry(namespace)));
-
+    d.addFamily(ACL_LIST_FAMILY);
     if (LOG.isDebugEnabled()) {
       LOG.debug("Removing permissions of removed namespace "+ namespace);
     }
@@ -839,17 +840,21 @@ public static ListMultimap<String, Permission> readPermissions(byte[] data,
   }
 
   public static boolean isGlobalEntry(byte[] entryName) {
-    return entryName != null && TableName.valueOf(entryName).equals(ACL_TABLE_NAME);
+    return Bytes.equals(entryName, ACL_GLOBAL_NAME);
   }
 
   public static boolean isNamespaceEntry(String entryName) {
-    return entryName != null && entryName.charAt(0) == NAMESPACE_PREFIX;
+    return isNamespaceEntry(Bytes.toBytes(entryName));
   }
 
   public static boolean isNamespaceEntry(byte[] entryName) {
     return entryName != null && entryName.length !=0 && entryName[0] == NAMESPACE_PREFIX;
   }
 
+  public static boolean isTableEntry(byte[] entryName) {
+    return !isNamespaceEntry(entryName) && !isGlobalEntry(entryName) && entryName != null;
+  }
+
   public static String toNamespaceEntry(String namespace) {
     return NAMESPACE_PREFIX + namespace;
   }