浏览代码

[fb] Upload to federated cluster

The location for the temporary file being uploaded should be
the destination of the final, renamed file. The temporary file
will be named <filename>.tmp.

No new tests were necessary since we are not changing upload
functionality. Simply changing where the temp files exist.

HACK: In order to get the destination when uploading in the
custom upload handler, needed to use a GET parameter. This
is due to the obscurity in Django's builtin POST data
parser. Basically, the 'dest' POST parameter is not accessible
until the file contents has been parsed.
abec 13 年之前
父节点
当前提交
eb38948ee2

+ 3 - 2
apps/filebrowser/src/filebrowser/templates/listdir_components.mako

@@ -916,9 +916,10 @@ from django.utils.translation import ugettext as _
 
 
             self.uploadFile = (function() {
             self.uploadFile = (function() {
                 var num_of_pending_uploads = 0;
                 var num_of_pending_uploads = 0;
+                var action = "/filebrowser/upload/file";
                 var uploader = new qq.FileUploader({
                 var uploader = new qq.FileUploader({
                     element: document.getElementById("fileUploader"),
                     element: document.getElementById("fileUploader"),
-                    action: "/filebrowser/upload/file",
+                    action: action,
                     template: '<div class="qq-uploader">' +
                     template: '<div class="qq-uploader">' +
                             '<div class="qq-upload-drop-area"><span>${_('Drop files here to upload')}</span></div>' +
                             '<div class="qq-upload-drop-area"><span>${_('Drop files here to upload')}</span></div>' +
                             '<div class="qq-upload-button">${_('Upload a file')}</div>' +
                             '<div class="qq-upload-button">${_('Upload a file')}</div>' +
@@ -947,7 +948,7 @@ from django.utils.translation import ugettext as _
                     debug: false
                     debug: false
                 });
                 });
 
 
-                $("#archiveUploader").on('fb:updatePath', function(e, options) {
+                $("#fileUploader").on('fb:updatePath', function(e, options) {
                     uploader.setParams({
                     uploader.setParams({
                         dest: options.dest,
                         dest: options.dest,
                         fileFieldLabel: "hdfs_file"
                         fileFieldLabel: "hdfs_file"

+ 2 - 1
desktop/core/static/ext/js/fileuploader.js

@@ -1231,7 +1231,8 @@ qq.extend(qq.UploadHandlerXhr.prototype, {
         formData.append(params.fileFieldLabel, file);
         formData.append(params.fileFieldLabel, file);
         formData.append('dest', params.dest);
         formData.append('dest', params.dest);
 
 
-        xhr.open("POST", this._options.action, true);
+        var action = this._options.action + "?dest=" + params.dest;
+        xhr.open("POST", action, true);
         xhr.send(formData);
         xhr.send(formData);
     },
     },
     _onComplete: function(id, xhr){
     _onComplete: function(id, xhr){

+ 32 - 18
desktop/libs/hadoop/src/hadoop/fs/hadoopfs.py

@@ -287,6 +287,38 @@ class Hdfs(object):
     else:
     else:
       LOG.info(_('Skipping %s (not a file).') % local_src)
       LOG.info(_('Skipping %s (not a file).') % local_src)
 
 
+  @_coerce_exceptions
+  def mktemp(self, subdir='', prefix='tmp', basedir=None):
+    """
+    mktemp(prefix) ->  <temp_dir or basedir>/<subdir>/prefix.<rand>
+    Return a unique temporary filename with prefix in the cluster's temp dir.
+    """
+    RANDOM_BITS = 64
+
+    base = self.join(basedir or self._temp_dir, subdir)
+    if not self.isdir(base):
+      self.mkdir(base)
+
+    while True:
+      name = prefix + '.' + str(random.getrandbits(RANDOM_BITS))
+      candidate = self.join(base, name)
+      if not self.exists(candidate):
+        return candidate
+
+  def mkswap(self, filename, subdir='', suffix='swp', basedir=None):
+    """
+    mkswap(filename, suffix) ->  <temp_dir or basedir>/<subdir>/filename.<suffix>
+    Return a unique temporary filename with prefix in the cluster's temp dir.
+    """
+    RANDOM_BITS = 64
+
+    base = self.join(basedir or self._temp_dir, subdir)
+    if not self.isdir(base):
+      self.mkdir(base)
+
+    candidate = self.join(base, "%s.%s" % (filename, suffix))
+    return candidate
+
   def exists(self):
   def exists(self):
     raise NotImplementedError(_("%(function)s has not been implemented.") % {'function': 'exists'})
     raise NotImplementedError(_("%(function)s has not been implemented.") % {'function': 'exists'})
 
 
@@ -548,24 +580,6 @@ class HadoopFileSystem(Hdfs):
     path = encode_fs_path(path)
     path = encode_fs_path(path)
     self.nn_client.chown(self.request_context, normpath(path), user, group)
     self.nn_client.chown(self.request_context, normpath(path), user, group)
 
 
-  @_coerce_exceptions
-  def mktemp(self, subdir='', prefix='tmp'):
-    """
-    mktemp(prefix) ->  <temp_dir>/subdir/prefix.<rand>
-    Return a unique temporary filename with prefix in the cluster's temp dir.
-    """
-    RANDOM_BITS = 64
-
-    base = self.join(self._temp_dir, subdir)
-    if not self.isdir(base):
-      self.mkdir(base)
-
-    while True:
-      name = prefix + '.' + str(random.getrandbits(RANDOM_BITS))
-      candidate = self.join(base, name)
-      if not self.exists(candidate):
-        return candidate
-
   @_coerce_exceptions
   @_coerce_exceptions
   def get_namenode_info(self):
   def get_namenode_info(self):
     (capacity, used, available) = self.nn_client.df(self.request_context)
     (capacity, used, available) = self.nn_client.df(self.request_context)

+ 4 - 7
desktop/libs/hadoop/src/hadoop/fs/upload.py

@@ -46,7 +46,7 @@ class HDFStemporaryUploadedFile(object):
   A temporary HDFS file to store upload data.
   A temporary HDFS file to store upload data.
   This class does not have any file read methods.
   This class does not have any file read methods.
   """
   """
-  def __init__(self, request, name):
+  def __init__(self, request, name, destination):
     self.name = name
     self.name = name
     self.size = None
     self.size = None
     self._do_cleanup = False
     self._do_cleanup = False
@@ -65,12 +65,8 @@ class HDFStemporaryUploadedFile(object):
     self._fs.setuser(self._fs.DEFAULT_USER)
     self._fs.setuser(self._fs.DEFAULT_USER)
     self._fs.setuser(self._fs.superuser)
     self._fs.setuser(self._fs.superuser)
 
 
-    self._path = self._fs.mktemp(
-        subdir='hue-uploads',
-        prefix='tmp.%s' % (request.environ['REMOTE_ADDR'],))
+    self._path = self._fs.mkswap(name, suffix='tmp', basedir=destination)
 
 
-    # Make the tmp dir 0777
-    self._fs.chmod(self._fs.dirname(self._path), 0777)
     self._file = self._fs.open(self._path, 'w')
     self._file = self._fs.open(self._path, 'w')
     self._do_cleanup = True
     self._do_cleanup = True
 
 
@@ -122,6 +118,7 @@ class HDFSfileUploadHandler(FileUploadHandler):
     self._file = None
     self._file = None
     self._starttime = 0
     self._starttime = 0
     self._activated = False
     self._activated = False
+    self._destination = request.GET.get('dest', None)
     # Need to directly modify FileUploadHandler.chunk_size
     # Need to directly modify FileUploadHandler.chunk_size
     FileUploadHandler.chunk_size = UPLOAD_CHUNK_SIZE.get()
     FileUploadHandler.chunk_size = UPLOAD_CHUNK_SIZE.get()
 
 
@@ -132,7 +129,7 @@ class HDFSfileUploadHandler(FileUploadHandler):
     #       running the auth middleware.
     #       running the auth middleware.
     if field_name.upper().startswith('HDFS'):
     if field_name.upper().startswith('HDFS'):
       try:
       try:
-        self._file = HDFStemporaryUploadedFile(self.request, file_name)
+        self._file = HDFStemporaryUploadedFile(self.request, file_name, self._destination)
       except (HDFSerror, IOError), ex:
       except (HDFSerror, IOError), ex:
         LOG.error("Not using HDFS upload handler: %s" % (ex,))
         LOG.error("Not using HDFS upload handler: %s" % (ex,))
         return
         return

+ 0 - 17
desktop/libs/hadoop/src/hadoop/fs/webhdfs.py

@@ -248,23 +248,6 @@ class WebHdfs(Hdfs):
     if not success:
     if not success:
       raise IOError("Mkdir failed: %s" % (smart_str(path),))
       raise IOError("Mkdir failed: %s" % (smart_str(path),))
 
 
-  def mktemp(self, subdir='', prefix='tmp'):
-    """
-    mktemp(subdir, prefix) ->  <temp_dir>/subdir/prefix.<rand>
-    Return a unique temporary filename with prefix in the cluster's temp dir.
-    """
-    RANDOM_BITS = 64
-
-    base = self.join(self._temp_dir, subdir)
-    if not self.isdir(base):
-      self.mkdir(base)
-
-    while True:
-      name = "%s.%s" % (prefix, random.getrandbits(RANDOM_BITS))
-      candidate = self.join(base, name)
-      if not self.exists(candidate):
-        return candidate
-
   def rename(self, old, new):
   def rename(self, old, new):
     """rename(old, new)"""
     """rename(old, new)"""
     old = Hdfs.normpath(old)
     old = Hdfs.normpath(old)