Update alarm_kafka_connect_v2.py

dags/kafka/alarm_kafka_connect_v2.py

@@ -24,6 +24,9 @@ SERVICES = [
 DINGDING = 'https://oapi.dingtalk.com/robot/send?access_token=dac084248b38ef564c30e7f7d0c3901f3967c8e5ffdb33efe188495d5b058fdd'
 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_kafka_connectors_status(cmd):
-    logging.info('exec cmd:' + cmd)
+    logger.info('exec cmd:' + cmd)
     connectors_dict = []
     timeout = 10
 
@@ -83,8 +86,8 @@ def get_kafka_connectors_status(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,24 +98,24 @@ def get_kafka_connectors_status(cmd):
         connectors_dict = json.loads(stdout)
         if 'error_code' in connectors_dict:
             msg = '\n- **error_code**: ' + connectors_dict['error_code'] + '**error_msg**: ' + connectors_dict['message']
-            logging.error(msg)
+            logger.error(msg)
             raise Exception(msg)
     except TimeoutError:
         msg = '\n- **error_msg**: excute command=(%s) timeout after %i' % (cmd, timeout)
-        logging.error(msg)
+        logger.error(msg)
         raise Exception(msg)
 
-    logging.info(str(connectors_dict))
+    logger.info(str(connectors_dict))
     return connectors_dict
 
 def judge_connector_status(connector_status_dict, connector):
     if 'error_code' in connector_status_dict:
         msg = "\n- **connector_name**: %s, **error_code**: %s, **error_msg**: %s" % (connector, connector_status_dict['error_code'], connector_status_dict['message'])
-        logging.error(msg)
+        logger.error(msg)
         return (False, msg)
     if 'RUNNING' != connector_status_dict['status']['connector']['state']:
         msg = "\n- **connector_name**: %s, **is_running**: false" % (connector)
-        logging.error(msg)
+        logger.error(msg)
         return (False, msg)
     tasks_list = connector_status_dict['status']['tasks']
     error_tasks = 0
@@ -123,18 +126,18 @@ def judge_connector_status(connector_status_dict, connector):
         all_tasks = len(tasks_list)
         running_tasks = all_tasks - error_tasks
         msg = "\n- **connector_name**: %s, **is_running**: true, **running_tasks/all_tasks**: %s/%s" % (connector, str(running_tasks), str(all_tasks) )
-        logging.error(msg)
+        logger.error(msg)
         return (False, msg)
     return (True, str(connector_status_dict))
 
 def python_callable(**kwargs):
-    logging.info('start kafka connect status analyze .')
+    logger.info('start kafka connect status analyze .')
 
     curl = 'curl '
     kafka_connect_url = 'http://' + kwargs['task_instance'].task_id + ':8083/connectors/?expand=status'
 
     connectors_dict = get_kafka_connectors_status(curl + kafka_connect_url)
-    logging.info('exec cmd: ' + curl + kafka_connect_url + ' success!')
+    logger.info('exec cmd: ' + curl + kafka_connect_url + ' success!')
 
     error_msg = ""
     for connector in connectors_dict:
@@ -143,7 +146,7 @@ def python_callable(**kwargs):
             error_msg += msg
     if error_msg:
         raise Exception(error_msg)
-    logging.info('kafka connect status ok!')
+    logger.info('kafka connect status ok!')
 
 for service in SERVICES:
     task = PythonOperator(