Merge pull request #40 from pndaproject/RELEASE_0.5.0

Release 0.5.0

Author: stephanesan

CHANGELOG.md

@@ -1,10 +1,18 @@
 # Change Log
 All notable changes to this project will be documented in this file.
 
+## [Unreleased]
+
+## [0.5.0] 2017-11-24
+### Added:
+- PNDA-3330: Change to use a default application user instead of the hdfs user.
+- PNDA-2445: Support for Hortonworks HDP
+
 ## [0.4.0] 2017-05-23
 ### Added
 - PNDA-2729: Added support for spark streaming jobs written in python (pyspark). Use `main_py` instead of `main_jar` in properties.json and specify additional files using `py_files`.
 - PNDA-2784: Make tests pass on RedHat
+
 ### Changed
 - PNDA-2700: Spark streaming jobs no longer require upstart.conf or yarn-kill.py files, default ones are supplied by the deployment manager.
 - PNDA-2782: Disabled Ubuntu-only test

README.md

@@ -399,9 +399,9 @@ hdfspath_path_name      generated from entries in hdfs.json
 ## Environment Variables ##
 These can be obtained with the [environment endpoints API](#environment-endpoints-api)
 ````
-environment_cloudera_manager_host       192.168.1.2
-environment_cloudera_manager_password   admin
-environment_cloudera_manager_username   admin
+environment_hadoop_manager_host       192.168.1.2
+environment_hadoop_manager_password   admin
+environment_hadoop_manager_username   admin
 environment_cluster_private_key         ./dm.pem
 environment_cluster_root_user           cloud-user
 environment_hbase_rest_port             20550

api/src/main/resources/deployer_utils.py

@@ -41,26 +41,126 @@ def connect_cm(cm_api, cm_username, cm_password):
     return api
 
 
-def get_named_service(cm_host, cluster_name, service_name, user_name='admin', password='admin'):
+def get_nameservice(cm_host, cluster_name, service_name, user_name='admin', password='admin'):
     request_url = 'http://%s:7180/api/v11/clusters/%s/services/%s/nameservices' % (cm_host,
                                                                                    cluster_name,
                                                                                    service_name)
     result = requests.get(request_url, auth=(user_name, password))
-    named_service = ""
+    nameservice = ""
     if result.status_code == 200:
         response = result.json()
         if 'items' in response:
-            named_service = response['items'][0]['name']
-            logging.debug("Found named service %s for %s", named_service, service_name)
-    return named_service
+            nameservice = response['items'][0]['name']
+            logging.debug("Found named service %s for %s", nameservice, service_name)
+    return nameservice
 
 
 def fill_hadoop_env(env):
+    if env['hadoop_distro'] == 'CDH':
+        fill_hadoop_env_cdh(env)
+    else:
+        fill_hadoop_env_hdp(env)
+
+    logging.debug(env)
+
+def ambari_request(ambari, uri):
+    hadoop_manager_ip = ambari[0]
+    hadoop_manager_username = ambari[1]
+    hadoop_manager_password = ambari[2]
+    if uri.startswith("http"):
+        full_uri = uri
+    else:
+        full_uri = 'http://%s:8080/api/v1%s' % (hadoop_manager_ip, uri)
+
+    headers = {'X-Requested-By': hadoop_manager_username}
+    auth = (hadoop_manager_username, hadoop_manager_password)
+    return requests.get(full_uri, auth=auth, headers=headers).json()
+
+def get_hdfs_hdp(ambari, cluster_name):
+    core_site = ambari_request(ambari, '/clusters/%s?fields=Clusters/desired_configs/core-site' % cluster_name)
+    config_version = core_site['Clusters']['desired_configs']['core-site']['tag']
+    core_site_config = ambari_request(ambari, '/clusters/%s/configurations/?type=core-site&tag=%s' % (cluster_name, config_version))
+    return core_site_config['items'][0]['properties']['fs.defaultFS']
+
+def component_host(component_detail):
+    host_list = ''
+    for host_detail in component_detail['host_components']:
+        if len(host_list) > 0:
+            host_list += ','
+        host_list += host_detail['HostRoles']['host_name']
+    return host_list
+
+def fill_hadoop_env_hdp(env):
+
+    hadoop_manager_ip = env['hadoop_manager_host']
+    hadoop_manager_username = env['hadoop_manager_username']
+    hadoop_manager_password = env['hadoop_manager_password']
+    ambari = (hadoop_manager_ip, hadoop_manager_username, hadoop_manager_password)
+    cluster_name = ambari_request(ambari, '/clusters')['items'][0]['Clusters']['cluster_name']
+
+    logging.debug('getting service list for %s', cluster_name)
+    env['cm_status_links'] = {}
+    nameservice = None
+
+    env['name_node'] = get_hdfs_hdp(ambari, cluster_name)
+
+    services = ambari_request(ambari, '/clusters/%s/services' % cluster_name)['items']
+    for service in services:
+        service_name = service['ServiceInfo']['service_name']
+        env['cm_status_links']['%s' % service_name] = 'http://%s:8080/#/main/services/%s/summary' % (hadoop_manager_ip, service_name)
+        service_components = ambari_request(ambari, service['href'] + '/components')['items']
+
+        for component in service_components:
+            component_detail = ambari_request(ambari, component['href'])
+            role_name = component_detail['ServiceComponentInfo']['component_name']
+
+            if role_name == "NAMENODE":
+                env['webhdfs_host'] = '%s' % component_host(component_detail).split(',')[0]
+                env['webhdfs_port'] = '14000'
+
+            elif role_name == "RESOURCEMANAGER":
+                rm_host = component_host(component_detail)
+                if len(rm_host.split(',')) > 1:
+                    main_rm_host = rm_host.split(',')[0]
+                    backup_rm_host = rm_host.split(',')[1]
+                else:
+                    main_rm_host = rm_host
+                    backup_rm_host = None
+                env['yarn_resource_manager_host'] = '%s' % main_rm_host
+                env['yarn_resource_manager_port'] = '8088'
+                env['yarn_resource_manager_mr_port'] = '8050'
+                if backup_rm_host is not None:
+                    env['yarn_resource_manager_host_backup'] = '%s' % component_host(component_detail)
+                    env['yarn_resource_manager_port_backup'] = '8088'
+                    env['yarn_resource_manager_mr_port_backup'] = '8050'
+
+            elif role_name == "NODEMANAGER":
+                env['yarn_node_managers'] = '%s' % component_host(component_detail)
+
+            elif role_name == "ZOOKEEPER_SERVER":
+                env['zookeeper_quorum'] = '%s' % component_host(component_detail)
+                env['zookeeper_port'] = '2181'
+
+            elif role_name == "HBASE_MASTER":
+                env['hbase_rest_server'] = '%s' % component_host(component_detail).split(',')[0]
+                env['hbase_rest_port'] = '20550'
+                env['hbase_thrift_server'] = '%s' % component_host(component_detail).split(',')[0]
+
+            elif role_name == "OOZIE_SERVER":
+                env['oozie_uri'] = 'http://%s:11000/oozie' % component_host(component_detail)
+
+            elif role_name == "HIVE_SERVER":
+                env['hive_server'] = '%s' % component_host(component_detail)
+                env['hive_port'] = '10000'
+
+    logging.debug(env)
+
+def fill_hadoop_env_cdh(env):
     # pylint: disable=E1103
     api = connect_cm(
-        env['cloudera_manager_host'],
-        env['cloudera_manager_username'],
-        env['cloudera_manager_password'])
+        env['hadoop_manager_host'],
+        env['hadoop_manager_username'],
+        env['hadoop_manager_password'])
 
     for cluster_detail in api.get_all_clusters():
         cluster_name = cluster_detail.name
@@ -73,14 +173,14 @@ def fill_hadoop_env(env):
     for service in cluster.get_all_services():
         env['cm_status_links']['%s' % service.name] = service.serviceUrl
         if service.type == "HDFS":
-            named_service = get_named_service(env['cloudera_manager_host'], cluster_name,
+            nameservice = get_nameservice(env['hadoop_manager_host'], cluster_name,
                                               service.name,
-                                              user_name=env['cloudera_manager_username'],
-                                              password=env['cloudera_manager_password'])
-            if named_service:
-                env['name_node'] = 'hdfs://%s' % named_service
+                                              user_name=env['hadoop_manager_username'],
+                                              password=env['hadoop_manager_password'])
+            if nameservice:
+                env['name_node'] = 'hdfs://%s' % nameservice
             for role in service.get_all_roles():
-                if not named_service and role.type == "NAMENODE":
+                if not nameservice and role.type == "NAMENODE":
                     env['name_node'] = 'hdfs://%s:8020' % api.get_host(role.hostRef.hostId).hostname
                 if role.type == "HTTPFS":
                     env['webhdfs_host'] = '%s' % api.get_host(role.hostRef.hostId).ipAddress
@@ -151,9 +251,6 @@ def fill_hadoop_env(env):
                     env['hue_port'] = '8888'
                     break
 
-    logging.debug(env)
-
-
 def tree(archive_filepath):
     file_handle = file(archive_filepath, 'rb')
     tar_file = tarfile.open(None, 'r', file_handle)

api/src/main/resources/dm-config.json

@@ -36,6 +36,7 @@
 import json
 import string
 import collections
+import getpass
 import requests
 import hbase_descriptor
 import opentsdb_descriptor
@@ -166,6 +167,7 @@ def _instantiate_properties(self, application_name, component, property_override
         props['component_name'] = component['component_name']
         props['component_job_name'] = '%s-%s-job' % (props['component_application'], props['component_name'])
         props['component_hdfs_root'] = '/user/%s/%s' % (application_name, component['component_name'])
+        props['application_user'] = self._get_application_user()
         return props
 
     def _fill_properties(self, local_file, props):
@@ -327,3 +329,10 @@ def _find_yarn_app_info(self, all_yarn_applications, job_name):
                     if result is None or self._get_yarn_start_time(app) > self._get_yarn_start_time(result):
                         result = app
         return result
+
+    def _get_application_user(self):
+        application_user = getpass.getuser()
+        # if running as root, make sure to start the application under a different user.
+        if application_user == 'root':
+            application_user = self._environment['application_default_user']
+        return application_user

api/src/main/resources/plugins/base_creator.py

@@ -67,11 +67,14 @@ def create_component(self, staged_component_path, application_name, component, p
         key_file = self._environment['cluster_private_key']
         root_user = self._environment['cluster_root_user']
         target_host = self._environment['jupyter_host']
+        application_user = properties['application_user']
         delete_commands = []
 
         mkdircommands = []
-        remote_component_tmp_path = '%s/%s/%s' % ('/tmp/%s' % self._namespace, application_name, component['component_name'])
+        remote_component_tmp_path = '%s/%s/%s/%s' % ('/tmp/%s' % self._namespace, application_user, application_name, component['component_name'])
+        remote_notebook_path = '/home/%s/%s' % (application_user, self._environment['jupyter_notebook_directory'])
         mkdircommands.append('mkdir -p %s' % remote_component_tmp_path)
+        mkdircommands.append('sudo -u %s mkdir -p %s' % (application_user, remote_notebook_path))
         deployer_utils.exec_ssh(target_host, root_user, key_file, mkdircommands)
 
         file_list = component['component_detail']
@@ -82,7 +85,7 @@ def create_component(self, staged_component_path, application_name, component, p
                 os.system("scp -i %s -o StrictHostKeyChecking=no %s/%s %s@%s:%s" %
                           (key_file, staged_component_path, file_name, root_user, target_host, remote_component_tmp_path))
 
-                remote_component_install_path = '%s/%s_%s' % (self._environment['jupyter_notebook_directory'], application_name, file_name)
+                remote_component_install_path = '%s/%s_%s' % (remote_notebook_path, application_name, file_name)
                 deployer_utils.exec_ssh(
                     target_host, root_user, key_file,
                     ['sudo mv %s %s' % (remote_component_tmp_path + '/*.ipynb', remote_component_install_path)])

api/src/main/resources/plugins/jupyter.py

@@ -34,6 +34,8 @@
 
 class OozieCreator(Creator):
 
+    OOZIE_USER_NAME = 'hdfs'
+
     def validate_component(self, component):
         errors = []
         file_list = component['component_detail']
@@ -50,19 +52,19 @@ def destroy_component(self, application_name, create_data):
             application_name,
             json.dumps(create_data))
         # terminate oozie jobs
-        self._kill_oozie(create_data['job_handle'])
+        self._kill_oozie(create_data['job_handle'], create_data['application_user'])
 
         # delete component from hdfs
         remote_path = create_data['component_hdfs_root'][1:]
         self._hdfs_client.remove(remote_path, recursive=True)
 
     def start_component(self, application_name, create_data):
         logging.debug("start_component: %s %s", application_name, json.dumps(create_data))
-        self._start_oozie(create_data['job_handle'])
+        self._start_oozie(create_data['job_handle'], create_data['application_user'])
 
     def stop_component(self, application_name, create_data):
         logging.debug("stop_component: %s %s", application_name, json.dumps(create_data))
-        self._stop_oozie(create_data['job_handle'])
+        self._stop_oozie(create_data['job_handle'], create_data['application_user'])
 
     def create_component(self, staged_component_path, application_name, component, properties):
         logging.debug(
@@ -87,7 +89,7 @@ def create_component(self, staged_component_path, application_name, component, p
         properties['deployment_end'] = end.strftime("%Y-%m-%dT%H:%MZ")
 
         # insert required oozie properties
-        properties['user.name'] = 'hdfs'  # or serviceId?
+        properties['user.name'] = properties['application_user']
         # Oozie ShareLib - supports actions
         properties['oozie.use.system.libpath'] = 'true'
         # platform shared libs e.g. hbase
@@ -102,12 +104,14 @@ def create_component(self, staged_component_path, application_name, component, p
         properties[def_path] = '%s/%s' % (self._environment['name_node'], remote_path)
 
         # deploy everything to various hadoop services
-        undeploy = self._deploy_to_hadoop(properties, staged_component_path, remote_path)
+        undeploy = self._deploy_to_hadoop(properties, staged_component_path, remote_path, properties['application_user'])
 
         # return something that can be used to undeploy later
-        return {'job_handle': undeploy['id'], 'component_hdfs_root': properties['component_hdfs_root']}
+        return {'job_handle': undeploy['id'],
+                'component_hdfs_root': properties['component_hdfs_root'],
+                'application_user': properties['application_user']}
 
-    def _deploy_to_hadoop(self, properties, staged_component_path, remote_path, exclude=None):
+    def _deploy_to_hadoop(self, properties, staged_component_path, remote_path, application_user, exclude=None):
         if exclude is None:
             exclude = []
         exclude.extend(['hdfs.json',
@@ -125,7 +129,7 @@ def _deploy_to_hadoop(self, properties, staged_component_path, remote_path, excl
 
         # submit to oozie
         result = self._submit_oozie(properties)
-        self._stop_oozie(result['id'])
+        self._stop_oozie(result['id'], application_user)
 
         return result
 
@@ -149,20 +153,20 @@ def _submit_oozie(self, job_properties):
 
         return result
 
-    def _kill_oozie(self, job_id):
+    def _kill_oozie(self, job_id, oozie_user):
         logging.debug("_kill_oozie: %s", job_id)
-        oozie_url = '%s/v1/job/%s?action=kill' % (self._environment['oozie_uri'], job_id)
+        oozie_url = '%s/v1/job/%s?action=kill&user.name=%s' % (self._environment['oozie_uri'], job_id, oozie_user)
         requests.put(oozie_url)
 
-    def _start_oozie(self, job_id):
+    def _start_oozie(self, job_id, oozie_user):
         logging.debug("_start_oozie: %s", job_id)
-        oozie_url = '%s/v1/job/%s?action=resume' % (self._environment['oozie_uri'], job_id)
+        oozie_url = '%s/v1/job/%s?action=resume&user.name=%s' % (self._environment['oozie_uri'], job_id, oozie_user)
         requests.put(oozie_url)
-        oozie_url = '%s/v1/job/%s?action=start' % (self._environment['oozie_uri'], job_id)
+        oozie_url = '%s/v1/job/%s?action=start&user.name=%s' % (self._environment['oozie_uri'], job_id, oozie_user)
         requests.put(oozie_url)
 
-    def _stop_oozie(self, job_id):
+    def _stop_oozie(self, job_id, oozie_user):
         logging.debug("_stop_oozie: %s", job_id)
-        oozie_url = '%s/v1/job/%s?action=suspend' % (self._environment['oozie_uri'], job_id)
+        oozie_url = '%s/v1/job/%s?action=suspend&user.name=%s' % (self._environment['oozie_uri'], job_id, oozie_user)
         print oozie_url
         requests.put(oozie_url)