support distributed chmod

hadoop-hdfs-project/hadoop-hdfs-db/src/main/java/org/apache/hadoop/hdfs/db/DatabaseINode.java

@@ -587,6 +587,32 @@ public static void updateModificationTime(final long id, final long childId) {
     }
   }
 
+  // (distributed) transaction
+  public static void setPermissions(final List<String> parents, final List<String> names, final long permission) {
+    try {
+      DatabaseConnection obj = Database.getInstance().getConnection();
+      String env = System.getenv("DATABASE");
+      if (env.equals("VOLT")) {
+        try {
+          obj.getVoltClient().callProcedure("SetPermissions",
+            parents.toArray(new String[parents.size()]),
+            names.toArray(new String[names.size()]),
+            permission);
+        } catch (Exception e) {
+          e.printStackTrace();
+        }
+      } else {
+        throw new SQLException("[UNSUPPORT] Invalid operation ...");
+      }
+      Database.getInstance().retConnection(obj);
+    } catch (SQLException ex) {
+      System.err.println(ex.getMessage());
+    }
+    if (LOG.isInfoEnabled()) {
+      LOG.info("permissions [UPDATE]: (" + permission + ")");
+    } 
+  }
+
   public static void setPermission(final long id, final long permission) {
     try {
       DatabaseConnection obj = Database.getInstance().getConnection();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/nnproxy/server/mount/MountsManager.java

@@ -6,6 +6,7 @@
 import java.util.*;
 import java.util.concurrent.*;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.cache.NodeCache;
@@ -136,6 +137,17 @@ public String resolve(String path) {
     return chosen.fsUri;
   }
 
+  public Set<Pair<String, String>> resolveSubPaths(String path) {
+    Set<Pair<String, String>> subPaths = new HashSet<>();
+    ImmutableList<MountEntry> entries = this.mounts;
+    for (MountEntry entry: entries) {
+      if (entry.mountPoint.startsWith(path)) {
+        subPaths.add(Pair.of(entry.mountPoint, entry.fsUri.replace("hdfs://","").split(":")[0]));
+      }
+    }
+    return subPaths;
+  }
+
   public String resolveForBench(String path) {
     String parent = "";
     if (path.charAt(49) == '/')

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAttrOp.java

@@ -253,7 +253,7 @@ static void unprotectedSetPermission(
       FSDirectory fsd, INodesInPath iip, FsPermission permissions)
       throws FileNotFoundException, UnresolvedLinkException,
              QuotaExceededException, SnapshotAccessControlException {
-    assert fsd.hasWriteLock();
+    // assert fsd.hasWriteLock();
     final INode inode = FSDirectory.resolveLastINode(iip);
     int snapshotId = iip.getLatestSnapshotId();
     inode.setPermission(permissions, snapshotId);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogProtocol.java

@@ -29,4 +29,5 @@ public interface FSEditLogProtocol extends VersionedProtocol {
     public INodeDirectory loadINodeDirectory(INodeSection.INode n);
     public INodeFile loadINodeFile(INodeSection.INode n);
     public void logEdit(byte[] inode) throws IOException;
+    public void invalidateAndWriteBackDB(byte[] mpoint) throws IOException;
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogProtocolImpl.java

@@ -54,6 +54,7 @@
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.FilesUnderConstructionSection.FileUnderConstructionEntry;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.INodeDirectorySection;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.NamespaceSubtree;
+import org.apache.hadoop.hdfs.server.namenode.FsImageProto.MountPoint;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.INodeSection;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.INodeSection.AclFeatureProto;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.INodeSection.XAttrCompactProto;
@@ -203,6 +204,17 @@ public void logEdit(byte[] in) throws IOException {
         }
     }
 
+    @Override
+    public void invalidateAndWriteBackDB(byte[] in) throws IOException {
+        MountPoint mpoint = null;
+        try {	
+            mpoint = MountPoint.parseFrom(in);	
+        } catch (InvalidProtocolBufferException e) {	
+            e.printStackTrace();	
+        }	
+        INodeWithAdditionalFields.invalidateAndWriteBackDB(mpoint.getParent(), mpoint.getName());
+    }
+
     @Override
     public long getProtocolVersion(String s, long l) throws IOException {
         return FSEditLogProtocol.versionID;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -1908,7 +1908,7 @@ void setPermission(String src, FsPermission permission) throws IOException {
     FileStatus auditStat;
     checkOperation(OperationCategory.WRITE);
     final FSPermissionChecker pc = getPermissionChecker();
-    writeLock();
+    // writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot set permission for " + src);
@@ -1917,7 +1917,7 @@ void setPermission(String src, FsPermission permission) throws IOException {
       logAuditEvent(false, operationName, src);
       throw e;
     } finally {
-      writeUnlock(operationName);
+      // writeUnlock(operationName);
     }
     getEditLog().logSync();
     logAuditEvent(true, operationName, src, null, auditStat);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeWithAdditionalFields.java

@@ -14,6 +14,8 @@
  */
 package org.apache.hadoop.hdfs.server.namenode;
 
+import java.io.File;
+import java.util.*;
 import com.google.common.base.Preconditions;
 import java.util.concurrent.CompletableFuture;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -23,6 +25,14 @@
 import org.apache.hadoop.hdfs.db.*;
 import org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot;
 import org.apache.hadoop.hdfs.util.LongBitFormat;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdfs.server.namenode.FsImageProto.MountPoint;
+import org.apache.hadoop.ipc.RPC;
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import com.google.protobuf.ByteString;
+import org.apache.hadoop.conf.Configuration;
 
 /**
  * {@link INode} with additional fields including id, name, permission, access time and modification
@@ -361,9 +371,122 @@ private final void setPermission(long perm) {
     INodeKeyedObjects.getUpdateSet().add(getPath());
   }
 
+  private static void update_subtree(Set<INode> inodes) {
+    List<Long> longAttr = new ArrayList<>();
+    List<String> strAttr = new ArrayList<>();
+
+    List<Long> fileIds = new ArrayList<>();
+    List<String> fileAttr = new ArrayList<>();
+    Iterator<INode> iterator = inodes.iterator();
+    while (iterator.hasNext()) {
+      INode inode = iterator.next();
+      if (inode == null) continue;
+      strAttr.add(inode.getLocalName());
+      if (inode.getId() == 16385) {
+        strAttr.add(" ");
+      } else {
+        strAttr.add(inode.getParentName());
+      }
+      longAttr.add(inode.getParentId());
+      longAttr.add(inode.getId());
+      longAttr.add(inode.getModificationTime());
+      longAttr.add(inode.getAccessTime());
+      longAttr.add(inode.getPermissionLong());
+      if (inode.isDirectory()) {
+        longAttr.add(0L);
+      } else {
+        longAttr.add(inode.asFile().getHeaderLong());
+        FileUnderConstructionFeature uc = inode.asFile().getFileUnderConstructionFeature();
+        if (uc != null) {
+          fileIds.add(inode.getId());
+          fileAttr.add(uc.getClientName(inode.getId()));
+          fileAttr.add(uc.getClientMachine(inode.getId()));
+        }
+      }
+      iterator.remove();
+    }
+    try {
+      if (strAttr.size() > 0) {
+        DatabaseINode.batchUpdateINodes(longAttr, strAttr, fileIds, fileAttr);
+      }
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+  }
+
+  public static final void invalidateAndWriteBackDB(String parent, String name) {
+    Queue<ImmutablePair<String, String>> q = new LinkedList<>();
+    q.add(new ImmutablePair<>(parent, name));
+
+    ImmutablePair<String, String> id = null;
+    Set<INode> inodes = new HashSet<>();
+    while ((id = q.poll()) != null) {
+      INode child = FSDirectory.getInstance().getInode(id.getLeft(), id.getRight());   
+      if (child != null) {
+        if (child.isDirectory()) {
+          HashSet<String> childNames = ((INodeDirectory)child).getCurrentChildrenList2();
+          for (String cname : childNames) {
+            q.add(new ImmutablePair<>(child.getPath(), cname));
+          }
+        }
+        inodes.add(child);
+        // invalidate inode
+        INodeKeyedObjects.getCache().invalidate(child.getPath());
+        if (inodes.size() >= 5120) {
+          // write back to db
+          update_subtree(inodes);
+        }
+      }
+      if (inodes.size() > 0) {
+        // write back to db
+        update_subtree(inodes);
+      }
+    }
+  }
+
+  private final void remoteChmod(Set<Pair<String, String>> mpoints) {
+    // 1. invalidate cache and write back dirty data
+    invalidateAndWriteBackDB(getParentName(), getLocalName());
+    List<String> parents = new ArrayList<>();
+    List<String> names = new ArrayList<>();
+    for (Pair<String, String> pair : mpoints) {
+      File file = new File(pair.getLeft());
+      String parent = file.getParent();
+      String name = file.getName();
+      String url = pair.getRight();
+      try {
+        MountPoint.Builder b = MountPoint.newBuilder().setParent(parent).setName(name);
+        byte[] data = b.build().toByteArray();
+
+        FSEditLogProtocol proxy = (FSEditLogProtocol) RPC.getProxy(
+          FSEditLogProtocol.class, FSEditLogProtocol.versionID,
+          new InetSocketAddress(url, 10087), new Configuration());
+        proxy.invalidateAndWriteBackDB(data);
+      } catch (Exception e) {
+        e.printStackTrace();
+      }
+
+      parents.add(parent);
+      names.add(name);
+    }
+
+    // 2. execute distributed txn
+    DatabaseINode.setPermissions(parents, names, this.permission);
+  }
+
   private final void updatePermissionStatus(PermissionStatusFormat f, long n) {
     permission = f.BITS.combine(n, getPermissionLong());
-    INodeKeyedObjects.getUpdateSet().add(getPath());
+    if (FSDirectory.getInstance().isLocalNN()) {
+      INodeKeyedObjects.getUpdateSet().add(getPath());
+    } else if (isDirectory()) {
+      try {
+        Set<Pair<String, String>> mpoints = FSDirectory.getInstance().getMountsManager().resolveSubPaths(getPath());
+        LOG.info(getPath() + " has sub-paths that are mounted into: " + mpoints);
+        remoteChmod(mpoints);
+      } catch (Exception e) {
+        e.printStackTrace();
+      }
+    }
   }
 
   @Override

hadoop-hdfs-project/hadoop-hdfs/src/main/proto/fsimage.proto

@@ -120,6 +120,11 @@ message Operation {
   }
 }
 
+message MountPoint {
+  required string parent = 1;
+  required string name = 2;
+}
+
 // namespace subtree
 message NamespaceSubtree {
   repeated INodeSection.INode inodes = 1;