Browse Source

HUE-3838 [oozie] Deprecate jar_path copy and use oozie.libpath instead for actions with jars

Romain Rigaux 9 years ago
parent
commit
368e6a6

+ 7 - 0
desktop/libs/liboozie/src/liboozie/conf.py

@@ -50,6 +50,13 @@ SSL_CERT_CA_VERIFY=Config(
   dynamic_default=default_ssl_validate,
   type=coerce_bool)
 
+USE_LIBPATH_FOR_JARS = Config(
+  key="use_libpath_for_jars",
+  help=_t("Whether Hue append jar paths to the oozie.libpath instead of copying them into the workspace."
+          " This makes submissions faster and less prone to HDFS permission errors"),
+  default=True,
+  type=coerce_bool)
+
 
 def get_oozie_status(user):
   from liboozie.oozie_api import get_oozie

+ 22 - 13
desktop/libs/liboozie/src/liboozie/submission2.py

@@ -32,7 +32,7 @@ from hadoop.fs.hadoopfs import Hdfs
 from oozie.utils import convert_to_server_timezone
 
 from liboozie.oozie_api import get_oozie
-from liboozie.conf import REMOTE_DEPLOYMENT_DIR
+from liboozie.conf import REMOTE_DEPLOYMENT_DIR, USE_LIBPATH_FOR_JARS
 from liboozie.credentials import Credentials
 
 
@@ -323,18 +323,27 @@ class Submission(object):
           if not jar_path.startswith(lib_path): # If not already in lib
             files.append(jar_path)
 
-    # Copy the jar files to the workspace lib
-    if files:
-      for jar_file in files:
-        LOG.debug("Updating %s" % jar_file)
-        jar_lib_path = self.fs.join(lib_path, self.fs.basename(jar_file))
-        # Refresh if needed
-        if self.fs.exists(jar_lib_path) and self.fs.exists(jar_file):
-          stat_src = self.fs.stats(jar_file)
-          stat_dest = self.fs.stats(jar_lib_path)
-          if stat_src.fileId != stat_dest.fileId:
-            self.fs.remove(jar_lib_path, skip_trash=True)
-        self.fs.copyfile(jar_file, jar_lib_path)
+    if USE_LIBPATH_FOR_JARS.get():
+      # Add the jar files to the oozie.libpath
+      if files:
+        files = list(set(files))
+        LOG.debug("Adding to oozie.libpath %s" % files)
+        if self.properties.get('oozie.libpath'):
+          files.append(self.properties['oozie.libpath'])
+        self.properties['oozie.libpath'] = ','.join(files)
+    else:
+      # Copy the jar files to the workspace lib
+      if files:
+        for jar_file in files:
+          LOG.debug("Updating %s" % jar_file)
+          jar_lib_path = self.fs.join(lib_path, self.fs.basename(jar_file))
+          # Refresh if needed
+          if self.fs.exists(jar_lib_path) and self.fs.exists(jar_file):
+            stat_src = self.fs.stats(jar_file)
+            stat_dest = self.fs.stats(jar_lib_path)
+            if stat_src.fileId != stat_dest.fileId:
+              self.fs.remove(jar_lib_path, skip_trash=True)
+          self.fs.copyfile(jar_file, jar_lib_path)
 
   def _do_as(self, username, fn, *args, **kwargs):
     prev_user = self.fs.user

+ 43 - 33
desktop/libs/liboozie/src/liboozie/submittion2_tests.py

@@ -29,6 +29,7 @@ from desktop.lib.django_test_util import make_logged_in_client
 from oozie.models2 import Node
 from oozie.tests import OozieMockBase
 
+from liboozie.conf import USE_LIBPATH_FOR_JARS
 from liboozie.submission2 import Submission
 
 
@@ -102,39 +103,48 @@ def test_copy_files():
     deployment_dir = deployment_dir + '/lib'
     external_deployment_dir = external_deployment_dir + '/lib'
 
-    list_dir_workspace = cluster.fs.listdir(deployment_dir)
-    list_dir_deployement = cluster.fs.listdir(external_deployment_dir)
-
-    # All destinations there
-    assert_true(cluster.fs.exists(deployment_dir + '/udf1.jar'), list_dir_workspace)
-    assert_true(cluster.fs.exists(deployment_dir + '/udf2.jar'), list_dir_workspace)
-    assert_true(cluster.fs.exists(deployment_dir + '/udf3.jar'), list_dir_workspace)
-    assert_true(cluster.fs.exists(deployment_dir + '/udf4.jar'), list_dir_workspace)
-    assert_true(cluster.fs.exists(deployment_dir + '/udf5.jar'), list_dir_workspace)
-    assert_true(cluster.fs.exists(deployment_dir + '/udf6.jar'), list_dir_workspace)
-
-    assert_true(cluster.fs.exists(external_deployment_dir + '/udf1.jar'), list_dir_deployement)
-    assert_true(cluster.fs.exists(external_deployment_dir + '/udf2.jar'), list_dir_deployement)
-    assert_true(cluster.fs.exists(external_deployment_dir + '/udf3.jar'), list_dir_deployement)
-    assert_true(cluster.fs.exists(external_deployment_dir + '/udf4.jar'), list_dir_deployement)
-    assert_true(cluster.fs.exists(external_deployment_dir + '/udf5.jar'), list_dir_deployement)
-    assert_true(cluster.fs.exists(external_deployment_dir + '/udf6.jar'), list_dir_deployement)
-
-    stats_udf1 = cluster.fs.stats(deployment_dir + '/udf1.jar')
-    stats_udf2 = cluster.fs.stats(deployment_dir + '/udf2.jar')
-    stats_udf3 = cluster.fs.stats(deployment_dir + '/udf3.jar')
-    stats_udf4 = cluster.fs.stats(deployment_dir + '/udf4.jar')
-    stats_udf5 = cluster.fs.stats(deployment_dir + '/udf5.jar')
-    stats_udf6 = cluster.fs.stats(deployment_dir + '/udf6.jar')
-
-    submission._copy_files('%s/workspace' % prefix, "<xml>My XML</xml>", {'prop1': 'val1'})
-
-    assert_not_equal(stats_udf1['fileId'], cluster.fs.stats(deployment_dir + '/udf1.jar')['fileId'])
-    assert_not_equal(stats_udf2['fileId'], cluster.fs.stats(deployment_dir + '/udf2.jar')['fileId'])
-    assert_not_equal(stats_udf3['fileId'], cluster.fs.stats(deployment_dir + '/udf3.jar')['fileId'])
-    assert_equal(stats_udf4['fileId'], cluster.fs.stats(deployment_dir + '/udf4.jar')['fileId'])
-    assert_not_equal(stats_udf5['fileId'], cluster.fs.stats(deployment_dir + '/udf5.jar')['fileId'])
-    assert_equal(stats_udf6['fileId'], cluster.fs.stats(deployment_dir + '/udf6.jar')['fileId'])
+    if USE_LIBPATH_FOR_JARS.get():
+      assert_true(jar_1 in submission.properties['oozie.libpath'])
+      assert_true(jar_2 in submission.properties['oozie.libpath'])
+      assert_true(jar_3 in submission.properties['oozie.libpath'])
+      assert_true(jar_4 in submission.properties['oozie.libpath'])
+      print deployment_dir + '/' + jar_5
+      assert_true((deployment_dir + '/' + jar_5) in submission.properties['oozie.libpath'], submission.properties['oozie.libpath'])
+      assert_true((deployment_dir + '/' + jar_6) in submission.properties['oozie.libpath'], submission.properties['oozie.libpath'])
+    else:
+      list_dir_workspace = cluster.fs.listdir(deployment_dir)
+      list_dir_deployement = cluster.fs.listdir(external_deployment_dir)
+  
+      # All destinations there
+      assert_true(cluster.fs.exists(deployment_dir + '/udf1.jar'), list_dir_workspace)
+      assert_true(cluster.fs.exists(deployment_dir + '/udf2.jar'), list_dir_workspace)
+      assert_true(cluster.fs.exists(deployment_dir + '/udf3.jar'), list_dir_workspace)
+      assert_true(cluster.fs.exists(deployment_dir + '/udf4.jar'), list_dir_workspace)
+      assert_true(cluster.fs.exists(deployment_dir + '/udf5.jar'), list_dir_workspace)
+      assert_true(cluster.fs.exists(deployment_dir + '/udf6.jar'), list_dir_workspace)
+  
+      assert_true(cluster.fs.exists(external_deployment_dir + '/udf1.jar'), list_dir_deployement)
+      assert_true(cluster.fs.exists(external_deployment_dir + '/udf2.jar'), list_dir_deployement)
+      assert_true(cluster.fs.exists(external_deployment_dir + '/udf3.jar'), list_dir_deployement)
+      assert_true(cluster.fs.exists(external_deployment_dir + '/udf4.jar'), list_dir_deployement)
+      assert_true(cluster.fs.exists(external_deployment_dir + '/udf5.jar'), list_dir_deployement)
+      assert_true(cluster.fs.exists(external_deployment_dir + '/udf6.jar'), list_dir_deployement)
+  
+      stats_udf1 = cluster.fs.stats(deployment_dir + '/udf1.jar')
+      stats_udf2 = cluster.fs.stats(deployment_dir + '/udf2.jar')
+      stats_udf3 = cluster.fs.stats(deployment_dir + '/udf3.jar')
+      stats_udf4 = cluster.fs.stats(deployment_dir + '/udf4.jar')
+      stats_udf5 = cluster.fs.stats(deployment_dir + '/udf5.jar')
+      stats_udf6 = cluster.fs.stats(deployment_dir + '/udf6.jar')
+  
+      submission._copy_files('%s/workspace' % prefix, "<xml>My XML</xml>", {'prop1': 'val1'})
+  
+      assert_not_equal(stats_udf1['fileId'], cluster.fs.stats(deployment_dir + '/udf1.jar')['fileId'])
+      assert_not_equal(stats_udf2['fileId'], cluster.fs.stats(deployment_dir + '/udf2.jar')['fileId'])
+      assert_not_equal(stats_udf3['fileId'], cluster.fs.stats(deployment_dir + '/udf3.jar')['fileId'])
+      assert_equal(stats_udf4['fileId'], cluster.fs.stats(deployment_dir + '/udf4.jar')['fileId'])
+      assert_not_equal(stats_udf5['fileId'], cluster.fs.stats(deployment_dir + '/udf5.jar')['fileId'])
+      assert_equal(stats_udf6['fileId'], cluster.fs.stats(deployment_dir + '/udf6.jar')['fileId'])
 
     # Test _create_file()
     submission._create_file(deployment_dir, 'test.txt', data='Test data')