Spark streaming applications work on redhat

No longer require upstart config to be submitted with application, but
automatically insert either upstart or systemd config as required.

PNDA-2700

Author: jeclarke

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Change Log
 All notable changes to this project will be documented in this file.
 
+## Unreleased
+### Changed
+ - PNDA-2700: Spark streaming jobs no longer require upstart.conf or yarn-kill.py files, default ones are supplied by the deployment manager.
+
 ## [0.3.0] 2017-01-20
 ### Fixed
 - PNDA-2498: Application package data is now stored in HDFS with a reference to the path only held in the HBase record. This prevents HBase being overloaded with large packages (100MB+).

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

@@ -25,6 +25,8 @@
 import json
 import os
 import logging
+import platform
+from shutil import copy
 
 import deployer_utils
 from plugins.base_creator import Creator
@@ -35,12 +37,8 @@ class SparkStreamingCreator(Creator):
     def validate_component(self, component):
         errors = []
         file_list = component['component_detail']
-        if 'yarn-kill.py' not in file_list:
-            errors.append('missing file yarn-kill.py')
         if 'application.properties' not in file_list:
             errors.append('missing file application.properties')
-        if 'upstart.conf' not in file_list:
-            errors.append('missing file upstart.conf')
         if 'log4j.properties' not in file_list:
             errors.append('missing file log4j.properties')
         return errors
@@ -68,20 +66,47 @@ def _control_component(self, cmds):
 
     def create_component(self, staged_component_path, application_name, component, properties):
         logging.debug("create_component: %s %s %s", application_name, json.dumps(component), properties)
-
+        distro = platform.dist()
+        redhat = distro[0] == 'redhat'
         remote_component_tmp_path = '%s/%s/%s' % (
             '/tmp/%s' % self._namespace, application_name, component['component_name'])
         remote_component_install_path = '%s/%s/%s' % (
             '/opt/%s' % self._namespace, application_name, component['component_name'])
+        service_name = '%s-%s-%s' % (self._namespace, application_name, component['component_name'])
 
         key_file = self._environment['cluster_private_key']
         root_user = self._environment['cluster_root_user']
         target_host = 'localhost'
 
-        self._fill_properties('%s/%s' % (staged_component_path, 'upstart.conf'), properties)
-        self._fill_properties('%s/%s' % (staged_component_path, 'log4j.properties'), properties)
-        self._fill_properties('%s/%s' % (staged_component_path, 'application.properties'), properties)
-        self._fill_properties('%s/%s' % (staged_component_path, 'yarn-kill.py'), properties)
+        if 'component_spark_submit_args' not in properties:
+            properties['component_spark_submit_args'] = ''
+
+        if 'upstart.conf' in component['component_detail']:
+            # old style applications for backward compatibility
+            service_script = 'upstart.conf'
+            service_script_install_path = '/etc/init/%s.conf' % service_name
+        else:
+            # new style applications that don't need to provide upstart.conf or yarn-kill.py
+            if 'component_main_class' not in properties:
+                raise Exception('properties.json must contain "main_class" for %s sparkStreaming %s' % (application_name, component['component_name']))
+            if 'component_main_jar' not in properties:
+                raise Exception('properties.json must contain "main_jar" for %s sparkStreaming %s' % (application_name, component['component_name']))
+
+            this_dir = os.path.dirname(os.path.realpath(__file__))
+            copy(os.path.join(this_dir, 'yarn-kill.py'), staged_component_path)
+            distro = platform.dist()
+            if redhat:
+                service_script = 'systemd.service.tpl'
+                service_script_install_path = '/usr/lib/systemd/system/%s.service' % service_name
+            else:
+                service_script = 'upstart.conf.tpl'
+                service_script_install_path = '/etc/init/%s.conf' % service_name
+            copy(os.path.join(this_dir, service_script), staged_component_path)
+
+        self._fill_properties(os.path.join(staged_component_path, service_script), properties)
+        self._fill_properties(os.path.join(staged_component_path, 'log4j.properties'), properties)
+        self._fill_properties(os.path.join(staged_component_path, 'application.properties'), properties)
+        self._fill_properties(os.path.join(staged_component_path, 'yarn-kill.py'), properties)
 
         mkdircommands = []
         mkdircommands.append('mkdir -p %s' % remote_component_tmp_path)
@@ -99,32 +124,33 @@ def create_component(self, staged_component_path, application_name, component, p
                                     ['sudo mkdir -p %s' % remote_component_install_path,
                                      'sudo mv %s %s' % (remote_component_tmp_path + '/log4j.properties', remote_component_install_path + '/log4j.properties')])
 
+        if 'component_main_jar' in properties:
+            main_jar_name = properties['component_main_jar']
+        else:
+            main_jar_name = '*.jar'
+
         commands = []
-        service_name = '%s-%s-%s' % (self._namespace, application_name, component['component_name'])
-        upstart_script = '/etc/init/%s.conf' % service_name
-        commands.append('sudo cp %s/upstart.conf %s' %
-                        (remote_component_tmp_path, upstart_script))
-        commands.append('sudo cp %s/* %s' %
-                        (remote_component_tmp_path, remote_component_install_path))
-        commands.append('sudo chmod a+x %s/yarn-kill.py' %
-                        (remote_component_install_path))
-        commands.append('cd %s && sudo jar uf *.jar application.properties' %
-                        (remote_component_install_path))
+        commands.append('sudo cp %s/%s %s' % (remote_component_tmp_path, service_script, service_script_install_path))
+        commands.append('sudo cp %s/* %s' % (remote_component_tmp_path, remote_component_install_path))
+        commands.append('sudo chmod a+x %s/yarn-kill.py' % (remote_component_install_path))
+        commands.append('cd %s && sudo jar uf %s application.properties' % (remote_component_install_path, main_jar_name))
         commands.append('sudo rm -rf %s' % (remote_component_tmp_path))
         deployer_utils.exec_ssh(target_host, root_user, key_file, commands)
 
         undo_commands = []
-        undo_commands.append('sudo initctl stop %s\n' % service_name)
+        undo_commands.append('sudo service %s stop\n' % service_name)
         undo_commands.append('sudo rm -rf %s\n' % remote_component_install_path)
-        undo_commands.append('sudo rm  %s\n' % upstart_script)
+        undo_commands.append('sudo rm  %s\n' % service_script_install_path)
         logging.debug("uninstall commands: %s", undo_commands)
 
         start_commands = []
-        start_commands.append('sudo initctl start %s\n' % service_name)
+        if redhat:
+            start_commands.append('sudo systemctl daemon-reload\n')
+        start_commands.append('sudo service %s start\n' % service_name)
         logging.debug("start commands: %s", start_commands)
 
         stop_commands = []
-        stop_commands.append('sudo initctl stop %s\n' % service_name)
+        stop_commands.append('sudo service %s stop\n' % service_name)
         logging.debug("stop commands: %s", stop_commands)
 
         return {'ssh': undo_commands, 'start_cmds': start_commands, 'stop_cmds': stop_commands}

api/src/main/resources/plugins/systemd.service.tpl

@@ -0,0 +1,12 @@
+[Unit]
+Description=PNDA Application: ${component_application}-${component_name}
+
+[Service]
+Type=simple
+User=hdfs
+WorkingDirectory=/opt/${environment_namespace}/${component_application}/${component_name}/
+ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+ExecStart=/usr/bin/spark-submit --driver-java-options "-Dlog4j.configuration=file:////opt/${environment_namespace}/${component_application}/${component_name}/log4j.properties" --class ${component_main_class} --name '${component_job_name}' --master yarn-cluster --files log4j.properties ${component_spark_submit_args} ${component_main_jar}
+Restart=always
+RestartSec=2

api/src/main/resources/plugins/upstart.conf.tpl

@@ -0,0 +1,8 @@
+start on runlevel [2345]
+stop on runlevel [016]
+respawn
+respawn limit unlimited
+pre-start exec /opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+pre-stop exec /opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
+env programDir=/opt/${environment_namespace}/${component_application}/${component_name}/
+exec sudo -u hdfs spark-submit --driver-java-options "-Dlog4j.configuration=file:///${programDir}log4j.properties" --class ${component_main_class} --name '${component_job_name}' --master yarn-cluster --files ${programDir}log4j.properties ${component_spark_submit_args} ${programDir}${component_main_jar}

api/src/main/resources/plugins/yarn-kill.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python
+import subprocess
+COMMAND_OUTPUT = subprocess.check_output(['yarn', 'application', '-list'])
+
+IS_RUNNING = False
+
+for line in COMMAND_OUTPUT.splitlines():
+    fields = line.split('\t')
+    if len(fields) >= 6:
+        app = fields[1].strip()
+        state = fields[5].strip()
+        if app == '${component_job_name}':
+            IS_RUNNING = True
+            yarn_app_id = fields[0].strip()
+            break
+
+if IS_RUNNING:
+    print 'app is running, killing it...'
+    subprocess.check_output(['yarn', 'application', '-kill', yarn_app_id])

api/src/main/resources/test_application_creator.py

@@ -54,7 +54,8 @@ class ApplicationCreatorTests(unittest.TestCase):
                     "component_detail": {
                         "properties.json": {
                             "property1": "1",
-                            "property2": "two"
+                            "main_class": "abc",
+                            "main_jar": "abc.jar",
                         }
                     },
                     "component_path": "test_package-1.0.2/sparkStreaming/componentC",
@@ -155,8 +156,9 @@ class ApplicationCreatorTests(unittest.TestCase):
     @patch('application_creator.shutil')
     @patch('application_creator.os')
     @patch('application_creator.tarfile')
+    @patch('shutil.copy')
     # pylint: disable=unused-argument
-    def test_create_application(self, tar_mock, os_mock, shutil_mock, spur_ssh,
+    def test_create_application(self, copy_mock, tar_mock, os_mock, shutil_mock, spur_ssh,
                                 hdfs_client_mock, post_mock, put_mock, exec_ssh_mock,
                                 os_sys_mock, dt_mock, hive_mock, hbase_mock):
         dt_mock.utcnow.return_value = (datetime(2013, 01, 01))
@@ -183,7 +185,7 @@ def json(self):
         exec_ssh_mock.assert_any_call('nm1', 'root_user', 'keyfile.pem', ['sudo mkdir -p /opt/ns/aname/componentC', 'sudo mv /tmp/ns/aname/componentC/log4j.properties /opt/ns/aname/componentC/log4j.properties'])
         exec_ssh_mock.assert_any_call('nm2', 'root_user', 'keyfile.pem', ['mkdir -p /tmp/ns/aname/componentC'])
         exec_ssh_mock.assert_any_call('nm2', 'root_user', 'keyfile.pem', ['sudo mkdir -p /opt/ns/aname/componentC', 'sudo mv /tmp/ns/aname/componentC/log4j.properties /opt/ns/aname/componentC/log4j.properties'])
-        exec_ssh_mock.assert_any_call('localhost', 'root_user', 'keyfile.pem', ['sudo cp /tmp/ns/aname/componentC/upstart.conf /etc/init/ns-aname-componentC.conf', 'sudo cp /tmp/ns/aname/componentC/* /opt/ns/aname/componentC', 'sudo chmod a+x /opt/ns/aname/componentC/yarn-kill.py', 'cd /opt/ns/aname/componentC && sudo jar uf *.jar application.properties', 'sudo rm -rf /tmp/ns/aname/componentC'])
+        exec_ssh_mock.assert_any_call('localhost', 'root_user', 'keyfile.pem', ['sudo cp /tmp/ns/aname/componentC/upstart.conf.tpl /etc/init/ns-aname-componentC.conf', 'sudo cp /tmp/ns/aname/componentC/* /opt/ns/aname/componentC', 'sudo chmod a+x /opt/ns/aname/componentC/yarn-kill.py', 'cd /opt/ns/aname/componentC && sudo jar uf abc.jar application.properties', 'sudo rm -rf /tmp/ns/aname/componentC'])
 
     @patch('starbase.Connection')
     @patch('pyhs2.connect')
@@ -260,8 +262,9 @@ def json(self):
     @patch('application_creator.shutil')
     @patch('application_creator.os')
     @patch('application_creator.tarfile')
+    @patch('shutil.copy')
     # pylint: disable=unused-argument
-    def test_app_name_ok(self, tar_mock, os_mock, shutil_mock, spur_ssh,
+    def test_app_name_ok(self, copy_mock, tar_mock, os_mock, shutil_mock, spur_ssh,
                          hdfs_client_mock, post_mock, put_mock, exec_ssh_mock,
                          os_sys_mock, dt_mock, hive_mock, hbase_mock):
 
@@ -310,9 +313,7 @@ def test_validate_package(self):
                 'componentA': ['missing file workflow.xml'],
                 'componentB': ['missing file workflow.xml']},
             'sparkStreaming': {
-                'componentC': ['missing file yarn-kill.py',
-                               'missing file application.properties',
-                               'missing file upstart.conf',
+                'componentC': ['missing file application.properties',
                                'missing file log4j.properties']}}
         self.assertEqual(result, expected_report)