Merge branch 'master' of http://gogsb.soaringnova.com/sxwl_DL/datax-admin

app/routers/run.py

@@ -109,11 +109,9 @@ def add_airflow_run(item: Item, db: Session = Depends(get_db)):
 @web_try()
 @sxtimeit
 def add_notification(item: Item, db: Session = Depends(get_db)):
-    k8s_tool = KubernetesTools()
-    labels = {"dag_id": item.data['dag_id'], "task_id": item.data['task_id'], "run_ts": item.data['run_ts']}
-    logs = k8s_tool.get_pod_logs(namespaces="airflow", labels=labels)
+    uri = f"dags/{item.data['dag_id']}/dagRuns/{item.data['af_run_id']}/taskInstances/{item.data['task_id']}/logs/1"
+    logs = call_airflow_api("get", uri, {}).text
     job_item = crud.get_airflow_job_once(db=db, item_id=item.data["job_id"])
-
     if job_item.job_mode == 1:  # normal model, one job-> many runs
         run = crud.get_airflow_run_once_normal_mode(af_run_id=item.data['af_run_id'], db=db)
     elif job_item.job_mode == 2:  # debug model, one job-> one run

app/services/dag.py

@@ -122,7 +122,7 @@ def get_tmp_table_name(dag_uuid: str, node_id: str, out_pin: str, db: Session):
             task_id = t_id
             break
     if task_id:
-        table_name = f'{database_name}.job{job_id}_task{task_id}_subnode{node_id}_output{out_pin}_tmp'
+        table_name = f'job{job_id}_task{task_id}_subnode{node_id}_output{out_pin}_tmp'
         t_list = hiveDs.list_tables()
         if table_name.lower() not in t_list:
             raise Exception('该节点不存在中间结果')

configs/settings.py

@@ -20,8 +20,6 @@ class DefaultOption(dict):
 config = configparser.ConfigParser()
 if os.environ.get('APP_ENV', 'development') == 'development':
     config.readfp(open('development.ini'))
-    with open('development.ini') as f:
-        print(f.read())
 elif os.environ.get('APP_ENV') == 'production':
     config.readfp(open('production.ini'))