Răsfoiți Sursa

HUE-2321 [fb] Support dfs.umaskmode first pass

Christopher Conner 11 ani în urmă
părinte
comite
ab454c440f

+ 10 - 0
desktop/libs/hadoop/src/hadoop/conf.py

@@ -40,6 +40,14 @@ def find_file_recursive(desired_glob, root):
   return f
   return f
 
 
 
 
+def coerce_umask(umask):
+  if len(umask) < 4:
+    umask = "1" + umask
+    return int(umask)
+  else:
+    return int(umask)
+
+
 UPLOAD_CHUNK_SIZE = Config(
 UPLOAD_CHUNK_SIZE = Config(
   key="upload_chunk_size",
   key="upload_chunk_size",
   help="Size, in bytes, of the 'chunks' Django should store into memory and feed into the handler. Default is 64MB.",
   help="Size, in bytes, of the 'chunks' Django should store into memory and feed into the handler. Default is 64MB.",
@@ -69,6 +77,8 @@ HDFS_CLUSTERS = UnspecifiedConfigSection(
                               default=False, type=coerce_bool),
                               default=False, type=coerce_bool),
       TEMP_DIR=Config("temp_dir", help="HDFS directory for temporary files",
       TEMP_DIR=Config("temp_dir", help="HDFS directory for temporary files",
                       default='/tmp', type=str),
                       default='/tmp', type=str),
+      UMASK=Config("umask", help="Umask for creating files/directories in hdfs",
+                      default='1022', type=coerce_umask),
     )
     )
   )
   )
 )
 )

+ 20 - 0
desktop/libs/hadoop/src/hadoop/fs/test_webhdfs.py

@@ -135,6 +135,25 @@ class WebhdfsTests(unittest.TestCase):
     assert_raises(WebHdfsException, f.read)
     assert_raises(WebHdfsException, f.read)
     assert_raises(IOError, fs.open, "/test/doesnotexist.txt")
     assert_raises(IOError, fs.open, "/test/doesnotexist.txt")
 
 
+  
+  def test_umask(self):
+    fs = self.cluster.fs
+    try:
+      clear = UMASK.set_for_testing("0077")
+      test_dir = '/umask_test_dir'
+      fs.mkdir(test_dir)
+      test_file = '/umask_test.txt'
+      f.open(test_file, "w")
+      f.write("foo")
+      f.close()
+
+      # Check currrent permissions are not 777 (666 for file)
+      assert_equals(1700, fs.stats(test_dir).mode)
+      assert_equals(1700, fs.stats(test_file).mode)
+    finally:
+      clear()
+
+
   def test_copy_remote_dir(self):
   def test_copy_remote_dir(self):
     fs = self.cluster.fs
     fs = self.cluster.fs
 
 
@@ -265,6 +284,7 @@ class WebhdfsTests(unittest.TestCase):
     subdir1 = dir1 + '/test1'
     subdir1 = dir1 + '/test1'
     file1 = subdir1 + '/test1.txt'
     file1 = subdir1 + '/test1.txt'
     fs = self.cluster.fs
     fs = self.cluster.fs
+
     try:
     try:
       fs.mkdir(subdir1)
       fs.mkdir(subdir1)
       f = fs.open(file1, "w")
       f = fs.open(file1, "w")

+ 50 - 9
desktop/libs/hadoop/src/hadoop/fs/webhdfs.py

@@ -58,11 +58,13 @@ class WebHdfs(Hdfs):
                logical_name=None,
                logical_name=None,
                hdfs_superuser=None,
                hdfs_superuser=None,
                security_enabled=False,
                security_enabled=False,
-               temp_dir="/tmp"):
+               temp_dir="/tmp",
+               umask=1022):
     self._url = url
     self._url = url
     self._superuser = hdfs_superuser
     self._superuser = hdfs_superuser
     self._security_enabled = security_enabled
     self._security_enabled = security_enabled
     self._temp_dir = temp_dir
     self._temp_dir = temp_dir
+    self._umask = umask
     self._fs_defaultfs = fs_defaultfs
     self._fs_defaultfs = fs_defaultfs
     self._logical_name = logical_name
     self._logical_name = logical_name
 
 
@@ -82,7 +84,8 @@ class WebHdfs(Hdfs):
                fs_defaultfs=fs_defaultfs,
                fs_defaultfs=fs_defaultfs,
                logical_name=hdfs_config.LOGICAL_NAME.get(),
                logical_name=hdfs_config.LOGICAL_NAME.get(),
                security_enabled=hdfs_config.SECURITY_ENABLED.get(),
                security_enabled=hdfs_config.SECURITY_ENABLED.get(),
-               temp_dir=hdfs_config.TEMP_DIR.get())
+               temp_dir=hdfs_config.TEMP_DIR.get(),
+               umask=hdfs_config.UMASK.get())
 
 
   def __str__(self):
   def __str__(self):
     return "WebHdfs at %s" % self._url
     return "WebHdfs at %s" % self._url
@@ -106,6 +109,10 @@ class WebHdfs(Hdfs):
   def fs_defaultfs(self):
   def fs_defaultfs(self):
     return self._fs_defaultfs
     return self._fs_defaultfs
 
 
+  @property
+  def umask(self):
+    return self._umask
+
   @property
   @property
   def security_enabled(self):
   def security_enabled(self):
     return self._security_enabled
     return self._security_enabled
@@ -335,8 +342,14 @@ class WebHdfs(Hdfs):
     path = Hdfs.normpath(path)
     path = Hdfs.normpath(path)
     params = self._getparams()
     params = self._getparams()
     params['op'] = 'MKDIRS'
     params['op'] = 'MKDIRS'
-    if mode is not None:
+
+    if mode is None:
+      params['permission'] = self.getDefaultFilePerms()
+      LOG.debug("No permissions set, defaulting to umask: %s" % params['permission'])
+    else:
       params['permission'] = safe_octal(mode)
       params['permission'] = safe_octal(mode)
+      LOG.debug("Permissions set, using: %s" % params['permission'])
+
     success = self._root.put(path, params)
     success = self._root.put(path, params)
     if not success:
     if not success:
       raise IOError(_("Mkdir failed: %s") % path)
       raise IOError(_("Mkdir failed: %s") % path)
@@ -400,6 +413,7 @@ class WebHdfs(Hdfs):
     else:
     else:
       self._root.put(path, params)
       self._root.put(path, params)
 
 
+
   def get_home_dir(self):
   def get_home_dir(self):
     """get_home_dir() -> Home directory for the current user"""
     """get_home_dir() -> Home directory for the current user"""
     params = self._getparams()
     params = self._getparams()
@@ -440,6 +454,19 @@ class WebHdfs(Hdfs):
     return File(self, path, mode)
     return File(self, path, mode)
 
 
 
 
+  def getDefaultFilePerms(self):
+              umask = int(1777) - int(self.umask)
+ 
+              # Below we are making sure that we don't lose the 0 at the beginning
+              # of the permissions
+                
+              if len(str(umask)) < 4:
+                umask = "0" + str(umask)
+                return umask
+              else:
+                return umask
+
+
   def create(self, path, overwrite=False, blocksize=None,
   def create(self, path, overwrite=False, blocksize=None,
              replication=None, permission=None, data=None):
              replication=None, permission=None, data=None):
     """
     """
@@ -456,8 +483,12 @@ class WebHdfs(Hdfs):
       params['blocksize'] = long(blocksize)
       params['blocksize'] = long(blocksize)
     if replication is not None:
     if replication is not None:
       params['replication'] = int(replication)
       params['replication'] = int(replication)
-    if permission is not None:
+    if permission is None:
+      params['permission'] = self.getDefaultFilePerms()
+      LOG.debug("No permissions set, using umask: %s" % params['permission'])
+    else:
       params['permission'] = safe_octal(permission)
       params['permission'] = safe_octal(permission)
+      LOG.warn("Permissions already set, using: %s" % params['permission'])
 
 
     self._invoke_with_redirect('PUT', path, params, data)
     self._invoke_with_redirect('PUT', path, params, data)
 
 
@@ -555,11 +586,14 @@ class WebHdfs(Hdfs):
       offset += cnt
       offset += cnt
 
 
 
 
-  def copy_remote_dir(self, source, destination, dir_mode=0755, owner=None):
+  def copy_remote_dir(self, source, destination, dir_mode=None, owner=None):
     if owner is None:
     if owner is None:
       owner = self.DEFAULT_USER
       owner = self.DEFAULT_USER
+
+    if dir_mode is None:
+      dir_mode = self.getDefaultFilePerms()
+    LOG.debug("Making directory %s with permissions %s" % (destination, dir_mode))
     self.do_as_user(owner, self.mkdir, destination, mode=dir_mode)
     self.do_as_user(owner, self.mkdir, destination, mode=dir_mode)
-    self.do_as_user(owner, self.chmod, destination, mode=dir_mode) # To remove after HDFS-3491
 
 
     for stat in self.listdir_stats(source):
     for stat in self.listdir_stats(source):
       source_file = stat.path
       source_file = stat.path
@@ -571,7 +605,7 @@ class WebHdfs(Hdfs):
         self.do_as_superuser(self.chown, destination_file, owner, owner)
         self.do_as_superuser(self.chown, destination_file, owner, owner)
 
 
 
 
-  def copy(self, src, dest, recursive=False, dir_mode=0755, owner=None):
+  def copy(self, src, dest, recursive=False, dir_mode=None, owner=None):
     """
     """
     Copy file, or directory, in HDFS to another location in HDFS.
     Copy file, or directory, in HDFS to another location in HDFS.
 
 
@@ -591,6 +625,13 @@ class WebHdfs(Hdfs):
     if owner is None:
     if owner is None:
       owner = self.user
       owner = self.user
 
 
+    # Hue was defauling permissions on copying files to the permissions
+    # of the original file, but was not doing the same for directories
+    # changed below for directories to remain consistent
+    if dir_mode is None:
+      sb = self._stats(src)
+      dir_mode=oct(stat.S_IMODE(sb.mode))
+
     src = self.abspath(src)
     src = self.abspath(src)
     dest = self.abspath(dest)
     dest = self.abspath(dest)
 
 
@@ -612,8 +653,8 @@ class WebHdfs(Hdfs):
           dest = self.join(dest, self.basename(src))
           dest = self.join(dest, self.basename(src))
         else:
         else:
           raise IOError(errno.EEXIST, _("Destination file %s exists and is not a directory.") % dest)
           raise IOError(errno.EEXIST, _("Destination file %s exists and is not a directory.") % dest)
-      self.do_as_user(owner, self.mkdir, dest)
-      self.do_as_user(owner, self.chmod, dest, mode=dir_mode)
+      
+      self.do_as_user(owner, self.mkdir, dest, mode=dir_mode)
 
 
       # Copy files in 'src' directory to 'dest'.
       # Copy files in 'src' directory to 'dest'.
       self.copy_remote_dir(src, dest, dir_mode, owner)
       self.copy_remote_dir(src, dest, dir_mode, owner)