Update alarm_kafka_connect.py

dags/kafka/alarm_kafka_connect.py

@@ -25,6 +25,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
 
@@ -71,12 +74,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(cmd):
-    logging.info('exec cmd:' + cmd)
+    logger.info('exec cmd:' + cmd)
     connectors_list = []
     timeout = 10
 
@@ -84,8 +87,8 @@ def get_kafka_connectors(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)
@@ -96,18 +99,18 @@ def get_kafka_connectors(cmd):
         connectors_list = json.loads(stdout)
         if 'error_code' in connectors_list:
             msg = '\n- **error_code**: ' + connectors_list['error_code'] + '**error_msg**: ' + connectors_list['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_list))
+    logger.info(str(connectors_list))
     return connectors_list
 
 def get_connector_status(cmd):
-    logging.info('exec cmd: ' + cmd)
+    logger.info('exec cmd: ' + cmd)
     timeout = 10
     outdict = {}
 
@@ -115,13 +118,13 @@ def get_connector_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)
 
         outdict = json.loads(stdout)
     except TimeoutError:
         msg = '\n- **error_msg**: excute command=(%s) timeout after %i s' % (cmd, timeout)
-        logging.error(msg)
+        logger.error(msg)
         errdict = {'error_code':'600','message':'excute command=(%s) timeout after %i s' % (cmd, timeout)}
         return errdict
 
@@ -134,11 +137,11 @@ def judge_connector_status(connector_status_dict, connector):
             msg = msg = "\n- **connector_name**: %s, **error_msg**: %s" % (connector, connector_status_dict['message'])
         else:
             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['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['tasks']
     error_tasks = 0
@@ -149,19 +152,19 @@ 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/'
     status = '/status'
 
     connectors_list = get_kafka_connectors(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_list:
@@ -171,7 +174,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(