Commit ce4fafe4 authored by BO ZHANG's avatar BO ZHANG 🏀
Browse files

feat(dags): add msc l1 qc0 and mbi dags with docker support

parent dbef991b

docker-celery-3.0.5/dags/csst-msc-l1-mbi.py

0 → 100644
+61 −0
Original line number Diff line number Diff line 
from airflow import DAG

from airflow.providers.docker.operators.docker import DockerOperator

from airflow.operators.bash import BashOperator

from datetime import datetime, timedelta



default_args = {

    "owner": "airflow",

    "depends_on_past": False,

    "email": ["bozhang@nao.cas.cn"],

    "email_on_failure": False,

    "email_on_retry": False,

    "retries": 1,

    "retry_delay": timedelta(minutes=5),

    "params": {

        "input": '{"dag_id":"csst-msc-l1-mbi","dag_run_id":"csst-msc-l1-mbi-inttest","dataset":"test-msc-c9-25sqdeg-v3","instrument":"MSC","obs_type":"WIDE","obs_group":"W5","obs_id":"10100131914","detector":"09","batch_id":"inttest", "docker_images":{"csst-msc-l1-mbi":"latest"}}'

    },

}



with DAG(

    dag_id="csst-msc-l1-mbi",

    default_args=default_args,

    description="Run MSC L1 MBI container",

    start_date=datetime(2023, 1, 1),

    schedule=None,

    catchup=False,

    tags={"csst", "msc", "l1", "mbi"},

) as dag:

    # Create main output directory and subdirectory for MBI

    create_dirs = BashOperator(

        task_id="create_output_dirs",

        bash_command="mkdir -p /tmp/csst-msc-l1-mbi-{{ (params.input | fromjson).get('dag_run_id', run_id) }}/mbi"

    )

    

    # Run MBI container

    run_mbi = DockerOperator(

        task_id="run_msc_l1_mbi",

        image="csu-harbor.csst.nao:10443/csst/csst-msc-l1-mbi:{{ (params.input | fromjson).get('docker_images', {}).get('csst-msc-l1-mbi', 'latest') }}",

        command="run {{ params.input }}",

        docker_url="unix://var/run/docker.sock",

        network_mode="bridge",

        auto_remove=True,

        force_pull=True,

        mount_tmp_dir=False,

        volumes=["/tmp/csst-msc-l1-mbi-{{ (params.input | fromjson).get('dag_run_id', run_id) }}/mbi:/pipeline/output", "/home/cham/PycharmProjects/csst-airflow/docker-celery-3.0.5/volumes/env:/env"],

        env_file="/env/common.env",

        template_fields=["image", "command", "volumes"],

    )

    

    # Concatenate all log files into one

    concatenate_logs = BashOperator(

        task_id="concatenate_logs",

        bash_command="echo '=== Concatenated Pipeline Logs ===' > /tmp/csst-msc-l1-mbi-{{ (params.input | fromjson).get('dag_run_id', run_id) }}/combined.log && find /tmp/csst-msc-l1-mbi-{{ (params.input | fromjson).get('dag_run_id', run_id) }} -name 'pipeline.log' -exec echo '=== Log from {} ===' >> /tmp/csst-msc-l1-mbi-{{ (params.input | fromjson).get('dag_run_id', run_id) }}/combined.log \; -exec cat {} >> /tmp/csst-msc-l1-mbi-{{ (params.input | fromjson).get('dag_run_id', run_id) }}/combined.log \;"

    )

    

    # Display the concatenated log content

    display_combined_logs = BashOperator(

        task_id="display_combined_logs",

        bash_command="cat /tmp/csst-msc-l1-mbi-{{ (params.input | fromjson).get('dag_run_id', run_id) }}/combined.log"

    )

    

    create_dirs >> run_mbi >> concatenate_logs >> display_combined_logs
 
 No newline at end of file

docker-celery-3.0.5/dags/csst-msc-l1-qc0.py

0 → 100644
+55 −0
Original line number Diff line number Diff line 
from airflow import DAG

from airflow.providers.docker.operators.docker import DockerOperator

from airflow.operators.bash import BashOperator

from datetime import datetime, timedelta



default_args = {

    "owner": "airflow",

    "depends_on_past": False,

    "email": ["bozhang@nao.cas.cn"],

    "email_on_failure": False,

    "email_on_retry": False,

    "retries": 1,

    "retry_delay": timedelta(minutes=5),

    "params": {

        "input": '{"dag_id":"csst-msc-l1-qc0","dag_run_id":"csst-msc-l1-qc0-inttest","dataset":"test-msc-c9-25sqdeg-v3","instrument":"MSC","obs_type":"WIDE","obs_group":"W5","obs_id":"10100131914","detector":"09","batch_id":"inttest", "docker_images":{"csst-msc-l1-qc0":"latest"}}'

    },

}



with DAG(

    dag_id="csst-msc-l1-qc0",

    default_args=default_args,

    description="Run MSC L1 QC0 container",

    start_date=datetime(2023, 1, 1),

    schedule=None,

    catchup=False,

    tags={"csst", "msc", "l1", "qc0"},

) as dag:

    # Create output directory

    create_dir = BashOperator(

        task_id="create_output_dir",

        bash_command="mkdir -p /tmp/csst-msc-l1-qc0-{{ (params.input | fromjson).get('dag_run_id', run_id) }}"

    )

    

    # Run the container with volume mount

    run_qc0 = DockerOperator(

        task_id="run_msc_l1_qc0",

        image="csu-harbor.csst.nao:10443/csst/csst-msc-l1-qc0:{{ (params.input | fromjson).get('docker_images', {}).get('csst-msc-l1-qc0', 'latest') }}",

        command="run {{ params.input }}",

        docker_url="unix://var/run/docker.sock",

        network_mode="bridge",

        auto_remove=True,

        force_pull=True,

        mount_tmp_dir=False,

        volumes=["/tmp/csst-msc-l1-qc0-{{ (params.input | fromjson).get('dag_run_id', run_id) }}:/pipeline/output", "/home/cham/PycharmProjects/csst-airflow/docker-celery-3.0.5/volumes/env:/env"],

        env_file="/env/common.env",

        template_fields=["image", "command", "volumes"],

    )

    

    # Capture and display the log file content

    capture_logs = BashOperator(

        task_id="capture_logs",

        bash_command="echo '=== Pipeline Log Content ===' && cat /tmp/csst-msc-l1-qc0-{{ (params.input | fromjson).get('dag_run_id', run_id) }}/pipeline.log || echo 'Log file not found'"

    )

    

    create_dir >> run_qc0 >> capture_logs
 
 No newline at end of file

docker-celery-3.0.5/docker-compose.yaml

+1 −0
Original line number Diff line number Diff line
@@ -121,6 +121,7 @@ x-airflow-common: 
    - ${AIRFLOW_PROJ_DIR:-.}/volumes/logs:/opt/airflow/logs

    - ${AIRFLOW_PROJ_DIR:-.}/volumes/config:/opt/airflow/config

    - ${AIRFLOW_PROJ_DIR:-.}/volumes/plugins:/opt/airflow/plugins

    - /var/run/docker.sock:/var/run/docker.sock

  user: "${AIRFLOW_UID:-50000}:0"

  depends_on:

    &airflow-common-depends-on

docker-celery-3.0.5/envs/.env.csu

+1 −0
Original line number Diff line number Diff line
@@ -6,3 +6,4 @@ _AIRFLOW_WWW_USER_USERNAME=airflow 
_AIRFLOW_WWW_USER_PASSWORD=airflow

JWT_SECRET=eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhaXJmbG93IiwiaWF0IjoxNzU1NDQ3NjMxLCJleHAiOjE3ODY5ODM2MzF9.CbxQayEt370iHhvYXhHMESsk1IWmi8QLMOcctIlbicXhybNtBbfboDUdUDMClwvkSuTjC0AOeSHL23pGyy0plQ

MASTER_IP=192.168.25.153

_PIP_ADDITIONAL_REQUIREMENTS=apache-airflow-providers-docker