Update alarm_flink_job.py

dags/flink/alarm_flink_job.py

@@ -77,6 +77,29 @@ def failure_callback(context):
     cmd = " bash " + DING_PATH + "  \'%s\' \'%s\' \'%s\' " % (title, message, DINGDING)
     os.system(cmd)
 
+def get_name(cmd):
+    logger.info('exec cmd:' + cmd)
+    config_dict = {}
+    timeout = 10
+
+    try:
+        (stdout, stderr) = run_command(cmd, timeout)
+        stdout = stdout.decode()
+        stderr = stderr.decode()
+        logger.info(stderr)
+        logger.info(stdout)
+
+        config_dict = json.loads(stdout)
+        if 'errors' in config_dict:
+            msg = '\n- **errors**: ' + config_dict['errors'][0]
+            logger.error(msg)
+            raise Exception(msg)
+    except TimeoutError:
+        msg = '\n- **error_msg**: excute command=(%s) timeout after %i s' % (cmd, timeout)
+        logger.error(msg)
+        raise Exception(msg)
+    return config_dict['name']
+
 def get_jobs(cmd):
     logger.info('exec cmd:' + cmd)
     jobs_dict = {}
@@ -111,11 +134,15 @@ def judge_job_status(job):
         msg = "\n- **errors**: %s" % (job['errors'])
         logger.error(msg)
         return (False, msg)
+
+    job_id = job['id']
+    job_status = job['status']
+    job_name = get_name(cmd + job_id + "/config")
     if 'RUNNING' != job['status']:
-        msg = "\n- **job_id**: %s, **status**: %s" % (job['id'],job['status'])
+        msg = "\n- **job_id**: %s, **job_name**: %s, **status**: %s" % (job_id,job_name,job_status)
         logger.error(msg)
         return (False, msg)
-    msg = "\n- **job_id**: %s, **status**: %s" % (job['id'],job['status'])
+    msg = "\n- **job_id**: %s, **job_name**: %s, **status**: %s" % (job_id,job_name,job_status)
     logger.info(msg)
     return (True, msg)
 
@@ -135,7 +162,7 @@ def python_callable(**kwargs):
     running_job = 0
 
     for job in jobs_list:
-        (isrunning,msg) = judge_job_status(job)
+        (isrunning,msg) = judge_job_status(job, curl + flink_jobmanager_url)
         if not isrunning:
             error_msg += msg
         else :