Merge pull request #17 from pndaproject/PNDA-2498

Use thrift server role for happybase and store packages in hdfs

Author: trsmith2

api/src/main/resources/app.py

@@ -297,12 +297,16 @@ def main():
 
     deployer_utils.fill_hadoop_env(config['environment'])
 
-    package_repository = PackageRepoRestClient(config['config']["package_repository"])
+    package_repository = PackageRepoRestClient(config['config']["package_repository"], config['config']['stage_root'])
     dm = deployment_manager.DeploymentManager(package_repository,
                                               package_registrar.HbasePackageRegistrar(
-                                                  config['environment']['hbase_rest_server']),
+                                                  config['environment']['hbase_thrift_server'],
+                                                  config['environment']['webhdfs_host'],
+                                                  'hdfs',
+                                                  config['environment']['webhdfs_port'],
+                                                  config['config']['stage_root']),
                                               application_registrar.HbaseApplicationRegistrar(
-                                                  config['environment']['hbase_rest_server']),
+                                                  config['environment']['hbase_thrift_server']),
                                               config['environment'],
                                               config['config'])
 

api/src/main/resources/application_creator.py

@@ -22,7 +22,6 @@
 
 import tarfile
 import os
-import io
 import json
 import re
 
@@ -46,14 +45,14 @@ def __init__(self, config, environment, service):
                                  environment['webhdfs_port'],
                                  'hdfs')
 
-    def create_application(self, package_data, package_metadata, application_name, property_overrides):
+    def create_application(self, package_data_path, package_metadata, application_name, property_overrides):
 
         logging.debug("create_application: %s", application_name)
 
         if not re.match('^[a-zA-Z0-9_-]+$', application_name):
             raise FailedCreation('Application name %s may only contain a-z A-Z 0-9 - _' % application_name)
 
-        stage_path = self._stage_package(package_data)
+        stage_path = self._stage_package(package_data_path)
 
         # create each class of components in the package, aggregating any
         # component specific return data for destruction
@@ -164,16 +163,14 @@ def _load_creator(self, component_type):
 
         return creator
 
-    def _stage_package(self, package_data):
+    def _stage_package(self, package_data_path):
 
         logging.debug("_stage_package")
 
         if not os.path.isdir(self._config['stage_root']):
             os.mkdir(self._config['stage_root'])
 
-        file_like_object = io.BytesIO(package_data)
-        tar = tarfile.open(fileobj=file_like_object)
+        tar = tarfile.open(package_data_path)
         stage_path = "%s/%s" % (self._config['stage_root'], uuid.uuid4())
         tar.extractall(path=stage_path)
-        file_like_object.close()
         return stage_path

api/src/main/resources/deployer_utils.py

@@ -97,16 +97,14 @@ def fill_hadoop_env(env):
                     env['yarn_resource_manager_mr_port%s' % rm_instance] = '8032'
                 if role.type == "NODEMANAGER":
                     if 'yarn_node_managers' in env:
-                        env['yarn_node_managers'] = '%s,%s' % (
-                            env['yarn_node_managers'], api.get_host(role.hostRef.hostId).hostname)
+                        env['yarn_node_managers'] = '%s,%s' % (env['yarn_node_managers'], api.get_host(role.hostRef.hostId).hostname)
                     else:
                         env['yarn_node_managers'] = '%s' % api.get_host(
                             role.hostRef.hostId).hostname
         elif service.type == "MAPREDUCE":
             for role in service.get_all_roles():
                 if role.type == "JOBTRACKER":
-                    env['job_tracker'] = '%s:8021' % api.get_host(
-                        role.hostRef.hostId).hostname
+                    env['job_tracker'] = '%s:8021' % api.get_host(role.hostRef.hostId).hostname
                     break
         elif service.type == "ZOOKEEPER":
             for role in service.get_all_roles():
@@ -119,42 +117,37 @@ def fill_hadoop_env(env):
         elif service.type == "HBASE":
             for role in service.get_all_roles():
                 if role.type == "HBASERESTSERVER":
-                    env['hbase_rest_server'] = '%s' % api.get_host(
-                        role.hostRef.hostId).hostname
+                    env['hbase_rest_server'] = '%s' % api.get_host(role.hostRef.hostId).hostname
                     env['hbase_rest_port'] = '20550'
-                    break
+                elif role.type == "HBASETHRIFTSERVER":
+                    env['hbase_thrift_server'] = '%s' % api.get_host(role.hostRef.hostId).hostname
         elif service.type == "OOZIE":
             for role in service.get_all_roles():
                 if role.type == "OOZIE_SERVER":
-                    env['oozie_uri'] = 'http://%s:11000/oozie' % api.get_host(
-                        role.hostRef.hostId).hostname
+                    env['oozie_uri'] = 'http://%s:11000/oozie' % api.get_host(role.hostRef.hostId).hostname
                     break
         elif service.type == "HIVE":
             for role in service.get_all_roles():
                 if role.type == "HIVESERVER2":
-                    env['hive_server'] = '%s' % api.get_host(
-                        role.hostRef.hostId).hostname
+                    env['hive_server'] = '%s' % api.get_host(role.hostRef.hostId).hostname
                     env['hive_port'] = '10000'
                     break
         elif service.type == "IMPALA":
             for role in service.get_all_roles():
                 if role.type == "IMPALAD":
-                    env['impala_host'] = '%s' % api.get_host(
-                        role.hostRef.hostId).hostname
+                    env['impala_host'] = '%s' % api.get_host(role.hostRef.hostId).hostname
                     env['impala_port'] = '21050'
                     break
         elif service.type == "KUDU":
             for role in service.get_all_roles():
                 if role.type == "KUDU_MASTER":
-                    env['kudu_host'] = '%s' % api.get_host(
-                        role.hostRef.hostId).hostname
+                    env['kudu_host'] = '%s' % api.get_host(role.hostRef.hostId).hostname
                     env['kudu_port'] = '7051'
                     break
         elif service.type == "HUE":
             for role in service.get_all_roles():
                 if role.type == "HUE_SERVER":
-                    env['hue_host'] = '%s' % api.get_host(
-                        role.hostRef.hostId).hostname
+                    env['hue_host'] = '%s' % api.get_host(role.hostRef.hostId).hostname
                     env['hue_port'] = '8888'
                     break
 
@@ -240,6 +233,25 @@ def create_file(self, data, remote_file_path):
             sio,
             overwrite=True)
 
+    def append_file(self, data, remote_file_path):
+
+        logging.debug('append to: %s', remote_file_path)
+
+        self._hdfs.append_file(canonicalize(remote_file_path), data)
+
+
+    def stream_file_to_disk(self, remote_file_path, local_file_path):
+        chunk_size = 10*1024*1024
+        offset = 0
+        with open(local_file_path, 'wb') as dest_file:
+            data = self._hdfs.read_file(canonicalize(remote_file_path), offset=offset, length=chunk_size)
+            while True:
+                dest_file.write(data)
+                if len(data) < chunk_size:
+                    break
+                offset += chunk_size
+                data = self._hdfs.read_file(canonicalize(remote_file_path), offset=offset, length=chunk_size)
+
     def read_file(self, remote_file_path):
 
         data = self._hdfs.read_file(canonicalize(remote_file_path))

api/src/main/resources/deployment_manager.py

@@ -22,6 +22,7 @@
 
 import logging
 import json
+import os
 import time
 import datetime
 import threading
@@ -155,11 +156,11 @@ def _do_deploy():
                 package_file = package + '.tar.gz'
                 logging.info("deploy: %s", package)
                 # download package:
-                package_data = self._repository.get_package(package_file)
+                package_data_path = self._repository.get_package(package_file)
                 # put package in database:
-                metadata = self._package_parser.get_package_metadata(package_data)
+                metadata = self._package_parser.get_package_metadata(package_data_path)
                 self._application_creator.validate_package(package, metadata)
-                self._package_registrar.set_package(package, package_data)
+                self._package_registrar.set_package(package, package_data_path)
                 # set the operation status as complete
                 deploy_status = {"state": PackageDeploymentState.DEPLOYED,
                                  "information": "Deployed " + package + " at " + self.utc_string()}
@@ -175,6 +176,7 @@ def _do_deploy():
             finally:
                 # report final state of operation to database:
                 self._package_registrar.set_package_deploy_status(package, deploy_status)
+                os.remove(package_data_path)
 
         # schedule work to be done in the background:
         self._run_asynch_package_task(package_name=package,
@@ -373,7 +375,7 @@ def create_application(self, package, application, overrides):
             self._assert_application_status(application, ApplicationState.NOTCREATED)
             self._assert_package_status(package, PackageDeploymentState.DEPLOYED)
             defaults = self.get_package_info(package)['defaults']
-            package_data = self._package_registrar.get_package_data(package)
+            package_data_path = self._package_registrar.get_package_data(package)
             self._application_registrar.create_application(package, application, overrides, defaults)
             self._mark_creating(application)
 
@@ -383,7 +385,7 @@ def do_work():
                 try:
                     package_metadata = self._package_registrar.get_package_metadata(package)['metadata']
                     create_data = self._application_creator.create_application(
-                        package_data, package_metadata, application, overrides)
+                        package_data_path, package_metadata, application, overrides)
                     self._application_registrar.set_create_data(application, create_data)
                     self._application_registrar.set_application_status(application, ApplicationState.CREATED)
                 except ExceptionThatShouldBeDisplayedToCaller as ex:
@@ -398,6 +400,7 @@ def do_work():
                 # clear inner locks:
                 self._clear_package_progress(application)
                 self._state_change_event_application(application)
+                os.remove(package_data_path)
 
         self.dispatcher.run_as_asynch(task=do_work)
 

api/src/main/resources/package_parser.py

@@ -22,7 +22,6 @@
 
 import json
 import tarfile
-import io
 import traceback
 import logging
 
@@ -42,14 +41,13 @@ def properties_from_metadata(self, metadata):
                 properties[component_type][component_name] = component_detail['component_detail']['properties.json']
         return properties
 
-    def get_package_metadata(self, package_data):
+    def get_package_metadata(self, package_data_path):
 
         try:
             logging.debug("get_package_metadata")
             metadata = {}
 
-            file_like_object = io.BytesIO(package_data)
-            tar = tarfile.open(fileobj=file_like_object)
+            tar = tarfile.open(package_data_path)
             for name in sorted([member.name for member in tar.getmembers()]):
                 name_parts = name.split('/')
                 package_name = name_parts[0]

api/src/main/resources/package_registrar.py

@@ -22,19 +22,27 @@
 
 import logging
 import json
+
 import happybase
 from Hbase_thrift import AlreadyExists
 
 from package_parser import PackageParser
+from deployer_utils import HDFS
 
 
 class HbasePackageRegistrar(object):
     COLUMN_DEPLOY_STATUS = "cf:deploy_status"
 
-    def __init__(self, hbase_host):
+    def __init__(self, hbase_host, hdfs_host, hdfs_user, hdfs_port, package_local_dir_path):
         self._hbase_host = hbase_host
+        self._hdfs_user = hdfs_user
+        self._hdfs_host = hdfs_host
+        self._hdfs_port = hdfs_port
+        self._hdfs_client = HDFS(hdfs_host, hdfs_port, hdfs_user)
         self._parser = PackageParser()
         self._table_name = 'platform_packages'
+        self._package_hdfs_dir_path = "/user/pnda/application_packages"
+        self._package_local_dir_path = package_local_dir_path
         if self._hbase_host is not None:
             connection = happybase.Connection(self._hbase_host)
             try:
@@ -45,10 +53,11 @@ def __init__(self, hbase_host):
             finally:
                 connection.close()
 
-    def set_package(self, package_name, package_data):
-        logging.debug("Storing %s, %s bytes", package_name, len(package_data))
-        metadata = self._parser.get_package_metadata(package_data)
-        key, data = self.generate_record(metadata, package_data)
+    def set_package(self, package_name, package_data_path):
+        logging.debug("Storing %s", package_name)
+        metadata = self._parser.get_package_metadata(package_data_path)
+        key, data = self.generate_record(metadata)
+        self._write_to_hdfs(package_data_path, data['cf:package_data'])
         self._write_to_db(key, data)
 
     def set_package_deploy_status(self, package_name, deploy_status):
@@ -62,6 +71,8 @@ def set_package_deploy_status(self, package_name, deploy_status):
 
     def delete_package(self, package_name):
         logging.debug("Deleting %s", package_name)
+        package_data_hdfs_path = self._read_from_db(package_name, ['cf:package_data'])['cf:package_data']
+        self._hdfs_client.remove(package_data_hdfs_path)
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
@@ -71,10 +82,12 @@ def delete_package(self, package_name):
 
     def get_package_data(self, package_name):
         logging.debug("Reading %s", package_name)
-        package_data = self._read_from_db(package_name, ['cf:package_data'])
-        if len(package_data) == 0:
+        record = self._read_from_db(package_name, ['cf:package_data'])
+        if len(record) == 0:
             return None
-        return package_data['cf:package_data']
+        local_package_path = "%s/%s" % (self._package_local_dir_path, package_name)
+        self._read_from_hdfs(record['cf:package_data'], local_package_path)
+        return local_package_path
 
     def get_package_metadata(self, package_name):
         logging.debug("Reading %s", package_name)
@@ -114,12 +127,12 @@ def list_packages(self):
             connection.close()
         return result
 
-    def generate_record(self, metadata, package_data):
+    def generate_record(self, metadata):
         return metadata["package_name"], {
             'cf:name': '-'.join(metadata["package_name"].split("-")[:-1]),
             'cf:version': metadata["package_name"].split("-")[-1],
             'cf:metadata': json.dumps(metadata),
-            'cf:package_data': package_data
+            'cf:package_data': "%s/%s" %  (self._package_hdfs_dir_path, metadata["package_name"])
         }
 
     def _read_from_db(self, key, columns):
@@ -131,10 +144,26 @@ def _read_from_db(self, key, columns):
             connection.close()
         return data
 
+    def _read_from_hdfs(self, source_hdfs_path, dest_local_path):
+        self._hdfs_client.stream_file_to_disk(source_hdfs_path, dest_local_path)
+
     def _write_to_db(self, key, data):
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
             table.put(key, data)
         finally:
             connection.close()
+
+    def _write_to_hdfs(self, source_local_path, dest_hdfs_path):
+        with open(source_local_path, 'rb') as source_file:
+            first = True
+            chunk_size = 10*1024*1024
+            data_chunk = source_file.read(chunk_size)
+            while data_chunk:
+                if first:
+                    self._hdfs_client.create_file(data_chunk, dest_hdfs_path)
+                    first = False
+                else:
+                    self._hdfs_client.append_file(data_chunk, dest_hdfs_path)
+                data_chunk = source_file.read(chunk_size)

api/src/main/resources/package_repo_rest_client.py

@@ -25,12 +25,13 @@
 
 
 class PackageRepoRestClient(object):
-    def __init__(self, api_url):
+    def __init__(self, api_url, package_local_dir_path):
         """
         A client implementation for the package repository API
         :param api_url: A url describing the location to make REST calls to
         """
         self.api_url = api_url
+        self._package_local_dir_path = package_local_dir_path
 
     def put_package(self, package_name, package_data):
         """
@@ -48,12 +49,15 @@ def get_package(self, package_name, expected_codes=None):
         """
         gets a package from the repository
         :param package_nam:
-        :return: the http response
+        :return: local path to file
         """
         if not expected_codes:
             expected_codes = [200]
         response = self.make_rest_get_request("/packages/" + package_name, expected_codes)
-        return response.content
+        local_path = "%s/%s" % (self._package_local_dir_path, package_name)
+        with open(local_path, 'wb') as local_file:
+            local_file.write(response.content)
+        return local_path
 
     def get_package_list(self, recency=None):
         """

api/src/main/resources/requirements.txt

@@ -26,4 +26,4 @@ thrift==0.9.3
 thriftpy==0.3.9
 tornado==4.4.2
 tornado-cors==0.6.0
-wsgiref==0.1.2
+wsgiref==0.1.2