Merge pull request #91 from dharaneeshvrd/PNDA-4796

PNDA-4796: Flink-stop added to stop the flink applications

Author: willis27

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

@@ -0,0 +1,29 @@
+#!/usr/bin/env python
+
+import commands
+import json
+import requests
+
+COMMAND_OUTPUT = commands.getoutput('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()
+            tracking_url = fields[8].strip()
+            break
+if IS_RUNNING:
+    URL = '%s/jobs' % tracking_url
+    FLINK_JOB_LIST = requests.get(URL)
+    FLINK_JOB_LIST = json.loads(FLINK_JOB_LIST.text)
+    FLINK_JOBID = FLINK_JOB_LIST['jobs-running'][0]
+    STATUS, OUT = commands.getstatusoutput('flink stop %s -yid %s' % (FLINK_JOBID, yarn_app_id))
+    if STATUS != 0:
+        commands.getoutput('flink cancel %s -yid %s' % (FLINK_JOBID, yarn_app_id))
+    commands.getoutput('yarn application -kill %s' % yarn_app_id)

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

@@ -79,7 +79,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
             raise Exception('properties.json must contain "main_jar or main_py" for %s flink %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)
+        copy(os.path.join(this_dir, 'flink-stop.py'), staged_component_path)
         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:
@@ -98,7 +98,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
 
         self._fill_properties(os.path.join(staged_component_path, service_script), 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)
+        self._fill_properties(os.path.join(staged_component_path, 'flink-stop.py'), properties)
 
         mkdircommands = []
         mkdircommands.append('mkdir -p %s' % remote_component_tmp_path)
@@ -111,7 +111,7 @@ def create_component(self, staged_component_path, application_name, user_name, c
         commands = []
         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('sudo chmod a+x %s/flink-stop.py' % (remote_component_install_path))
 
         if 'component_main_jar' in properties:
             commands.append('cd %s && sudo jar uf %s application.properties' % (remote_component_install_path, properties['component_main_jar']))

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

@@ -5,8 +5,8 @@ Description=PNDA Application: ${component_application}-${component_name}
 Type=simple
 User=${application_user}
 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
+ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
+ExecStop=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
 Environment=FLINK_VERSION=${component_flink_version}
 ExecStart=${environment_flink} run -m  yarn-cluster ${component_flink_config_args} -ynm ${component_job_name} -v ${flink_python_jar} ${component_main_py} ${component_application_args}
 Restart=${component_respawn_type}

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

@@ -5,8 +5,8 @@ Description=PNDA Application: ${component_application}-${component_name}
 Type=simple
 User=${application_user}
 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
+ExecStartPre=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
+ExecStop=/opt/${environment_namespace}/${component_application}/${component_name}/flink-stop.py
 Environment=FLINK_VERSION=${component_flink_version}
 ExecStart=${environment_flink} run -m  yarn-cluster ${component_flink_config_args} -ynm ${component_job_name} --class ${component_main_class} ${component_main_jar} ${component_application_args}
 Restart=${component_respawn_type}