Merge pull request #27 from pndaproject/hdp

jgarnier · web-flow · commit a940e324c8a4 · 2017-08-09T15:57:47.000+02:00
Support for Hortonworks HDP
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,10 +1,15 @@
 # Change Log
 All notable changes to this project will be documented in this file.
 
+## [Unreleased]
+### Added:
+- 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
diff --git a/README.md b/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
diff --git a/api/src/main/resources/deployer_utils.py b/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)
diff --git a/api/src/main/resources/dm-config.json b/api/src/main/resources/dm-config.json
diff --git a/api/src/main/resources/plugins/oozie.py b/api/src/main/resources/plugins/oozie.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'], self.OOZIE_USER_NAME)
 
         # 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'], self.OOZIE_USER_NAME)
 
     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'], self.OOZIE_USER_NAME)
 
     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'] = self.OOZIE_USER_NAME
         # Oozie ShareLib - supports actions
         properties['oozie.use.system.libpath'] = 'true'
         # platform shared libs e.g. hbase
@@ -125,7 +127,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'], self.OOZIE_USER_NAME)
 
         return result
 
@@ -149,20 +151,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)
diff --git a/api/src/main/resources/test_application_creator.py b/api/src/main/resources/test_application_creator.py
