Update alarm_kafka_connect_v2.py

dags/kafka/alarm_kafka_connect_v2.py

@@ -14,7 +14,7 @@ import airflow
 #variable parameter
 DAG_ID = 'alarm_kafka_connect_v2'
 START_DATE = datetime.now() - timedelta(minutes=30)
-SCHEDULE_INTERVAL = timedelta(minutes=15)
+SCHEDULE_INTERVAL = timedelta(minutes=5)
 
 SERVICES = [
 'es-gmei-prod-cp-kafka-connect',
@@ -28,7 +28,7 @@ class TimeoutError(Exception):
 
 def run_command(cmd, timeout=60):
     is_linux = platform.system() == 'Linux'
-    p = subprocess.Popen(cmd, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setsid if is_linux else None)
+    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setsid if is_linux else None)
     t_beginning = time.time()
     seconds_passed = 0
     while True:
@@ -80,16 +80,12 @@ def get_kafka_connectors_status(cmd):
 
     try:
         result = run_command(cmd, timeout)
-        result = result.decode()
+        #result = result.decode()
         logging.info(result)
-        if 'Could not resolve host' in result:
+        if not result:
             msg = '\n- **error_msg**: Could not resolve host %s' % (cmd)
             raise Exception(msg)
-        resMsg = result.split('\n')[-1]
-        logging.info("result of last line to resMag : " + resMsg)
-        if '--:--:--' in resMsg:
-            raise TimeoutError(cmd, timeout)
-        connectors_dict = json.loads(resMsg)
+        connectors_dict = json.loads(result)
         if 'error_code' in connectors_dict:
             msg = '\n- **error_code**: ' + connectors_dict['error_code'] + '**error_msg**: ' + connectors_dict['message']
             logging.error(msg)