Merge pull request #81 from pndaproject/PNDA-4540

Add component properties to control restart of systemd jobs

Author: trsmith2

README.md

@@ -531,6 +531,8 @@ The following varibles are only injected for Spark streaming components. They ma
 ````
 component_spark_version            major version of spark to use. Only applicable to HDP clusters, when using CDH PNDA does not support side-by-side Spark frameworks and whatever version is run by the spark-submit command will be used.
 component_spark_submit_args        additional arguments to spark-submit
+component_respawn_type             whether to restart the process when it exits. Valid values are 'always' and 'no'.
+component_respawn_timeout_sec      used with component_respawn_type to set how long to wait (in seconds) before restarting the process when it exits.
 (java only) component_main_jar     the jar containing the job code
 (python only) component_main_py    the python file containing the job code
 (python only) component_py_files   additional python files to pass to spark-submit

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

@@ -25,7 +25,6 @@
 import json
 import os
 import logging
-import platform
 from shutil import copy
 import deployer_utils
 from plugins.base_common import Common
@@ -46,8 +45,6 @@ def get_component_type(self):
 
     def create_component(self, staged_component_path, application_name, user_name, component, properties):
         logging.debug("create_component: %s %s %s %s", application_name, user_name, json.dumps(component), properties)
-        distro = platform.dist()
-        usesSystemd = distro[0] in ('redhat', 'centos')
         remote_component_tmp_path = '%s/%s/%s' % (
             '/tmp/%s' % self._namespace, application_name, component['component_name'])
         remote_component_install_path = '%s/%s/%s' % (
@@ -83,15 +80,19 @@ def create_component(self, staged_component_path, application_name, user_name, c
 
         this_dir = os.path.dirname(os.path.realpath(__file__))
         copy(os.path.join(this_dir, 'yarn-kill.py'), staged_component_path)
-        if usesSystemd:
-            service_script = 'flink.systemd.service.tpl' if java_app else 'flink.systemd.service.py.tpl'
-            service_script_install_path = '/usr/lib/systemd/system/%s.service' % service_name
+        service_script = 'flink.systemd.service.tpl' if java_app else 'flink.systemd.service.py.tpl'
+        service_script_install_path = '/usr/lib/systemd/system/%s.service' % service_name
+        if 'component_respawn_type' not in properties:
             if properties['component_flink_job_type'] == 'batch':
-                properties['respawn'] = '# Restart=always'
-                properties['respawn_limit'] = '# RestartSec=2'
+                properties['component_respawn_type'] = 'no'
             else:
-                properties['respawn'] = 'Restart=always'
-                properties['respawn_limit'] = 'RestartSec=2'
+                properties['component_respawn_type'] = 'always'
+        
+        if 'component_respawn_timeout_sec' not in properties:
+            if properties['component_flink_job_type'] == 'batch':
+                properties['component_respawn_timeout_sec'] = '0'
+            else:
+                properties['component_respawn_timeout_sec'] = '2'
 
         copy(os.path.join(this_dir, service_script), staged_component_path)
 
@@ -124,8 +125,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
         logging.debug("uninstall commands: %s", undo_commands)
 
         start_commands = []
-        if usesSystemd:
-            start_commands.append('sudo systemctl daemon-reload\n')
+        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)
 

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

@@ -9,5 +9,5 @@ ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_
 ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
 Environment=FLINK_VERSION=${component_flink_version}
 ExecStart=/usr/bin/flink run -m  yarn-cluster ${component_flink_config_args} -ynm ${component_job_name} -v ${flink_python_jar} ${component_main_py} ${component_application_args}
-${respawn}
-${respawn_limit}
+Restart=${component_respawn_type}
+RestartSec=${component_respawn_timeout_sec}

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

@@ -9,5 +9,5 @@ ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_
 ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
 Environment=FLINK_VERSION=${component_flink_version}
 ExecStart=/usr/bin/flink run -m  yarn-cluster ${component_flink_config_args} -ynm ${component_job_name} --class ${component_main_class} ${component_main_jar} ${component_application_args}
-${respawn}
-${respawn_limit}
+Restart=${component_respawn_type}
+RestartSec=${component_respawn_timeout_sec}

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

@@ -89,6 +89,10 @@ def create_component(self, staged_component_path, application_name, user_name, c
             copy(os.path.join(this_dir, 'yarn-kill.py'), staged_component_path)
             service_script = 'systemd.service.tpl' if java_app else 'systemd.service.py.tpl'
             service_script_install_path = '/usr/lib/systemd/system/%s.service' % service_name
+            if 'component_respawn_type' not in properties:
+                properties['component_respawn_type'] = 'always'
+            if 'component_respawn_timeout_sec' not in properties:
+                properties['component_respawn_timeout_sec'] = '2'
             copy(os.path.join(this_dir, service_script), staged_component_path)
 
         self._fill_properties(os.path.join(staged_component_path, service_script), properties)

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

@@ -9,5 +9,5 @@ ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_
 ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
 Environment=SPARK_MAJOR_VERSION=${component_spark_version}
 ExecStart=${environment_spark_submit} --driver-java-options "-Dlog4j.configuration=file:////opt/${environment_namespace}/${component_application}/${component_name}/log4j.properties" --conf 'spark.executor.extraJavaOptions=-Dlog4j.configuration=file:////opt/${environment_namespace}/${component_application}/${component_name}/log4j.properties' --name '${component_job_name}' --master yarn-cluster --py-files application.properties,${component_py_files} ${component_spark_submit_args} ${component_main_py}
-Restart=always
-RestartSec=2
+Restart=${component_respawn_type}
+RestartSec=${component_respawn_timeout_sec}

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

@@ -9,5 +9,5 @@ ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_
 ExecStopPost=/opt/${environment_namespace}/${component_application}/${component_name}/yarn-kill.py
 Environment=SPARK_MAJOR_VERSION=${component_spark_version}
 ExecStart=${environment_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
+Restart=${component_respawn_type}
+RestartSec=${component_respawn_timeout_sec}