Update alarm_flink_job.py

dags/flink/alarm_flink_job.py

@@ -24,6 +24,9 @@ SERVICES = [
 DINGDING = 'https://oapi.dingtalk.com/robot/send?access_token=4e00d7f7b3b8686ea7d37bd01264f86e197294f9f995ef8e12cc853760a30c60'
 DING_PATH = '/opt/bitnami/airflow/dags/*/script/ding.sh'
 
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
 class TimeoutError(Exception):
     pass
 
@@ -70,12 +73,12 @@ def failure_callback(context):
     \n> **For more details, please check the airflow task log.**
     ''' % (context['task_instance'].task_id, context['task_instance'].dag_id, context['task_instance'].task_id, context['exception'])
     
-    logging.error('message : \n' + message)
+    logger.error('message : \n' + message)
     cmd = " bash " + DING_PATH + "  \'%s\' \'%s\' \'%s\' " % (title, message, DINGDING)
     os.system(cmd)
 
 def get_jobs(cmd):
-    logging.info('exec cmd:' + cmd)
+    logger.info('exec cmd:' + cmd)
     jobs_dict = {}
     timeout = 10
 
@@ -83,8 +86,8 @@ def get_jobs(cmd):
         (stdout, stderr) = run_command(cmd, timeout)
         stdout = stdout.decode()
         stderr = stderr.decode()
-        logging.info(stderr)
-        logging.info(stdout)
+        logger.info(stderr)
+        logger.info(stdout)
 
         if 'Could not resolve host' in stderr and not stdout:
             msg = '\n- **error_msg**: Could not resolve host %s' % (cmd)
@@ -95,36 +98,36 @@ def get_jobs(cmd):
         jobs_dict = json.loads(stdout)
         if 'errors' in jobs_dict:
             msg = '\n- **errors**: ' + jobs_dict['errors'][0]
-            logging.error(msg)
+            logger.error(msg)
             raise Exception(msg)
     except TimeoutError:
         msg = '\n- **error_msg**: excute command=(%s) timeout after %i s' % (cmd, timeout)
-        logging.error(msg)
+        logger.error(msg)
         raise Exception(msg)
     return jobs_dict
 
 def judge_job_status(job):
     if 'errors' in job:
         msg = "\n- **errors**: %s" % (job['errors'])
-        logging.error(msg)
+        logger.error(msg)
         return (False, msg)
     if 'RUNNING' != job['status']:
         msg = "\n- **job_id**: %s, **status**: %s" % (job['id'],job['status'])
-        logging.error(msg)
+        logger.error(msg)
         return (False, msg)
     msg = "\n- **job_id**: %s, **status**: %s" % (job['id'],job['status'])
-    logging.info(msg)
+    logger.info(msg)
     return (True, msg)
 
 
 def python_callable(**kwargs):
-    logging.info('start kafka connect status analyze .')
+    logger.info('start kafka connect status analyze .')
 
     curl = 'curl '
     flink_jobmanager_url = 'http://'+ kwargs['task_instance'].task_id +':8081/jobs/'
 
     jobs_dict = get_jobs(curl + flink_jobmanager_url)
-    logging.info('exec cmd: ' + curl + flink_jobmanager_url + ' success!')
+    logger.info('exec cmd: ' + curl + flink_jobmanager_url + ' success!')
     jobs_list = jobs_dict['jobs']
     
     error_msg = ""
@@ -141,7 +144,7 @@ def python_callable(**kwargs):
         msg = "\n- **running_job/all_job**: %s/%s" % (running_job, all_job)
         error_msg += msg
         raise Exception(error_msg)
-    logging.info('Flink job status ok!')
+    logger.info('Flink job status ok!')
 
 for service in SERVICES:
     task = PythonOperator(