Added support to deploy jupyter components.

Author: cgiraldo

Dockerfile

@@ -8,3 +8,17 @@ COPY api/src/main/resources /deployment-manager/
 WORKDIR /deployment-manager
 
 RUN pip3 install -r requirements.txt
+
+# GIT http server to serve notebooks to jupyterhub
+RUN yum install -y git nginx fcgiwrap && \
+    git config --global user.email "[email protected]" && \
+    git config --global user.name "pnda" && \
+    mkdir -p /data/git-repos/ && \
+    mkdir -p /data/stage/ 
+COPY docker/nginx.conf /etc/nginx/nginx.conf
+COPY docker/entrypoint.sh /entrypoint.sh
+# PNDA platform users must transition from Linux SO users to cloud-native. For now we add a pnda user to container images.
+RUN useradd pnda
+
+ENTRYPOINT "/entrypoint.sh"
+

api/src/main/resources/application_creator.py

@@ -47,7 +47,7 @@ def __init__(self, config, environment, service):
                                  environment['webhdfs_user'])
 
     def assert_application_properties(self, override_properties, default_properties):
-        for component_type, component_properties in default_properties.iteritems():
+        for component_type, component_properties in default_properties.items():
             creator = self._load_creator(component_type)
             creator.assert_application_properties(override_properties.get(component_type, {}), component_properties)
 
@@ -70,7 +70,7 @@ def create_application(self, package_data_path, package_metadata, application_na
         # component specific return data for destruction
         create_metadata = {}
         try:
-            for component_type, components in package_metadata['component_types'].iteritems():
+            for component_type, components in package_metadata['component_types'].items():
                 creator = self._load_creator(component_type)
                 result = creator.create_components(stage_path,
                                                    application_name,
@@ -89,7 +89,7 @@ def destroy_application(self, application_name, application_create_data):
         logging.debug("destroy_application: %s %s", application_name, application_create_data)
 
         app_hdfs_root = None
-        for component_type, component_create_data in application_create_data.iteritems():
+        for component_type, component_create_data in application_create_data.items():
             creator = self._load_creator(component_type)
             creator.destroy_components(application_name, component_create_data)
             if component_create_data and 'application_hdfs_root' in component_create_data[0]:
@@ -106,15 +106,15 @@ def start_application(self, application_name, application_create_data):
 
         logging.debug("start_application: %s %s", application_name, application_create_data)
 
-        for component_type, component_create_data in application_create_data.iteritems():
+        for component_type, component_create_data in application_create_data.items():
             creator = self._load_creator(component_type)
             creator.start_components(application_name, component_create_data)
 
     def stop_application(self, application_name, application_create_data):
 
         logging.debug("stop_application: %s %s", application_name, application_create_data)
 
-        for component_type, component_create_data in application_create_data.iteritems():
+        for component_type, component_create_data in application_create_data.items():
             creator = self._load_creator(component_type)
             creator.stop_components(application_name, component_create_data)
 
@@ -124,7 +124,7 @@ def validate_package(self, package_name, package_metadata):
 
         result = {}
         self._validate_name(package_name, package_metadata)
-        for component_type, component_metadata in package_metadata['component_types'].iteritems():
+        for component_type, component_metadata in package_metadata['component_types'].items():
             creator = self._load_creator(component_type)
             validation_errors = creator.validate_components(component_metadata)
             if validation_errors:
@@ -152,7 +152,7 @@ def get_application_runtime_details(self, application_name, application_create_d
 
         details = {}
         details['yarn_applications'] = {}
-        for component_type, component_create_data in application_create_data.iteritems():
+        for component_type, component_create_data in application_create_data.items():
             creator = self._load_creator(component_type)
             type_details = creator.get_component_runtime_details(component_create_data)
             details['yarn_applications'].update(type_details['yarn_applications'])

api/src/main/resources/application_detailed_summary.py

@@ -71,7 +71,7 @@ def _do_generate():
             try:
                 create_data = self._application_registrar.get_create_data(application)
                 input_data = {}
-                for component_name, component_data in create_data.iteritems():
+                for component_name, component_data in create_data.items():
                     input_data[component_name] = {}
                     input_data[component_name]["component_ref"] = self._load_creator(component_name)
                     input_data[component_name]["component_data"] = component_data

api/src/main/resources/application_registrar.py

@@ -26,6 +26,7 @@
 from Hbase_thrift import AlreadyExists
 
 from lifecycle_states import ApplicationState
+from hbase_utils import encode,decode
 
 
 class HbaseApplicationRegistrar(object):
@@ -101,7 +102,7 @@ def list_applications(self):
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
-            result = [key for key, data in table.scan(columns=['cf:status']) if data['cf:status'] != ApplicationState.NOTCREATED]
+            result = [decode(key) for key, data in table.scan(columns=[b'cf:status']) if decode(data[b'cf:status']) != ApplicationState.NOTCREATED]
         finally:
             connection.close()
         return result
@@ -112,8 +113,8 @@ def list_applications_for_package(self, package_name):
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
-            result = [key for key, data in table.scan(columns=['cf:package_name', 'cf:status'])
-                      if data['cf:package_name'] == package_name and data['cf:status'] != ApplicationState.NOTCREATED]
+            result = [key for key, data in table.scan(columns=[b'cf:package_name', b'cf:status'])
+                      if decode(data[b'cf:package_name']) == package_name and decode(data[b'cf:status']) != ApplicationState.NOTCREATED]
         finally:
             connection.close()
         return result
@@ -131,15 +132,15 @@ def _read_from_db(self, key):
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
-            data = table.row(key)
+            data = table.row(encode(key))
         finally:
             connection.close()
-        return data
+        return decode(data)
 
     def _write_to_db(self, key, data):
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
-            table.put(key, data)
+            table.put(encode(key), encode(data))
         finally:
             connection.close()

api/src/main/resources/application_summary_registrar.py

@@ -85,19 +85,19 @@ def get_dm_status(self, key):
         try:
             connection = happybase.Connection(self._hbase_host)
             table = connection.table("platform_applications")
-            row = table.row(key, columns=['cf:status'])
-            status = row['cf:status']
+            row = table.row(key, columns=[b'cf:status'])
+            status = row[b'cf:status']
         except TTransportException as error_message:
             logging.error(str(error_message))
         finally:
             connection.close()
-        return status
+        return status.decode()
 
     def get_flink_job_id(self, key):
         jid = ''
         data = self._read_from_db(key)
         if data:
-            data = json.loads(data['cf:component_data'])
+            data = json.loads(data[b'cf:component_data'])
             for component in data:
                 if 'flink' in component:
                     tracking_url = data[component]['tracking_url']
@@ -111,9 +111,9 @@ def get_summary_data(self, application):
             summary_data = self._read_from_db(application)
             if summary_data:
                 record[application].update({
-                    'aggregate_status': summary_data['cf:aggregate_status']
+                    'aggregate_status': summary_data[b'cf:aggregate_status']
                 })
-                record[application].update(json.loads(summary_data['cf:component_data']))
+                record[application].update(json.loads(summary_data[b'cf:component_data']))
             else:
                 record[application].update({
                     'status': 'Not Available'

api/src/main/resources/deployer_utils.py

@@ -22,7 +22,7 @@
 
 import os
 import tarfile
-from io import StringIO
+from io import BytesIO
 import logging
 import traceback
 import time
@@ -243,7 +243,7 @@ def create_file(self, data, remote_file_path, permission=755):
 
         logging.debug('create_file: %s', remote_file_path)
 
-        sio = StringIO.StringIO(data)
+        sio = BytesIO(data)
 
         self._hdfs.create_file(
             canonicalize(remote_file_path),
@@ -310,7 +310,7 @@ def exec_ssh(host, user, key, ssh_commands):
 
 def dict_to_props(dict_props):
     props = []
-    for key, value in dict_props.iteritems():
+    for key, value in dict_props.items():
         props.append('%s=%s' % (key, value))
     return '\n'.join(props)
 

api/src/main/resources/hbase_utils.py

@@ -0,0 +1,13 @@
+def decode(data):
+    if isinstance(data, bytes):  return data.decode('utf-8')
+    if isinstance(data, dict):   return dict(map(decode, data.items()))
+    if isinstance(data, tuple):  return map(decode, data)
+    if isinstance(data, list):  return list(map(decode, data))
+    return data
+
+def encode(data):
+    if isinstance(data, str):  return data.encode('utf-8')
+    if isinstance(data, dict):   return dict(map(encode, data.items()))
+    if isinstance(data, tuple):  return map(encode, data)
+    if isinstance(data, list):  return list(map(encode, data))
+    return data

api/src/main/resources/package_parser.py

@@ -35,9 +35,9 @@ def __init__(self):
 
     def properties_from_metadata(self, metadata):
         properties = {}
-        for component_type, components in metadata['component_types'].iteritems():
+        for component_type, components in metadata['component_types'].items():
             properties[component_type] = {}
-            for component_name, component_detail in components.iteritems():
+            for component_name, component_detail in components.items():
                 properties[component_type][component_name] = component_detail['component_detail']['properties.json']
         return properties
 
@@ -81,8 +81,8 @@ def get_package_metadata(self, package_data_path):
             if len(metadata['component_types']) <= 0:
                 raise FailedValidation("Expected to find at least one component within the package directory")
 
-            for component_type, components in metadata['component_types'].iteritems():
-                for component_name, component_detail in components.iteritems():
+            for component_type, components in metadata['component_types'].items():
+                for component_name, component_detail in components.items():
                     if 'properties.json' not in component_detail['component_detail']:
                         component_detail['component_detail']['properties.json'] = {}
 

api/src/main/resources/package_registrar.py

@@ -31,9 +31,10 @@
 
 from exceptiondef import FailedConnection
 
+from hbase_utils import encode,decode
 
 class HbasePackageRegistrar(object):
-    COLUMN_DEPLOY_STATUS = "cf:deploy_status"
+    COLUMN_DEPLOY_STATUS = 'cf:deploy_status'
 
     def __init__(self, hbase_host, hdfs_host, hdfs_user, hdfs_port, package_local_dir_path):
         self._hbase_host = hbase_host
@@ -110,8 +111,8 @@ def get_package_metadata(self, package_name):
             package_name, ['cf:metadata', 'cf:name', 'cf:version'])
         if not package_data:
             return None
-        return {"metadata": json.loads(package_data["cf:metadata"]), "name": package_data[
-            "cf:name"], "version": package_data["cf:version"]}
+        return {"metadata": json.loads(package_data['cf:metadata']), "name": package_data[
+            'cf:name'], "version": package_data['cf:version']}
 
     def package_exists(self, package_name):
         logging.debug("Checking %s", package_name)
@@ -138,7 +139,7 @@ def list_packages(self):
         try:
             connection = happybase.Connection(self._hbase_host)
             table = connection.table(self._table_name)
-            result = [key for key, _ in table.scan(columns=['cf:name'])]
+            result = [key.decode() for key, _ in table.scan(columns=['cf:name'])]
         except Exception as exc:
             logging.debug(str(exc))
             raise FailedConnection('Unable to connect to the HBase master')
@@ -159,10 +160,11 @@ def _read_from_db(self, key, columns):
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
-            data = table.row(key, columns=columns)
+            data = table.row(encode(key), columns=encode(columns))
         finally:
             connection.close()
-        return data
+        
+        return decode(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)
@@ -171,7 +173,7 @@ def _write_to_db(self, key, data):
         connection = happybase.Connection(self._hbase_host)
         try:
             table = connection.table(self._table_name)
-            table.put(key, data)
+            table.put(encode(key), encode(data))
         finally:
             connection.close()
 

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

@@ -246,7 +246,7 @@ def _destroy_optional_descriptors(self, descriptors):
     def create_components(self, stage_path, application_name, user_name, components,
                           components_overrides):
         results = []
-        for component_name, component in components.iteritems():
+        for component_name, component in components.items():
             staged_component_path = '%s/%s' % (stage_path, component['component_path'])
             overrides = components_overrides.get(component_name) if components_overrides is not None else {}
             overrides = {} if overrides is None else overrides
@@ -280,7 +280,7 @@ def stop_components(self, application_name, stop_data):
     def validate_components(self, components):
         logging.debug("validate_components: %s", components)
         result = {}
-        for component_name, component in components.iteritems():
+        for component_name, component in components.items():
             validation_errors = self.validate_component(component)
             if validation_errors:
                 result[component_name] = validation_errors