Merge pull request #61 from childsb/super_user

multi-user/multi-tenancy support for hadoop on glusterfs

Author: childsb

conf/core-site.xml

@@ -24,7 +24,12 @@
     <value>glusterfs://ambari-3.abrv8.com:gv0<</value>
   </property>
 
-
+  <!-- a super user that the hadoop daemons run under. required for multi-user secure clusters -->
+  <property>
+    <name>gluster.daemon.user</name>
+    <value>hadoop</value>
+  </property>
+  
   <!--  this is the default glusterfs hook with no additional integrity checking on files -->
   <property>
     <name>fs.AbstractFileSystem.glusterfs.impl</name>

conf/mapred-site.xml

@@ -16,6 +16,16 @@
     <name>mapred.system.dir</name>
     <value>glusterfs:///mapred/system</value>
   </property>
+  <property>
+    <name>mapreduce.jobhistory.done-dir</name>
+    <value>glusterfs:///job-history/done</value>
+  </property>
+  
+  <property>
+    <name>mapreduce.jobhistory.intermediate-done-dir</name>
+    <value>glusterfs:///job-history/intermediate-done</value>
+  </property>
+  
   <property>
     <name>mapreduce.jobtracker.staging.root.dir</name>
     <value>glusterfs:///user</value>

src/main/java/org/apache/hadoop/fs/glusterfs/AclPathFilter.java

@@ -0,0 +1,59 @@
+package org.apache.hadoop.fs.glusterfs;
+
+import java.io.IOException;
+import java.util.ListIterator;
+import java.util.Vector;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
+
+
+public class AclPathFilter {
+	
+	Vector<String> paths = null;
+	
+	public AclPathFilter(){
+		paths = new Vector<String>();
+	}
+	/* generates a white list of ACL path regular expressions */
+	public AclPathFilter(Configuration conf){
+		this();
+	    UserGroupInformation ugi = null;
+	    try {
+			ugi = UserGroupInformation.getCurrentUser();
+		} catch (IOException e) {
+			
+		}
+		String stagingRootDir = new Path(conf.get("yarn.app.mapreduce.am.staging-dir",conf.get("mapreduce.jobtracker.staging.root.dir", "/tmp/hadoop/mapred/staging"))).toString();
+		String user;
+		String randid = "\\d*";
+		if (ugi != null) {
+			user = ugi.getShortUserName();
+		} else {
+		    user = "dummy";
+		}
+		paths.add("^" + new Path(stagingRootDir, user +  randid +"/.staging").toString() + ".*");
+		paths.add("^" + (new Path(stagingRootDir, user +  randid +"/.staging")).toUri().getPath() + ".*");
+		stagingRootDir = new Path(conf.get("mapreduce.jobtracker.staging.root.dir","/tmp/hadoop/mapred/staging")).toString();
+        paths.add("^" + new Path(stagingRootDir, user+"/.staging").toString() + ".*");
+        paths.add("^" + (new Path(stagingRootDir, user+"/.staging")).toUri().getPath() + ".*");
+		
+	}
+	
+	public boolean matches(String path){
+		
+		boolean needsAcl = false;
+		ListIterator<String> list = paths.listIterator();
+		String filterPath = null;
+		while(list.hasNext() && !needsAcl){
+			filterPath = list.next();
+			if(path.matches(filterPath)){
+				needsAcl = true;
+			}
+		}
+		return needsAcl;
+	}
+	public boolean matches(Path path){
+		return matches(path.toString());
+	}
+}

src/main/java/org/apache/hadoop/fs/glusterfs/GlusterVolume.java

@@ -35,6 +35,7 @@
 import org.apache.hadoop.fs.RawLocalFileSystem;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.permission.FsPermission;
 
 public class GlusterVolume extends RawLocalFileSystem{
 
@@ -43,6 +44,8 @@ public class GlusterVolume extends RawLocalFileSystem{
     public static final URI NAME = URI.create("glusterfs:///");
 
     protected String root=null;
+    protected String superUser=null;
+    protected AclPathFilter aclFilter = null;
 
     protected static GlusterFSXattr attr = null;
 
@@ -81,6 +84,10 @@ public void setConf(Configuration conf){
                     mkdirs(mapredSysDirectory);
                 }
 
+                superUser =  conf.get("gluster.daemon.user", null);
+                
+                aclFilter = new AclPathFilter(conf);
+                
                 /* ensure the initial working directory exists */
                 Path workingDirectory = getInitialWorkingDirectory();
                 mkdirs(workingDirectory);
@@ -104,7 +111,7 @@ public File pathToFile(Path path) {
     }
 
     @Override
-	protected Path getInitialWorkingDirectory() {
+    protected Path getInitialWorkingDirectory() {
 		/* apache's unit tests use a default working direcotry like this: */
        return new Path(this.NAME + "user/" + System.getProperty("user.name"));
         /* The super impl returns the users home directory in unix */
@@ -113,9 +120,9 @@ protected Path getInitialWorkingDirectory() {
 
 	public Path fileToPath(File path) {
         return new Path(NAME.toString() + path.toURI().getRawPath().substring(root.length()));
-    }
+     }
 
-	public boolean rename(Path src, Path dst) throws IOException {
+     public boolean rename(Path src, Path dst) throws IOException {
 		File dest = pathToFile(dst);
 
 		/* two HCFS semantics java.io.File doesn't honor */
@@ -200,7 +207,36 @@ public long getBlockSize(Path path) throws IOException{
 
         return blkSz;
     }
-   
+    /*
+     * ensures the 'super user' is given read/write access.  
+     * the ACL drops off after a chmod or chown.
+     */
+    
+    private void updateAcl(Path p){
+    	if(superUser!=null && aclFilter.matches(p)  ){
+    		File f = pathToFile(p);
+    		String path = f.getAbsolutePath();
+    		String command = "setfacl -m u:" + superUser + ":rwx " + path;
+    		try{
+    			Runtime.getRuntime().exec(command);
+    		}catch(IOException ex){
+    			throw new RuntimeException(ex);
+    		}
+    	}
+    }
+    
+    public void setOwner(Path p, String username, String groupname)
+            throws IOException {
+    	super.setOwner(p,username,groupname);
+    	updateAcl(p);
+    	
+    }
+    
+    public void setPermission(Path p, FsPermission permission)
+            throws IOException {
+    	super.setPermission(p,permission);
+    	updateAcl(p);
+    }
     public BlockLocation[] getFileBlockLocations(FileStatus file,long start,long len) throws IOException{
         File f=pathToFile(file.getPath());
         BlockLocation[] result=null;