modify subprocess popen stdout and stderr

dags/flink/alarm_flink_job.py

@@ -29,7 +29,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, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setsid if is_linux else None)
     t_beginning = time.time()
     seconds_passed = 0
     while True:
@@ -44,7 +44,7 @@ def run_command(cmd, timeout=60):
                 p.terminate()
             raise TimeoutError(cmd, timeout)
         time.sleep(0.1)
-    return p.stdout.read()
+    return (p.stdout.read(), p.stderr.read())
 
 default_args = {
     'owner': 'tangxianggang',
@@ -80,12 +80,19 @@ def get_jobs(cmd):
     timeout = 10
 
     try:
-        result = run_command(cmd, timeout)
-        result = result.decode()
-        if 'Could not resolve host' in result:
+        (stdout, stderr) = run_command(cmd, timeout)
+        stdout = stdout.decode()
+        stderr = stderr.decode()
+        logging.debug(stderr)
+        logging.debug(stdout)
+
+        if 'Could not resolve host' in stderr and not stdout:
             msg = '\n- **error_msg**: Could not resolve host %s' % (cmd)
             raise Exception(msg)
-        jobs_dict = json.loads(result.split('\n')[-1])
+        elif not stdout:
+            msg = '\n- **error_msg**: connectors is empty'
+            raise Exception(msg)
+        jobs_dict = json.loads(stdout)
         if 'errors' in jobs_dict:
             msg = '\n- **errors**: ' + jobs_dict['errors'][0]
             logging.error(msg)

dags/kafka/alarm_kafka_connect.py

@@ -30,7 +30,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, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setsid if is_linux else None)
     t_beginning = time.time()
     seconds_passed = 0
     while True:
@@ -45,7 +45,7 @@ def run_command(cmd, timeout=60):
                 p.terminate()
             raise TimeoutError(cmd, timeout)
         time.sleep(0.1)
-    return p.stdout.read()
+    return (p.stdout.read(), p.stderr.read())
 
 default_args = {
     'owner': 'tangxianggang',
@@ -81,12 +81,19 @@ def get_kafka_connectors(cmd):
     timeout = 10
 
     try:
-        result = run_command(cmd, timeout)
-        result = result.decode()
-        if 'Could not resolve host' in result:
+        (stdout, stderr) = run_command(cmd, timeout)
+        stdout = stdout.decode()
+        stderr = stderr.decode()
+        logging.debug(stderr)
+        logging.debug(stdout)
+
+        if 'Could not resolve host' in stderr and not stdout:
             msg = '\n- **error_msg**: Could not resolve host %s' % (cmd)
             raise Exception(msg)
-        connectors_list = json.loads(result.split('\n')[-1])
+        elif not stdout:
+            msg = '\n- **error_msg**: connectors is empty'
+            raise Exception(msg)
+        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)
@@ -105,16 +112,19 @@ def get_connector_status(cmd):
     outdict = {}
 
     try:
-        result = run_command(cmd, timeout)
-        result = result.decode()
-        outdict = json.loads(result.split('\n')[-1])
+        (stdout, stderr) = run_command(cmd, timeout)
+        stdout = stdout.decode()
+        stderr = stderr.decode()
+        logging.debug(stderr)
+        logging.debug(stdout)
+
+        outdict = json.loads(stdout)
     except TimeoutError:
         msg = '\n- **error_msg**: excute command=(%s) timeout after %i s' % (cmd, timeout)
         logging.error(msg)
         errdict = {'error_code':'600','message':'excute command=(%s) timeout after %i s' % (cmd, timeout)}
         return errdict
 
-    logging.info('get connector status : \n' + str(outdict))
     return outdict
 
 def judge_connector_status(connector_status_dict, connector):

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=5)
+SCHEDULE_INTERVAL = timedelta(minutes=15)
 
 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, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setsid if is_linux else None)
+    p = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=True, preexec_fn=os.setsid if is_linux else None)
     t_beginning = time.time()
     seconds_passed = 0
     while True:
@@ -43,7 +43,7 @@ def run_command(cmd, timeout=60):
                 p.terminate()
             raise TimeoutError(cmd, timeout)
         time.sleep(0.1)
-    return p.stdout.read()
+    return (p.stdout.read(), p.stderr.read())
 
 default_args = {
     'owner': 'tangxianggang',
@@ -79,13 +79,19 @@ def get_kafka_connectors_status(cmd):
     timeout = 10
 
     try:
-        result = run_command(cmd, timeout)
-        #result = result.decode()
-        logging.info(result)
-        if not result:
+        (stdout, stderr) = run_command(cmd, timeout)
+        stdout = stdout.decode()
+        stderr = stderr.decode()
+        logging.debug(stderr)
+        logging.debug(stdout)
+
+        if 'Could not resolve host' in stderr and not stdout:
             msg = '\n- **error_msg**: Could not resolve host %s' % (cmd)
             raise Exception(msg)
-        connectors_dict = json.loads(result)
+        elif not stdout:
+            msg = '\n- **error_msg**: connectors is empty'
+            raise Exception(msg)
+        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)