Merge pull request #100 from cgiraldo/containers

Support for non ipynb files in jupyter component

Author: donaldh

Dockerfile

@@ -0,0 +1,24 @@
+FROM centos:7.6.1810
+
+RUN yum install -y epel-release
+RUN yum install -y python36-devel python36-pip cyrus-sasl-devel cyrus-sasl-gssapi cyrus-sasl-plain libffi-devel openssl-devel gcc gcc-c++
+
+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"
+

Makefile

@@ -0,0 +1,16 @@
+NAME=platform-deployment-manager
+VERSION=release5.0
+REGISTRY=pnda
+
+build:	## Build the docker image
+	docker build -t "$(NAME):$(VERSION)" -f Dockerfile .
+
+upload:	## Upload image to registry
+	docker tag "$(NAME):$(VERSION)" "$(REGISTRY)/$(NAME):$(VERSION)"
+	docker push "$(REGISTRY)/$(NAME):$(VERSION)"
+
+help:	## This help
+	@awk 'BEGIN {FS = ":.*?## "} /^[a-zA-Z_-]+:.*?## / {printf "\033[36m%-20s\033[0m %s\n", $$1, $$2}' $(MAKEFILE_LIST)
+
+.PHONY: help
+.DEFAULT_GOAL := help

api/src/main/resources/app.py

@@ -328,7 +328,7 @@ def main():
                                               package_registrar.HbasePackageRegistrar(
                                                   config['environment']['hbase_thrift_server'],
                                                   config['environment']['webhdfs_host'],
-                                                  'hdfs',
+                                                  config['environment']['webhdfs_user'],
                                                   config['environment']['webhdfs_port'],
                                                   config['config']['stage_root']),
                                               application_registrar.HbaseApplicationRegistrar(

api/src/main/resources/application_creator.py

@@ -44,10 +44,10 @@ def __init__(self, config, environment, service):
         self._name_regex = re.compile('')
         self._hdfs_client = HDFS(environment['webhdfs_host'],
                                  environment['webhdfs_port'],
-                                 'hdfs')
+                                 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/authorizer_local.py

@@ -57,7 +57,7 @@ def authorize(self, identity, resource, action):
                     authorize = True
                     logging.debug("authorize: %s for %s", authorize, grant_rule)
                     break
-            except KeyError, ex:
+            except KeyError as ex:
                 logging.warning("missing attribute %s", ex)
 
         if not authorize:

api/src/main/resources/deployer_utils.py

@@ -22,7 +22,7 @@
 
 import os
 import tarfile
-import StringIO
+from io import BytesIO
 import logging
 import traceback
 import time
@@ -32,18 +32,6 @@
 import spur
 from pywebhdfs.webhdfs import PyWebHdfsClient
 
-from cm_api.api_client import ApiResource
-
-
-def connect_cm(cm_api, cm_username, cm_password):
-    api = ApiResource(
-        cm_api,
-        version=6,
-        username=cm_username,
-        password=cm_password)
-    return api
-
-
 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,
@@ -66,9 +54,11 @@ def update_hadoop_env(env):
     tmp_env = dict(env)
     logging.debug('Updating environment descriptor')
     if env['hadoop_distro'] == 'CDH':
-        fill_hadoop_env_cdh(tmp_env)
-    else:
+        logging.error('CDH is not a supported hadoop distribution')
+    elif env['hadoop_distro'] == 'HDP':
         fill_hadoop_env_hdp(tmp_env)
+    else:
+        logging.warning('Skipping update_hadoop_env for hadoop distro "%s"', env['hadoop_distro'])
     logging.debug('Updated environment descriptor')
     for key in tmp_env:
         # Dictionary get/put operations are atomic so inherently thread safe and don't need a lock
@@ -182,105 +172,6 @@ def fill_hadoop_env_hdp(env):
                 env['hive_server'] = '%s' % component_host(component_detail)
                 env['hive_port'] = '10001'
 
-def fill_hadoop_env_cdh(env):
-    # pylint: disable=E1103
-    api = connect_cm(
-        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
-        break
-
-    logging.debug('getting %s', cluster_name)
-    env['cm_status_links'] = {}
-    env.pop('yarn_node_managers', None)
-    env.pop('yarn_resource_manager_host', None)
-    env.pop('zookeeper_quorum', None)
-
-    cluster = api.get_cluster(cluster_name)
-    for service in cluster.get_all_services():
-        env['cm_status_links']['%s' % service.name] = service.serviceUrl
-        if service.type == "HDFS":
-            nameservice = get_nameservice(env['hadoop_manager_host'], cluster_name,
-                                          service.name,
-                                          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 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).hostname
-                    env['webhdfs_port'] = '14000'
-        elif service.type == "YARN":
-            for role in service.get_all_roles():
-                if role.type == "RESOURCEMANAGER":
-                    if 'yarn_resource_manager_host' in env:
-                        rm_instance = '_backup'
-                    else:
-                        rm_instance = ''
-                    env['yarn_resource_manager_host%s' % rm_instance] = '%s' % api.get_host(role.hostRef.hostId).hostname
-                    env['yarn_resource_manager_port%s' % rm_instance] = '8088'
-                    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)
-                    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
-                    break
-        elif service.type == "ZOOKEEPER":
-            for role in service.get_all_roles():
-                if role.type == "SERVER":
-                    if 'zookeeper_quorum' in env:
-                        env['zookeeper_quorum'] += ',%s' % api.get_host(role.hostRef.hostId).hostname
-                    else:
-                        env['zookeeper_quorum'] = '%s' % api.get_host(role.hostRef.hostId).hostname
-                        env['zookeeper_port'] = '2181'
-        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_port'] = '20550'
-                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
-                    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_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_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_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_port'] = '8888'
-                    break
-
 def tree(archive_filepath):
     file_handle = file(archive_filepath, 'rb')
     tar_file = tarfile.open(None, 'r', file_handle)
@@ -297,7 +188,6 @@ def tree(archive_filepath):
 
     return root
 
-
 def canonicalize(path):
     path = path.replace('\\', '/')
     path = path.replace('//', '/')
@@ -353,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),
@@ -420,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)