updates for hpc jobs

dags/dag1_calculate_hpc_worker.py

@@ -98,9 +98,9 @@ with dag:
       queue='igf-lims',
       command="""
       {% if ti.xcom_pull(key=params.job_name,task_ids="calculate_new_worker_size_and_branch" ) > 1 %}
-        source /etc/bashrc;qsub -k n -m n -N {{ params.job_name }} -J 1-{{ ti.xcom_pull(key=params.job_name,task_ids="calculate_new_worker_size_and_branch" ) }}  {{ params.pbs_resource }} -- /project/tgu/data2/airflow_test/github/igf-airflow-hpc/scripts/hpc/airflow_worker.sh {{  params.airflow_queue }} {{ params.job_name }}
+        source /etc/bashrc;qsub -o /dev/null -e /dev/null -k n -m n -N {{ params.job_name }} -J 1-{{ ti.xcom_pull(key=params.job_name,task_ids="calculate_new_worker_size_and_branch" ) }}  {{ params.pbs_resource }} -- /project/tgu/data2/airflow_test/github/igf-airflow-hpc/scripts/hpc/airflow_worker.sh {{  params.airflow_queue }} {{ params.job_name }}
       {% else %}
-        source /etc/bashrc;qsub -k n -m n -N {{ params.job_name }} {{ params.pbs_resource }} -- /project/tgu/data2/airflow_test/github/igf-airflow-hpc/scripts/hpc/airflow_worker.sh {{  params.airflow_queue }} {{ params.job_name }}
+        source /etc/bashrc;qsub -o /dev/null -e /dev/null -k n -m n -N {{ params.job_name }} {{ params.pbs_resource }} -- /project/tgu/data2/airflow_test/github/igf-airflow-hpc/scripts/hpc/airflow_worker.sh {{  params.airflow_queue }} {{ params.job_name }}
       {% endif %}
       """,
       params={'pbs_resource':pbs_resource,'airflow_queue':airflow_queue,'job_name':q}

dags/test_dag.py

@@ -22,7 +22,7 @@ with dag:
       dag=dag,
       bash_command='hostname -A',
       queue='hpc_4G',
-      do_xcom_push=True
+      xcom_push=True
     )
   task2 = \
     BashOperator(
@@ -30,6 +30,6 @@ with dag:
       dag=dag,
       bash_command='hostname -A',
       queue='hpc_1G',
-      do_xcom_push=True
+      xcom_push=True
     )
   task1 >> task2

docker/airflow_env_template

@@ -4,9 +4,9 @@ AIRFLOW__WEBSERVER__WEB_SERVER_SSL_CERT=/SSL/airflow.cert
 AIRFLOW__WEBSERVER__WEB_SERVER_SSL_KEY=/SSL/airflow.key
 AIRFLOW__WEBSERVER__DAG_ORIENTATION=TB
 AIRFLOW__SCHEDULER__CHILD_PROCESS_LOG_DIRECTORY=/rds/general/user/igf/home/data2/airflow_test/logs/scheduler
-AIRFLOW__LOGGING__BASE_LOG_FOLDER=/rds/general/user/igf/home/data2/airflow_test/logs
-AIRFLOW__LOGGING__DAG_PROCESSOR_MANAGER_LOG_LOCATION=/rds/general/user/igf/home/data2/airflow_test/logs/dag_processor_manager/dag_processor_manager.log
-AIRFLOW__LOGGING__LOGGING_LEVEL=WARN
+AIRFLOW__CORE__BASE_LOG_FOLDER=/rds/general/user/igf/home/data2/airflow_test/logs
+AIRFLOW__CORE__DAG_PROCESSOR_MANAGER_LOG_LOCATION=/rds/general/user/igf/home/data2/airflow_test/logs/dag_processor_manager/dag_processor_manager.log
+AIRFLOW__CORE__LOGGING_LEVEL=WARN
 AIRFLOW__CORE__DAGS_FOLDER=/rds/general/user/igf/home/data2/airflow_test/github/igf-airflow-hpc/dags
 AIRFLOW__CORE__PLUGIN_FOLDER=/rds/general/user/igf/home/data2/airflow_test/plugin
 AIRFLOW__CORE__SQL_ALCHEMY_CONN=''

scripts/hpc/airflow_worker.sh

@@ -5,4 +5,4 @@ job_name=${2:?'Missing job name'}
 
 source /rds/general/user/igf/home/data2/airflow_test/secrets/hpc_env.sh
 
-airflow worker --pid $TMPDIR -cn ${PBS_JOBID}-${job_name} -q ${job_queue}
+airflow worker --pid $TMPDIR -cn ${PBS_JOBID}-${job_name} -q ${job_queue} --skip_serve_logs --log-file ${AIRFLOW__CORE__BASE_LOG_FOLDER}

scripts/hpc/hpc_env_template.sh

@@ -9,9 +9,9 @@ export AIRFLOW__WEBSERVER__WEB_SERVER_SSL_CERT=/SSL/airflow.cert
 export AIRFLOW__WEBSERVER__WEB_SERVER_SSL_KEY=/SSL/airflow.key
 export AIRFLOW__WEBSERVER__DAG_ORIENTATION=TB
 export AIRFLOW__SCHEDULER__CHILD_PROCESS_LOG_DIRECTORY=/rds/general/user/igf/home/data2/airflow_test/logs/scheduler
-export AIRFLOW__LOGGING__BASE_LOG_FOLDER=/rds/general/user/igf/home/data2/airflow_test/logs
-export AIRFLOW__LOGGING__DAG_PROCESSOR_MANAGER_LOG_LOCATION=/rds/general/user/igf/home/data2/airflow_test/logs/dag_processor_manager/dag_processor_manager.log
-export AIRFLOW__LOGGING__LOGGING_LEVEL=WARN
+export AIRFLOW__CORE__BASE_LOG_FOLDER=/rds/general/user/igf/home/data2/airflow_test/logs
+export AIRFLOW__CORE__DAG_PROCESSOR_MANAGER_LOG_LOCATION=/rds/general/user/igf/home/data2/airflow_test/logs/dag_processor_manager/dag_processor_manager.log
+export AIRFLOW__CORE__LOGGING_LEVEL=WARN
 export AIRFLOW__CORE__DAGS_FOLDER=/rds/general/user/igf/home/data2/airflow_test/github/igf-airflow-hpc/dags
 export AIRFLOW__CORE__PLUGIN_FOLDER=/rds/general/user/igf/home/data2/airflow_test/plugin
 export AIRFLOW__CORE__SQL_ALCHEMY_CONN=''