Merge pull request #27 from gluster/permissions_merged_2

Permissions patch to fix BZ_951305

Author: jayunit100

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

@@ -20,14 +20,16 @@
 package org.apache.hadoop.fs.glusterfs;
 
 import java.io.*;
-
+/**
+ * An OutputStream for writing to a FUSE mount intended for use with gluster. 
+ */
 public class GlusterFUSEOutputStream extends OutputStream{
     File f;
     long pos;
     boolean closed;
     OutputStream fuseOutputStream;
     org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(GlusterFUSEOutputStream.class);
-  
+
     public GlusterFUSEOutputStream(String file, boolean append) throws IOException{
         this(file,append,0);
     }
@@ -43,7 +45,6 @@ public GlusterFUSEOutputStream(String file, boolean append, int bufferSize) thro
         	fuseOutputStream = new BufferedOutputStream(fuseOutputStream, bufferSize);
         this.closed=false;
     }
-    
     public long getPos() throws IOException{
         return pos;
     }

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

@@ -203,15 +203,24 @@ public boolean exists(Path path) throws IOException{
      * org.apache.hadoop.fs.RawLocalFileSystem#mkdirs(org.apache.hadoop.fs.Path)
      * as incremental fix towards a re-write. of this class to remove duplicity.
      */
-    public boolean mkdirs(Path f,FsPermission permission) throws IOException{
-
-        if(f==null)
-            return true;
-
-        Path parent=f.getParent();
-        Path absolute=makeAbsolute(f);
-        File p2f=new File(absolute.toUri().getPath());
-        return (f==null||mkdirs(parent))&&(p2f.mkdir()||p2f.isDirectory());
+    public boolean mkdirs(Path path,FsPermission permission) throws IOException{
+      
+            String split[]=path.toString().split(Path.SEPARATOR);
+            String current="";
+            boolean success=true;
+            for(int i=0;i<split.length&&success;i++){
+                current+=split[i]+Path.SEPARATOR;
+                Path absolute=makeAbsolute(new Path(current));
+                File p2f=new File(absolute.toUri().getPath());
+                if(!p2f.exists()){
+                    p2f.mkdirs();
+                    setPermission(new Path(current), permission);
+                }
+                success=p2f.exists();
+            }
+       
+            return success;
+       
     }
 
     @Deprecated
@@ -298,14 +307,18 @@ public FUSEFileStatus(File f, boolean isdir, int block_replication, long blocksi
          */
         @Override
         public String getOwner(){
-            try{
-                return FileInfoUtil.getLSinfo(theFile.getAbsolutePath()).get("owner");
-            }catch (Exception e){
-                throw new RuntimeException(e);
-            }
+             loadPermissionInfo();
+             return super.getOwner();
         }
+        
+
+		@Override
+		public String getGroup() {
+            loadPermissionInfo();
+			return super.getGroup();
+		}
 
-        public FsPermission getPermission(){
+		public FsPermission getPermission(){
             // should be amortized, see method.
             loadPermissionInfo();
             return super.getPermission();
@@ -322,7 +335,6 @@ private void loadPermissionInfo(){
             try{
                 String output;
                 StringTokenizer t=new StringTokenizer(output=execCommand(theFile, Shell.getGET_PERMISSION_COMMAND()));
-
                 // log.info("Output of PERMISSION command = " + output
                 // + " for " + this.getPath());
                 // expected format
@@ -371,6 +383,9 @@ public static String execCommand(File f,String...cmd) throws IOException{
      */
     @Override
     public void setPermission(Path p,FsPermission permission){
+        
+        if(permission==null) return;
+        
         try{
             Path absolute=makeAbsolute(p);
             final File f=new File(absolute.toUri().getPath());
@@ -409,7 +424,7 @@ public FSDataOutputStream create(Path path,FsPermission permission,boolean overw
         Path absolute=makeAbsolute(path);
         Path parent=null;
         File f=null;
-        File fParent=null;
+      
         FSDataOutputStream glusterFileStream=null;
 
         f=new File(absolute.toUri().getPath());
@@ -422,7 +437,10 @@ public FSDataOutputStream create(Path path,FsPermission permission,boolean overw
         }
 
         parent=path.getParent();
-        mkdirs(parent);
+        mkdirs(parent, permission);
+        
+        f.createNewFile();
+        setPermission(path, permission);
 
         glusterFileStream=new FSDataOutputStream(new GlusterFUSEOutputStream(f.getPath(), false, writeBufferSize));
 
@@ -494,7 +512,7 @@ public boolean delete(Path path,boolean recursive) throws IOException{
         if(dirEntries!=null)
             for(int i=0;i<dirEntries.length;i++)
                 delete(new Path(absolute, dirEntries[i].getPath()), recursive);
-
+   
         return f.delete();
     }
 
@@ -562,8 +580,31 @@ public BlockLocation[] getFileBlockLocations(FileStatus file,long start,long len
         return result;
     }
 
+    /**
+     * Adopted from {@link org.apache.hadoop.RawLocalFileSystem}, so that group privileges are
+     * set properly when hadoop fs chwon is called in {@link org.apache.hadoop.fs.FSShellPermissions}.
+     */
+    @Override
+    public void setOwner(Path p, String username, String groupname) throws IOException {
+        Path absolute=makeAbsolute(p);
+        File f=new File(absolute.toUri().getPath());
+
+    	if (username == null && groupname == null) {
+    	  throw new IOException("username == null && groupname == null");
+      }
+
+      if (username == null) {
+        execCommand(f, Shell.SET_GROUP_COMMAND, groupname); 
+      } 
+      else {
+        //OWNER[:[GROUP]]
+        String s = username + (groupname == null? "": ":" + groupname);
+        execCommand(f, Shell.SET_OWNER_COMMAND, s);
+      }
+    }
+    
     public void copyFromLocalFile(boolean delSrc,Path src,Path dst) throws IOException{
-        FileUtil.copy(glusterFs, src, this, dst, delSrc, getConf());
+    	FileUtil.copy(glusterFs, src, this, dst, delSrc, getConf());
     }
 
     public void copyToLocalFile(boolean delSrc,Path src,Path dst) throws IOException{

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

@@ -154,7 +154,7 @@ public void testTolerantMkdirs() throws Exception{
 
         /* delete the directories */
 
-        gfs.delete(longPath);
+        gfs.delete(new Path("a"), true);
         assertFalse(gfs.exists(longPath));
 
     }
@@ -168,8 +168,8 @@ public void testOwner() throws Exception{
         final String me=System.getProperties().getProperty("user.name");
         Path myFile=new Path("to_owned_by_me.txt");
         gfs.create(myFile);
-        System.out.println("Asserting that "+myFile+" is owned by "+me);
         Assert.assertEquals(gfs.getFileStatus(myFile).getOwner(), me);
+        gfs.delete(myFile);
     }
 
     @org.junit.Test
@@ -196,6 +196,61 @@ public void testTextWriteAndRead() throws Exception{
         assertFalse(gfs.exists(new Path("test1")));
     }
 
+    @Test
+    public void testGroupOwnership() throws Exception{
+        Path myFile=new Path("filePerm.txt");
+        //Create a file 
+        gfs.create(myFile);
+        
+        //Set the initial owner
+        gfs.setOwner(myFile, "daemon", "root");
+        String oldOwner = gfs.getFileStatus(myFile).getOwner(); 
+        String oldGroup = gfs.getFileStatus(myFile).getGroup();
+        Assert.assertEquals("daemon",oldOwner);
+        Assert.assertEquals("root",oldGroup);
+        
+        //Now, change it to "root" "wheel" 
+        gfs.setOwner(myFile, "root", "wheel");
+        String newOwner = gfs.getFileStatus(myFile).getOwner(); 
+        String newGroup = gfs.getFileStatus(myFile).getGroup();
+        Assert.assertEquals("root",newOwner);
+        Assert.assertEquals("wheel",newGroup);
+        
+        gfs.delete(myFile,true);
+    }
+    
+    @org.junit.Test
+    public void testPermissions() throws Exception{
+
+        Path myFile=new Path("filePerm.txt");
+        gfs.create(myFile);
+        short perm=0777;
+        gfs.setPermission(myFile, new FsPermission(perm));
+        assertEquals(gfs.getFileStatus(myFile).getPermission().toShort(), perm);
+
+        perm=0700;
+        gfs.setPermission(myFile, new FsPermission(perm));
+        assertEquals(gfs.getFileStatus(myFile).getPermission().toShort(), perm);
+
+        gfs.delete(myFile);
+        assertFalse(gfs.exists(myFile));
+        
+        /* directory permissions */
+        Path directory = new Path("aa/bb/cc");
+        perm = 0700;
+        gfs.mkdirs(directory, new FsPermission(perm));
+        assertEquals(gfs.getFileStatus(directory).getPermission().toShort(), perm);
+        gfs.delete(new Path("aa"),true);
+        assertFalse(gfs.exists(directory));
+        
+        
+        perm = 0777;
+        gfs.mkdirs(directory, new FsPermission(perm));
+        assertEquals(gfs.getFileStatus(directory).getPermission().toShort(), perm);
+        gfs.delete(new Path("aa"),true);
+        assertFalse(gfs.exists(directory));
+    }
+
     @org.junit.Test
     public void testZDirs() throws Exception{
         final Path subDir1=new Path("td_dir.1");
@@ -378,5 +433,7 @@ public void test0aPermissions() throws Exception{
         Assert.assertTrue(this.gfs.getFileStatus(theFile).getPermission().getGroupAction().equals(changeTo.getGroupAction()));
         Assert.assertTrue(this.gfs.getFileStatus(theFile).getPermission().getUserAction().equals(changeTo.getUserAction()));
         Assert.assertTrue(this.gfs.getFileStatus(theFile).getPermission().getOtherAction().equals(changeTo.getOtherAction()));
+        gfs.delete(new Path("mnt"),true);
+        
     }
 }