Browse Source

HUE-7248 [adls] fix file upload over 30mb, remove logging auth token

jdesjean 8 years ago
parent
commit
46441170ae

+ 3 - 0
desktop/core/src/desktop/lib/fs/proxyfs.py

@@ -235,3 +235,6 @@ class ProxyFS(object):
 
 
   def mkswap(self, filename, subdir='', suffix='swp', basedir=None):
   def mkswap(self, filename, subdir='', suffix='swp', basedir=None):
     return self._get_fs(basedir).mkswap(filename, subdir, suffix, basedir)
     return self._get_fs(basedir).mkswap(filename, subdir, suffix, basedir)
+
+  def get_upload_chuck_size(self, path):
+    return self._get_fs(path).get_upload_chuck_size()

+ 4 - 4
desktop/core/src/desktop/lib/rest/resource.py

@@ -63,7 +63,7 @@ class Resource(object):
     else:
     else:
       return resp.content
       return resp.content
 
 
-  def invoke(self, method, relpath=None, params=None, data=None, headers=None, files=None, allow_redirects=False, clear_cookies=False):
+  def invoke(self, method, relpath=None, params=None, data=None, headers=None, files=None, allow_redirects=False, clear_cookies=False, log_response=True):
     """
     """
     Invoke an API method.
     Invoke an API method.
     @return: Raw body or JSON dictionary (if response content type is JSON).
     @return: Raw body or JSON dictionary (if response content type is JSON).
@@ -79,7 +79,7 @@ class Resource(object):
                                 urlencode=self._urlencode,
                                 urlencode=self._urlencode,
                                 clear_cookies=clear_cookies)
                                 clear_cookies=clear_cookies)
 
 
-    if self._client.logger.isEnabledFor(logging.DEBUG):
+    if log_response and self._client.logger.isEnabledFor(logging.DEBUG):
       self._client.logger.debug(
       self._client.logger.debug(
           "%s Got response: %s%s" %
           "%s Got response: %s%s" %
           (method,
           (method,
@@ -114,7 +114,7 @@ class Resource(object):
 
 
 
 
   def post(self, relpath=None, params=None, data=None, contenttype=None, headers=None, files=None, allow_redirects=False,
   def post(self, relpath=None, params=None, data=None, contenttype=None, headers=None, files=None, allow_redirects=False,
-           clear_cookies=False):
+           clear_cookies=False, log_response=True):
     """
     """
     Invoke the POST method on a resource.
     Invoke the POST method on a resource.
     @param relpath: Optional. A relative path to this resource's path.
     @param relpath: Optional. A relative path to this resource's path.
@@ -128,7 +128,7 @@ class Resource(object):
     @return: A dictionary of the JSON result.
     @return: A dictionary of the JSON result.
     """
     """
     return self.invoke("POST", relpath, params, data, headers=self._make_headers(contenttype, headers), files=files,
     return self.invoke("POST", relpath, params, data, headers=self._make_headers(contenttype, headers), files=files,
-                       allow_redirects=allow_redirects, clear_cookies=clear_cookies)
+                       allow_redirects=allow_redirects, clear_cookies=clear_cookies, log_response=log_response)
 
 
 
 
   def put(self, relpath=None, params=None, data=None, contenttype=None, allow_redirects=False, clear_cookies=False, headers=None):
   def put(self, relpath=None, params=None, data=None, contenttype=None, allow_redirects=False, clear_cookies=False, headers=None):

+ 1 - 1
desktop/libs/azure/src/azure/active_directory.py

@@ -43,7 +43,7 @@ class ActiveDirectory(object):
         "client_id" : self._access_key_id,
         "client_id" : self._access_key_id,
         "client_secret" : self._secret_access_key
         "client_secret" : self._secret_access_key
       }
       }
-      self._token = self._root.post("/", data=data);
+      self._token = self._root.post("/", data=data, log_response=False);
       self._token["expires_on"] = int(self._token["expires_on"])
       self._token["expires_on"] = int(self._token["expires_on"])
 
 
     return self._token["token_type"] + " " + self._token["access_token"]
     return self._token["token_type"] + " " + self._token["access_token"]

+ 6 - 1
desktop/libs/azure/src/azure/adls/webhdfs.py

@@ -30,6 +30,8 @@ from urlparse import urlparse
 
 
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
 
 
+#Azure has a 30MB block limit on upload.
+UPLOAD_CHUCK_SIZE = 30 * 1000 * 1000
 
 
 class WebHdfs(HadoopWebHdfs):
 class WebHdfs(HadoopWebHdfs):
   def __init__(self, url,
   def __init__(self, url,
@@ -83,4 +85,7 @@ class WebHdfs(HadoopWebHdfs):
   def _getheaders(self):
   def _getheaders(self):
     return {
     return {
       "Authorization": self._auth_provider.get_token(),
       "Authorization": self._auth_provider.get_token(),
-    }
+    }
+
+  def get_upload_chuck_size(self):
+    return UPLOAD_CHUCK_SIZE

+ 4 - 3
desktop/libs/azure/src/azure/conf.py

@@ -17,7 +17,6 @@ from __future__ import absolute_import
 
 
 import logging
 import logging
 import re
 import re
-import azure
 
 
 from django.utils.translation import ugettext_lazy as _, ugettext as _t
 from django.utils.translation import ugettext_lazy as _, ugettext as _t
 from desktop.lib.conf import Config, UnspecifiedConfigSection, ConfigSection, coerce_bool, coerce_password_from_script
 from desktop.lib.conf import Config, UnspecifiedConfigSection, ConfigSection, coerce_bool, coerce_password_from_script
@@ -100,10 +99,12 @@ def has_adls_access(user):
 def config_validator(user):
 def config_validator(user):
   res = []
   res = []
 
 
+  import azure.client # Avoid cyclic loop
+
   if is_adls_enabled():
   if is_adls_enabled():
     try:
     try:
-      headers = azure.get_client('default')._getheaders()
-      if len(headers['authorization']) <= 0:
+      headers = azure.client.get_client('default')._getheaders()
+      if not headers.get('Authorization'):
         raise ValueError('Failed to obtain Azure authorization token')
         raise ValueError('Failed to obtain Azure authorization token')
     except Exception, e:
     except Exception, e:
       LOG.exception('Failed to obtain Azure authorization token.')
       LOG.exception('Failed to obtain Azure authorization token.')

+ 2 - 1
desktop/libs/hadoop/src/hadoop/fs/upload.py

@@ -142,7 +142,8 @@ class HDFSfileUploadHandler(FileUploadHandler):
     self._activated = False
     self._activated = False
     self._destination = request.GET.get('dest', None) # GET param avoids infinite looping
     self._destination = request.GET.get('dest', None) # GET param avoids infinite looping
     self.request = request
     self.request = request
-    FileUploadHandler.chunk_size = UPLOAD_CHUNK_SIZE.get()
+    fs = fsmanager.get_filesystem('default')
+    FileUploadHandler.chunk_size = fs.get_upload_chuck_size(self._destination) if self._destination else UPLOAD_CHUNK_SIZE.get()
 
 
     LOG.debug("Chunk size = %d" % FileUploadHandler.chunk_size)
     LOG.debug("Chunk size = %d" % FileUploadHandler.chunk_size)
 
 

+ 4 - 2
desktop/libs/hadoop/src/hadoop/fs/webhdfs.py

@@ -662,6 +662,8 @@ class WebHdfs(Hdfs):
       else:
       else:
         raise ex
         raise ex
 
 
+  def get_upload_chuck_size(self):
+    return UPLOAD_CHUNK_SIZE.get()
 
 
   def copyfile(self, src, dst, skip_header=False):
   def copyfile(self, src, dst, skip_header=False):
     sb = self._stats(src)
     sb = self._stats(src)
@@ -675,7 +677,7 @@ class WebHdfs(Hdfs):
     offset = 0
     offset = 0
 
 
     while True:
     while True:
-      data = self.read(src, offset, UPLOAD_CHUNK_SIZE.get())
+      data = self.read(src, offset, self.get_upload_chuck_size())
 
 
       if skip_header:
       if skip_header:
         data = '\n'.join(data.splitlines())
         data = '\n'.join(data.splitlines())
@@ -696,7 +698,7 @@ class WebHdfs(Hdfs):
       else:
       else:
         self.append(dst, data)
         self.append(dst, data)
 
 
-      if cnt < UPLOAD_CHUNK_SIZE.get():
+      if cnt < self.get_upload_chuck_size():
         break
         break
 
 
       offset += cnt
       offset += cnt