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

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HIVE/package/scripts/hive_server.py

@@ -20,9 +20,9 @@ limitations under the License.
 
 import sys
 from resource_management import *
-
 from hive import hive
 from hive_service import hive_service
+import os
 
 class HiveServer(Script):
 
@@ -39,6 +39,7 @@ class HiveServer(Script):
     import params
     env.set_params(params)
     self.configure(env) # FOR SECURITY
+    self.install_tez_jars(params) # Put tez jars in hdfs
     hive_service( 'hiveserver2',
                   action = 'start'
     )
@@ -51,7 +52,6 @@ class HiveServer(Script):
                   action = 'stop'
     )
 
-
   def status(self, env):
     import status_params
     env.set_params(status_params)
@@ -59,5 +59,126 @@ class HiveServer(Script):
     # Recursively check all existing gmetad pid files
     check_process_status(pid_file)
 
+  def install_tez_jars(self, params):
+    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.tez_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.tez_stub_path
+        )
+      pass
+
+      ExecuteHadoop(format('dfs -touchz {tez_stub_path}'),
+                    user = params.tez_user,
+                    conf_dir=params.hadoop_conf_dir
+      )
+
+
+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.tez_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)
+  pass
+
+  if not mode:
+    chmod_cmd = format('fs -chmod {mode} {dest_dir}')
+
+    ExecuteHadoop(chmod_cmd,
+                  user=owner,
+                  conf_dir=params.hadoop_conf_dir)
+  pass
+
+
 if __name__ == "__main__":
   HiveServer().execute()

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HIVE/package/scripts/params.py

@@ -150,6 +150,14 @@ hadoop_conf_dir = "/etc/hadoop/conf"
 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"])
+
+# Tez libraries
+tez_lib_uris = default("/configurations/tez-site/tez.lib.uris", None)
+tez_local_api_jars = '/usr/lib/tez/tez*.jar'
+tez_local_lib_jars = '/usr/lib/tez/lib/*.jar'
+tez_stub_path = '/tmp/tez_jars_copied'
+tez_user = 'tez'
+
 import functools
 #create partial functions with common arguments for every HdfsDirectory call
 #to create hdfs directory we need to call params.HdfsDirectory in code

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

@@ -29,29 +29,4 @@ hadoop_home = '/usr'
 java64_home = config['hostLevelParams']['java_home']
 
 tez_user = 'tez'
-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
-)
+user_group = config['configurations']['global']['user_group']

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

@@ -21,7 +21,6 @@ Ambari Agent
 """
 
 from resource_management import *
-import os
 
 def tez():
   import params
@@ -42,68 +41,6 @@ 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
@@ -116,56 +53,3 @@ def tez_TemplateConfig(name):
         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/test/python/stacks/2.0.6/HIVE/test_hive_server.py

@@ -71,6 +71,56 @@ class TestHiveServer(RMFTestCase):
                               owner='hive',
                               group='hadoop'
     )
+
+    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.assertResourceCalled('Execute', 'env JAVA_HOME=/usr/jdk64/jdk1.7.0_45 /tmp/start_hiveserver2_script /var/log/hive/hive-server2.out /var/log/hive/hive-server2.log /var/run/hive/hive-server.pid /etc/hive/conf.server',
                               not_if = 'ls /var/run/hive/hive-server.pid >/dev/null 2>&1 && ps `cat /var/run/hive/hive-server.pid` >/dev/null 2>&1',
                               user = 'hive'

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

@@ -486,7 +486,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"
         },

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

@@ -47,55 +47,6 @@ 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()