Merge pull request #7 from jeclarke/jupyter-plugin

Added a Jupyter plugin to deploy notebooks

Author: jeclarke

CHANGELOG.md

@@ -1,40 +1,44 @@
-# Change Log
-All notable changes to this project will be documented in this file.
-
-## [0.1.1] 2016-09-13
-### Changes
-- Improvements to documentation
-- Enhanced CI support
-
-## [0.1.0] 2016-07-01
-### First version
-
-## [Pre-release]
-
-### Added
-
-- Add hue endpoint to environment endpoints API
-- Application names checked to only contain alphanumeric characters (a-z A-Z 0-9 - and _) because they are used directly in file paths.
+# Change Log
+All notable changes to this project will be documented in this file.
+
+## [Unreleased]
+### Added
+- PNDA-2233 Jupyter notebook plugin added to deployment manager
+
+## [0.1.1] 2016-09-13
+### Changes
+- Improvements to documentation
+- Enhanced CI support
+
+## [0.1.0] 2016-07-01
+### First version
+
+## [Pre-release]
+
+### Added
+
+- Add hue endpoint to environment endpoints API
+- Application names checked to only contain alphanumeric characters (a-z A-Z 0-9 - and _) because they are used directly in file paths.
 - Added ability to discover HDFS namedservices
 - Added information field to status reports
-- Using an external pacakge repository API instead of internal swift integration
-- Application detail API (GET /applications/<application>/detail) now returns YARN IDs assigned to the running tasks for that application.
+- Using an external pacakge repository API instead of internal swift integration
+- Application detail API (GET /applications/<application>/detail) now returns YARN IDs assigned to the running tasks for that application.
 - Oozie error messages are reported when querying for status of an application creation call.
 - Packages are validated on deployment and the error messages reported when querying for status of a package deployment call.
 - Added support for opentsdb.json descriptor for creating metrics when deploying applications.
 - Callback events are sent to the console data logger.
 - Application detail API now completed to return Yarn IDs for any Yarn applications associated with a PNDA application.
-
-### Fixed 
-
-- Return IP address for webhdfs/HTTPFS endpoint instead of hostname
-- Timeout calls to package repository at 120 seconds.
-- Deploying a package that does not exist in the package repository now results in a useful error message being returned to the caller.
-- Fixed defect preventing '-' being used in application names.
-- Fix Zookeeper quorum bug issue
-- Improve package validation to catch packages without 3 point version numbers and where the folder inside the tar does not match the package name.
-- Add list of zookeeper nodes to quorum
-- Remove port=8020 for named service
-- Oozie creator plugin sets 'oozie.wf.application.path' and 'oozie.coord.application.path' to point at the folder not the xml files.
-- Removed some stdout printouts
-- Fixed bug preventing recency parameter being used on the repository/packages API.
+
+### Fixed 
+
+- Return IP address for webhdfs/HTTPFS endpoint instead of hostname
+- Timeout calls to package repository at 120 seconds.
+- Deploying a package that does not exist in the package repository now results in a useful error message being returned to the caller.
+- Fixed defect preventing '-' being used in application names.
+- Fix Zookeeper quorum bug issue
+- Improve package validation to catch packages without 3 point version numbers and where the folder inside the tar does not match the package name.
+- Add list of zookeeper nodes to quorum
+- Remove port=8020 for named service
+- Oozie creator plugin sets 'oozie.wf.application.path' and 'oozie.coord.application.path' to point at the folder not the xml files.
+- Removed some stdout printouts
+- Fixed bug preventing recency parameter being used on the repository/packages API.

api/src/main/resources/application_creator.py

@@ -31,6 +31,7 @@
 import uuid
 from importlib import import_module
 from exceptiondef import FailedValidation, FailedCreation
+from deployer_utils import HDFS
 
 
 class ApplicationCreator(object):
@@ -41,6 +42,9 @@ def __init__(self, config, environment, service):
         self._service = service
         self._component_creators = {}
         self._name_regex = re.compile('')
+        self._hdfs_client = HDFS(environment['webhdfs_host'],
+                                 environment['webhdfs_port'],
+                                 'hdfs')
 
     def create_application(self, package_data, package_metadata, application_name, property_overrides):
 
@@ -76,6 +80,11 @@ def destroy_application(self, application_name, application_create_data):
             creator = self._load_creator(component_type)
             creator.destroy_components(application_name, component_create_data)
 
+        local_path = '/opt/%s/%s/' % (self._service, application_name)
+        if os.path.isdir(local_path):
+            os.rmdir(local_path)
+        self._hdfs_client.remove('/user/%s' % application_name, recursive=False)
+
     def start_application(self, application_name, application_create_data):
 
         logging.debug("start_application: %s %s", application_name, application_create_data)

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

@@ -37,7 +37,6 @@
 import string
 import collections
 import subprocess
-import os
 import hbase_descriptor
 import opentsdb_descriptor
 from deployer_utils import HDFS
@@ -235,10 +234,6 @@ def destroy_components(self, application_name, create_data):
         for single_component_data in create_data:
             self._destroy_optional_descriptors(single_component_data['descriptors'])
             self.destroy_component(application_name, single_component_data)
-        local_path = '/opt/%s/%s/' % (self._namespace, application_name)
-        if os.path.isdir(local_path):
-            os.rmdir(local_path)
-        self._hdfs_client.remove('/user/%s' % application_name, recursive=False)
         return None
 
     def start_components(self, application_name, start_data):

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

@@ -0,0 +1,92 @@
+"""
+Name:       jupyter.py
+Purpose:    Installs a jupyter notebook
+Author:     PNDA team
+
+Created:    03/10/2016
+
+Copyright (c) 2016 Cisco and/or its affiliates.
+
+This software is licensed to you under the terms of the Apache License, Version 2.0 (the "License").
+You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+The code, technical concepts, and all information contained herein, are the property of Cisco Technology, Inc.
+and/or its affiliated entities, under various laws including copyright, international treaties, patent,
+and/or contract. Any use of the material herein must be in accordance with the terms of the License.
+All rights not expressly granted by the License are reserved.
+
+Unless required by applicable law or agreed to separately in writing, software distributed under the
+License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+either express or implied.
+"""
+
+# pylint: disable=C0103
+
+import json
+import os
+import logging
+
+import deployer_utils
+from plugins.base_creator import Creator
+
+
+class JupyterCreator(Creator):
+
+    def validate_component(self, component):
+        errors = []
+        notebook_found = False
+        file_list = component['component_detail']
+        for file_name in file_list:
+            if file_name.endswith(r'.ipynb'):
+                notebook_found = True
+
+        if notebook_found is False:
+            errors.append('missing ipynb file')
+
+        return errors
+
+    def get_component_type(self):
+        return 'jupyter'
+
+    def destroy_component(self, application_name, create_data):
+        logging.debug("destroy_component: %s %s", application_name, json.dumps(create_data))
+        key_file = self._environment['cluster_private_key']
+        root_user = self._environment['cluster_root_user']
+        target_host = self._environment['jupyter_host']
+        deployer_utils.exec_ssh(target_host, root_user, key_file, create_data['delete_commands'])
+
+    def start_component(self, application_name, create_data):
+        logging.debug("start_component (nothing to do for jupyter): %s %s", application_name, json.dumps(create_data))
+
+    def stop_component(self, application_name, create_data):
+        logging.debug("stop_component (nothing to do for jupyter): %s %s", application_name, json.dumps(create_data))
+
+    def create_component(self, staged_component_path, application_name, component, properties):
+        logging.debug("create_component: %s %s %s", application_name, json.dumps(component), properties)
+
+        key_file = self._environment['cluster_private_key']
+        root_user = self._environment['cluster_root_user']
+        target_host = self._environment['jupyter_host']
+        delete_commands = []
+
+        mkdircommands = []
+        remote_component_tmp_path = '%s/%s/%s' % ('/tmp/%s' % self._namespace, application_name, component['component_name'])
+        mkdircommands.append('mkdir -p %s' % remote_component_tmp_path)
+        deployer_utils.exec_ssh(target_host, root_user, key_file, mkdircommands)
+
+        file_list = component['component_detail']
+        for file_name in file_list:
+            if file_name.endswith(r'.ipynb'):
+                self._fill_properties('%s/%s' % (staged_component_path, file_name), properties)
+                logging.debug('Copying %s/* to %s:%s', staged_component_path, target_host, remote_component_tmp_path)
+                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)
+                deployer_utils.exec_ssh(
+                    target_host, root_user, key_file,
+                    ['sudo mv %s %s' % (remote_component_tmp_path + '/*.ipynb', remote_component_install_path)])
+                delete_commands.append('sudo rm -rf %s\n' % remote_component_install_path)
+
+        logging.debug("uninstall commands: %s", delete_commands)
+        return {'delete_commands': delete_commands}

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

@@ -61,7 +61,7 @@ def start_component(self, application_name, create_data):
         self._start_oozie(create_data['job_handle'])
 
     def stop_component(self, application_name, create_data):
-        logging.debug("start_component: %s %s", application_name, json.dumps(create_data))
+        logging.debug("stop_component: %s %s", application_name, json.dumps(create_data))
         self._stop_oozie(create_data['job_handle'])
 
     def create_component(self, staged_component_path, application_name, component, properties):

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

@@ -57,7 +57,7 @@ def start_component(self, application_name, create_data):
         self._control_component(create_data['start_cmds'])
 
     def stop_component(self, application_name, create_data):
-        logging.debug("start_component: %s %s", application_name, json.dumps(create_data))
+        logging.debug("stop_component: %s %s", application_name, json.dumps(create_data))
         self._control_component(create_data['stop_cmds'])
 
     def _control_component(self, cmds):

api/src/main/resources/test_application_creator.py

@@ -338,13 +338,16 @@ def test_invalid_package(self):
         creator.validate_package('test-package-1.0.2', {"package_name": "test-package-1.0.2", "component_types": {}})
 
     # pylint: disable=unused-argument
+    @patch('deployer_utils.HDFS')
     @patch('deployer_utils.exec_ssh')
     @patch('requests.put')
     @patch('os.path.isdir')
     @patch('os.rmdir')
-    def test_destroy_application(self, rmdir_mock, isdir_mock, put_mock, exec_ssh_mock):
+    def test_destroy_application(self, rmdir_mock, isdir_mock, put_mock, exec_ssh_mock, hdfs_client_mock):
         isdir_mock.return_value = True
         creator = ApplicationCreator(self.config, self.environment, self.service)
+        # pylint: disable=protected-access
+        creator._hdfs_client = hdfs_client_mock
         creator.destroy_application('name', self.create_data)
         print exec_ssh_mock.call_args_list
         exec_ssh_mock.assert_any_call('localhost', 'root_user', 'keyfile.pem', [

api/src/main/resources/test_deployer_manager.py

@@ -584,7 +584,7 @@ def test_get_environment(self):
         repository = Mock()
         package_registrar = Mock()
         application_registrar = Mock()
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -602,7 +602,7 @@ def test_list_packages(self):
         package_registrar = Mock()
         package_registrar.list_packages.return_value = expected_packages
         application_registrar = Mock()
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -633,7 +633,7 @@ def test_list_repository(self):
 
         package_registrar = Mock()
         application_registrar = Mock()
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -650,7 +650,7 @@ def test_get_unknown_package_info(self):
         package_registrar.get_package_deploy_status.return_value = None
         package_registrar.package_exists.return_value = False
         application_registrar = Mock()
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -673,7 +673,7 @@ def test_list_package_applications(self):
         package_registrar = Mock()
         application_registrar = Mock()
         application_registrar.list_applications_for_package.return_value = expected_applications
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -691,7 +691,7 @@ def test_list_applications(self):
         package_registrar = Mock()
         application_registrar = Mock()
         application_registrar.list_applications.return_value = expected_applications
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -713,7 +713,7 @@ def test_application_in_progress(self):
             'package_name': 'package_name',
             'status': ApplicationState.STARTING,
             'information': None}
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -728,7 +728,7 @@ def test_package_in_progress(self):
         repository = Mock()
         package_registrar = Mock()
         application_registrar = Mock()
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         class DeploymentManagerTester(DeploymentManager):
@@ -750,7 +750,7 @@ def test_package_not_exists(self):
         package_registrar.get_package_deploy_status.return_value = None
         package_registrar.package_exists.return_value = False
         application_registrar = Mock()
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
         dmgr = DeploymentManager(repository,
                                  package_registrar,
@@ -775,7 +775,7 @@ def test_get_application_detail(self, app_mock):
             'package_name': 'package_name',
             'status': ApplicationState.STARTED,
             'information': None}
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,
@@ -801,7 +801,7 @@ def test_fail_get_detail(self, app_mock):
             'package_name': 'package_name',
             'status': ApplicationState.NOTCREATED,
             'information': None}
-        environment = {"namespace": "some_namespace"}
+        environment = {"namespace": "some_namespace", 'webhdfs_host': 'webhdfshost', 'webhdfs_port': 'webhdfsport'}
         config = {"deployer_thread_limit": 10}
 
         dmgr = DeploymentManager(repository,