AMBARI-4754. Upload tez libraries to HDFS on Tez install. (swagle)

ambari-agent/src/main/python/resource_management/libraries/script/config_dictionary.py

@@ -72,7 +72,9 @@ class UnknownConfiguration():
     self.name = name
    
   def __getattr__(self, name):
-    raise Fail("Configuration parameter '"+self.name+"' was not found in configurations dictionary!")
+#    raise Fail("Configuration parameter '"+self.name+"' was not found in
+# configurations dictionary!")
+    pass
   
   def __getitem__(self, name):
     """

ambari-server/src/main/resources/stacks/HDP/2.1.1/services/TEZ/configuration/tez-site.xml

@@ -21,7 +21,7 @@
 
   <property>
     <name>tez.lib.uris</name>
-    <value>file:///usr/lib/tez/,file:///usr/lib/tez/lib/</value>
+    <value>hdfs:///apps/tez/,hdfs:///apps/tez/lib/</value>
     <description>The location of the Tez libraries which will be localized for DAGs</description>
   </property>
 

ambari-server/src/main/resources/stacks/HDP/2.1.1/services/TEZ/package/scripts/params.py

@@ -23,10 +23,35 @@ from resource_management import *
 # server configurations
 config = Script.get_config()
 
-conf_dir = "/etc/tez/conf"
+config_dir = "/etc/tez/conf"
 
 hadoop_home = '/usr'
 java64_home = config['hostLevelParams']['java_home']
 
 tez_user = 'tez'
-user_group = config['configurations']['global']['user_group']
+user_group = config['configurations']['global']['user_group']
+
+tez_lib_uris = config['configurations']['tez-site']['tez.lib.uris']
+tez_local_api_jars = '/usr/lib/tez/tez*.jar'
+tez_local_lib_jars = '/usr/lib/tez/lib/*.jar'
+
+stub_path = '/tmp/tez_jars_copied'
+
+hadoop_conf_dir = "/etc/hadoop/conf"
+_authentication = config['configurations']['core-site']['hadoop.security.authentication']
+security_enabled = ( not is_empty(_authentication) and _authentication == 'kerberos')
+hdfs_user_keytab = config['configurations']['global']['hdfs_user_keytab']
+hdfs_user = config['configurations']['global']['hdfs_user']
+kinit_path_local = functions.get_kinit_path([default("kinit_path_local",None), "/usr/bin", "/usr/kerberos/bin", "/usr/sbin"])
+
+import functools
+#create partial functions with common arguments for every HdfsDirectory call
+#to create hdfs directory we need to call params.HdfsDirectory in code
+HdfsDirectory = functools.partial(
+  HdfsDirectory,
+  conf_dir=hadoop_conf_dir,
+  hdfs_user=hdfs_user,
+  security_enabled = security_enabled,
+  keytab = hdfs_user_keytab,
+  kinit_path_local = kinit_path_local
+)

ambari-server/src/main/resources/stacks/HDP/2.1.1/services/TEZ/package/scripts/tez.py

@@ -21,18 +21,19 @@ Ambari Agent
 """
 
 from resource_management import *
+import os
 
 def tez():
   import params
 
-  Directory( params.conf_dir,
+  Directory(params.config_dir,
     owner = params.tez_user,
     group = params.user_group,
     recursive = True
   )
 
   XmlConfig( "tez-site.xml",
-            conf_dir = params.conf_dir,
+            conf_dir = params.config_dir,
             configurations = params.config['configurations']['tez-site'],
             owner = params.tez_user,
             group = params.user_group,
@@ -41,6 +42,68 @@ def tez():
 
   tez_TemplateConfig( ['tez-env.sh'])
 
+  destination_hdfs_dirs = get_tez_hdfs_dir_paths(params.tez_lib_uris)
+
+  # If tez libraries are to be stored in hdfs
+  if destination_hdfs_dirs:
+    for hdfs_dir in destination_hdfs_dirs:
+      params.HdfsDirectory(hdfs_dir,
+                          action="create_delayed",
+                          owner=params.tez_user,
+                          mode=0755
+      )
+    pass
+    params.HdfsDirectory(None, action="create")
+
+    if params.security_enabled:
+      kinit_if_needed = format("{kinit_path_local} -kt {hdfs_user_keytab} {hdfs_user};")
+    else:
+      kinit_if_needed = ""
+
+    if kinit_if_needed:
+      Execute(kinit_if_needed,
+              user=params.tez_user,
+              path='/bin'
+      )
+
+    app_dir_path = None
+    lib_dir_path = None
+
+    if len(destination_hdfs_dirs) > 1:
+      for path in destination_hdfs_dirs:
+        if 'lib' in path:
+          lib_dir_path = path
+        else:
+          app_dir_path = path
+        pass
+      pass
+    pass
+
+    if app_dir_path:
+      copyFromLocal(path=params.tez_local_api_jars,
+                    mode=0655,
+                    owner=params.tez_user,
+                    dest_dir=app_dir_path,
+                    kinnit_if_needed=kinit_if_needed,
+                    stub_path=params.stub_path
+      )
+    pass
+
+    if lib_dir_path:
+      copyFromLocal(path=params.tez_local_lib_jars,
+                    mode=0655,
+                    owner=params.tez_user,
+                    dest_dir=lib_dir_path,
+                    kinnit_if_needed=kinit_if_needed,
+                    stub_path=params.stub_path
+      )
+    pass
+
+    ExecuteHadoop(format('dfs -touchz {stub_path}'),
+                  user = params.tez_user,
+                  conf_dir=params.hadoop_conf_dir
+    )
+
 
 def tez_TemplateConfig(name):
   import params
@@ -49,6 +112,60 @@ def tez_TemplateConfig(name):
     name = [name]
 
   for x in name:
-    TemplateConfig( format("{conf_dir}/{x}"),
+    TemplateConfig(format("{config_dir}/{x}"),
         owner = params.tez_user
     )
+
+def get_tez_hdfs_dir_paths(tez_lib_uris = None):
+  hdfs_path_prefix = 'hdfs://'
+  lib_dir_paths = []
+  if tez_lib_uris and tez_lib_uris.strip().find(hdfs_path_prefix, 0) != -1:
+    dir_paths = tez_lib_uris.split(',')
+    for path in dir_paths:
+      lib_dir_path = path.replace(hdfs_path_prefix, '')
+      lib_dir_path = lib_dir_path if lib_dir_path.endswith(os.sep) else lib_dir_path + os.sep
+      lib_dir_paths.append(lib_dir_path)
+    pass
+  pass
+
+  return lib_dir_paths
+
+
+def copyFromLocal(path=None, owner=None, group=None, mode=None,
+                  dest_dir=None, kinnit_if_needed="", stub_path=None):
+  import params
+
+  if not stub_path:
+    stub_path = params.stub_path
+  pass
+
+  copy_cmd = format("fs -copyFromLocal {path} {dest_dir}")
+  unless_cmd = format("{kinnit_if_needed} hadoop fs -ls {stub_path} >/dev/null 2>&1")
+
+  ExecuteHadoop(copy_cmd,
+                not_if=unless_cmd,
+                user=owner,
+                conf_dir=params.hadoop_conf_dir,
+                ignore_failures = True)
+
+  if not owner:
+    chown = None
+  else:
+    if not group:
+      chown = owner
+    else:
+      chown = format('{owner}:{group}')
+
+  if not chown:
+    chown_cmd = format("fs -chown {chown} {dest_dir}")
+
+    ExecuteHadoop(copy_cmd,
+                  user=owner,
+                  conf_dir=params.hadoop_conf_dir)
+
+  if not mode:
+    chmod_cmd = format('fs -chmod {mode} {dest_dir}')
+
+    ExecuteHadoop(chmod_cmd,
+                  user=owner,
+                  conf_dir=params.hadoop_conf_dir)

ambari-server/src/main/resources/stacks/HDP/2.1.1/services/TEZ/package/templates/tez-env.sh.j2

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 # Tez specific configuration
-export TEZ_CONF_DIR={{conf_dir}}
+export TEZ_CONF_DIR={{config_dir}}
 
 # Set HADOOP_HOME to point to a specific hadoop install directory
 export HADOOP_HOME=${HADOOP_HOME:-{{hadoop_home}}}

ambari-server/src/test/python/stacks/2.1.1/TEZ/test_tez_client.py

@@ -22,7 +22,7 @@ from stacks.utils.RMFTestCase import *
 
 class TestTezClient(RMFTestCase):
 
-  def test_configure_defualt(self):
+  def test_configure_default(self):
     self.executeScript("2.1.1/services/TEZ/package/scripts/tez_client.py",
                        classname = "TezClient",
                        command = "configure",
@@ -47,6 +47,55 @@ class TestTezClient(RMFTestCase):
       owner = 'tez'
     )
 
+    self.assertResourceCalled('HdfsDirectory', '/apps/tez/',
+                              action = ['create_delayed'],
+                              mode = 0755,
+                              owner = 'tez',
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = "/usr/bin/kinit"
+    )
+
+    self.assertResourceCalled('HdfsDirectory', '/apps/tez/lib/',
+                              action = ['create_delayed'],
+                              mode = 0755,
+                              owner = 'tez',
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = "/usr/bin/kinit"
+    )
+    self.assertResourceCalled('HdfsDirectory', None,
+                              security_enabled = False,
+                              keytab = UnknownConfigurationMock(),
+                              conf_dir = '/etc/hadoop/conf',
+                              hdfs_user = 'hdfs',
+                              kinit_path_local = '/usr/bin/kinit',
+                              action = ['create']
+                              )
+
+    self.assertResourceCalled('ExecuteHadoop', 'fs -copyFromLocal /usr/lib/tez/tez*.jar /apps/tez/',
+                              not_if = ' hadoop fs -ls /tmp/tez_jars_copied >/dev/null 2>&1',
+                              user = 'tez',
+                              conf_dir = '/etc/hadoop/conf',
+                              ignore_failures=True
+    )
+
+    self.assertResourceCalled('ExecuteHadoop', 'fs -copyFromLocal /usr/lib/tez/lib/*.jar /apps/tez/lib/',
+                              not_if = ' hadoop fs -ls /tmp/tez_jars_copied >/dev/null 2>&1',
+                              user = 'tez',
+                              conf_dir = '/etc/hadoop/conf',
+                              ignore_failures=True
+    )
+
+    self.assertResourceCalled('ExecuteHadoop', 'dfs -touchz /tmp/tez_jars_copied',
+                              user = 'tez',
+                              conf_dir = '/etc/hadoop/conf'
+    )
+
     self.assertNoMoreResources()
 
 

ambari-server/src/test/python/stacks/2.1.1/configs/default.json

@@ -485,7 +485,7 @@
         },
         "tez-site": {
             "tez.am.log.level": "WARN",
-            "tez.lib.uris": "file:///usr/lib/tez/,file:///usr/lib/tez/lib/",
+            "tez.lib.uris": "hdfs:///apps/tez/,hdfs:///apps/tez/lib/",
             "tez.staging-dir": "/tmp/${user.name}/staging",
             "tez.am.am-rm.heartbeat.interval-ms.max": "250"
         },