Merge COSIDAG development branch with cleaned examples and workflow improvements #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open

falric05 wants to merge 99 commits into main from pr/dev_bgo

.gitignore

-Original file line number
+Diff line change
@@ Expand Up / @@ -160,3 +160,9 @@ cython_debug/ @@
     #  and can be added to the global gitignore or merged into this file.  For a more nuclear
     #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
     #.idea/
+    # Data Folder
+    data/*
+    # TODO: Remove when you need the cut background file anymore
+    # Raw data folder excluding only the background cut file
+    data/raw/

README.md

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -1,101 +1,217 @@
  
    # cosiflow

    # Cosiflow

    The COSI SDOC pipeline based on Apache Airflow

    Cosiflow provides an Airflow-based orchestration environment for managing and monitoring scientific pipelines for COSI.

    ## Build the cosiflow docker

    ---

    We assume that the cosiflow repository is in your $HOME directory.

    ### 1. REQUIREMENTS

    ```bash

    cd $HOME/cosiflow/env

    ```

    #### PREPARE THE ENVIRONMENT FILE

    Mac:

    1. Copy and rename the `.env.example` file as `.env`:

       ```bash

       cd env

       cp .env.example .env

       ```

    ```bash

    docker build --platform linux/arm64 -t airflow:1.0.0 -f Dockerfile .

    ```

    2. Open the `.env` file and set a secure password for Airflow:

       ```bash

       AIRFLOW_ADMIN_PASSWORD=<YOUR_AIRFLOW_PASSWORD>

       ```

    Linux:

    3. Manually bootstrap user and group IDs for the container:

    ```bash

    docker build -t airflow:1.1.0 -f Dockerfile .

    ```

       ```bash

       id -u

       # Copy the output as YOUR_USER_ID

    ## Execute the docker compose to start containers

       id -g

       # Copy the output as YOUR_GROUP_ID

       ```

    ```bash

    docker compose up -d

    ```

    4. Open `.env` and modify the following environment variables:

    If you want to enter into the postgre docker container: `docker compose exec postgres bash`

       ```bash

       UID=<YOUR_USER_ID>

       GID=<YOUR_GROUP_ID>

       ```

    If you want to enter into the postgre docker container: `docker compose exec airflow bash`

    #### PREPARE THE DOCKERFILE

    ## Connect to the web server using a browser

    Open the `Dockerfile.airflow` file and paste the same values:

    localhost:8080

    ```dockerfile

    ARG UID=<YOUR_USER_ID>

    ARG GID=<YOUR_GROUP_ID>

    ```

    #### PREPARE THE FOLDER FOR STORING POSTGRESS DATA

    ```bash

    cd ..

    mkdir -p data/postgres_data

    ```

    Note: if you use a remote server you can change the `docker-compose.yaml` file to use another port.

    ---

    For example:

      ```yaml

      ports:

        - "28080:8080"

      ```

    ### 2. BUILD THE COMPOSE

    then from your local pc you can forward the port in this way:

    Build all containers defined in `docker-compose.yml`:

    ```bash

    ssh -N -L 28080:localhost:28080 [user]@[remote machine]

    cd env

    docker compose build

    ```

    and open the airflow webpace from your local pc at `localhost:28080`

    ⏱ Estimated build time: **~490 seconds**

    ---

    Login with username: `admin`  password: `<password>`

    ### 3. RUN THE CONTAINER

    To obtain the password `<password>` execute this command after the initialization of the containers

    To run with logs visible:

    ```bash

    docker compose logs | grep pass

    docker compose up

    ```

    ### Shutdown the dockers

    To run in detached mode (no logs):

    ```bash

    docker compose down -v

    docker compose up -d

    ```

    ## Test the cosipy DAG

    ---

    ### 4. ENTER THE CONTAINER

    Enter in the docker airflow

    To open a terminal inside the running Airflow container:

    ```bash

    docker compose exec airflow bash

    ```

    First download the data file from wasabi.

    ---

    ```bash

    cd /shared_dir/pipeline

    source activate cosipy

    python initialize_pipeline.py

    ```

    ### 5. CONNECT TO THE AIRFLOW WEB UI

    1. Open your web browser and go to:

    This script downloads the input file from wasabi and move it in `/home/gamma/workspace/data`

       [http://localhost:8080/home](http://localhost:8080/home)

    Now we must activate the DAG named `"cosipt_test_v0"` from the airflow website

    2. Insert the user credentials:

       ```text

       user:     admin

       password: <YOUR_AIRFLOW_PASSWORD>

       ```

    Then we have to copy the file in the input directory to trigger the DAG

    ---

    ### 6. STOP THE CONTAINER

    To stop and remove all running containers, networks, and volumes:

    ```bash

    cd /home/gamma/workspace/data

    cp GalacticScan.inc1.id1.crab2hr.extracted.tra.gz input

    docker compose down -v

    ```

    We should see that the DAG started to process the data.

    ---

    ### 7. CONFIGURATIONS

    Below is the list of environment variables defined in `.env` with their purpose:

    | Variable | Description |

    |-----------|--------------|

    | **UID** | User ID for container bootstrap |

    | **GID** | Group ID for container bootstrap |

    | **DISPLAY** | Display variable for X11 forwarding (optional) |

    | **AIRFLOW_ADMIN_USERNAME** | Default Airflow Web UI username |

    | **AIRFLOW_ADMIN_EMAIL** | Email associated with Airflow admin user |

    | **AIRFLOW_ADMIN_PASSWORD** | Secure password for Airflow Web UI |

    | **ALERT_USERS_LIST_PATH** | Path to YAML file containing user alert configurations |

    | **ALERT_SMTP_SERVER** | SMTP server used for alert notifications |

    | **ALERT_SMTP_PORT** | Port of the SMTP server |

    | **ALERT_EMAIL_SENDER** | Email address used as sender for system alerts |

    | **ALERT_LOG_PATH** | Path to Airflow log file monitored by alert system |

    | **AIRFLOW__SMTP__SMTP_STARTTLS** | Enables/disables STARTTLS (default: False) |

    | **AIRFLOW__SMTP__SMTP_SSL** | Enables/disables SMTP over SSL (default: False) |

    | **MAILHOG_WEBUI_URL** | URL for MailHog web interface (for testing alerts) |

    | **COSI_DATA_DIR** | Root directory for COSI data |

    | **COSI_INPUT_DIR** | Directory for COSI input data |

    | **COSI_LOG_DIR** | Directory for COSI log files |

    | **COSI_OBS_DIR** | Directory for observation data |

    | **COSI_TRANSIENT_DIR** | Directory for transient event data |

    | **COSI_TRIGGER_DIR** | Directory for trigger event data |

    | **COSI_MAPS_DIR** | Directory for map data products |

    | **COSI_SOURCE_DIR** | Directory for source-level data products |

    ---

    ### NOTES

    - Make sure `.env` and `docker-compose.yml` are located in the same directory.

    - Do **not** commit your personal `.env` file to version control.

    - To inspect container logs, use:

      ```bash

      docker compose logs -f airflow

      ```

    ---

    **Cosiflow environment ready for use.**

    ---

    ## What is COSIDAG

    A **COSIDAG** (COSI DAG) is a structured abstraction built on top of Apache Airflow DAGs.

    It provides a **standardized workflow layout** for scientific pipelines, reducing boilerplate and enforcing consistent patterns across different analyses.

    In particular, a COSIDAG:

    * defines a common execution skeleton (input resolution, optional monitoring, result handling)

    * encapsulates best practices for:

      * file discovery

      * parameter propagation

      * XCom-based communication

    * allows developers to focus only on **scientific tasks**, while orchestration logic is handled automatically

    COSIDAGs are used for all production scientific pipelines (e.g. Light Curve, TS Map), while standard DAGs are reserved for orchestration, testing, or utilities.

    **How to write and customize a COSIDAG** is explained in detail in the [tutorial section](modules/README.md).

    ---

    ## Tutorials and developer guide

    A complete, step-by-step guide on how to:

    * understand the COSIDAG execution model

    * write new COSIDAGs

    * add custom tasks

    * use XCom correctly

    * integrate external Python environments

    is available in:

    [tutorial section](tutorials/README.md).

    This is the **recommended starting point for developers**.

    ---

    ## Available DAGs and COSIDAGs

    A complete and up-to-date list of all DAGs and COSIDAGs implemented in this repository — including:

    * workflow purpose

    * inputs and outputs

    * task structure

    * operators used

    * XCom usage

    This directory `/home/gamma/workspace/heasarc/dl0` contains several folders with this format `2025-01-24_14-31-56`.

    is documented in: [DAG and COSIDAG LIST README](dags/README.md)

    Inside the folder we have the results of the analysis.

    This document serves as the **catalog and reference** for all workflows available in Cosiflow.

callbacks/__init__.py

Empty file.

callbacks/on_failure_callback.py

-Original file line number
+Diff line change
@@ -0,0 +1,82 @@
+    import yaml
+    import os
+    from airflow.utils.email import send_email
+    import urllib.parse
+    ALERT_CONFIG_PATH = os.getenv("ALERT_USERS_LIST_PATH", "/home/gamma/env/alert_users.yaml")
+    def load_alert_config():
+        with open(ALERT_CONFIG_PATH, "r") as f:
+            return yaml.safe_load(f)
+    def get_recipients(keyword: str) -> list[str]:
+        config = load_alert_config()
+        matched_groups = set()
+        for rule in config.get("rules", []):
+            if rule["pattern"] == keyword:
+                matched_groups.update(rule["notify"])
+        emails = set()
+        for group in matched_groups:
+            group_data = config["groups"].get(group)
+            if group_data:
+                emails.update(group_data.get("emails", []))
+        return sorted(emails)
+    def notify_email(context):
+        task = context["task_instance"]
+        dag_id = task.dag_id
+        task_id = task.task_id
+        run_id = task.run_id
+        execution_date = context.get("execution_date")
+        # URL-encode i parametri per sicurezza
+        base_url = "http://localhost:8080"
+        query = urllib.parse.urlencode({
+            "execution_date": execution_date.isoformat(),
+            "tab": "logs",
+            "dag_run_id": run_id,
+            "task_id": task_id
+        })
+        log_url = f"{base_url}/dags/{dag_id}/grid?{query}"
+        # Percorso log locale (personalizzabile)
+        log_path = f"/home/gamma/airflow/logs/dag_id={dag_id}/run_id={run_id}/task_id={task_id}/attempt=1.log"
+        if not os.path.exists(log_path):
+            log_preview = "⚠️ Log file not found."
+        else:
+            with open(log_path, "r") as f:
+                lines = f.readlines()[-30:]  # Ultime 30 righe
+                log_preview = "".join(lines)
+                log_preview = log_preview.replace("<", "&lt;").replace(">", "&gt;")  # Escaping HTML
+        recipients = get_recipients("ALERT_FAIL")
+        if not recipients:
+            return  # no recipients, skip
+        subject = f"[ALERT] Task {task.task_id} in DAG {task.dag_id} has failed"
+        html_content = f"""
+        <html>
+          <body style="font-family:Arial, sans-serif; font-size:14px; color:#333;">
+            <h2 style="color:#c0392b;">⚠️ Task Failure Alert</h2>
+            <table style="border-collapse:collapse;">
+              <tr><td><strong>DAG:</strong></td><td>{dag_id}</td></tr>
+              <tr><td><strong>Task:</strong></td><td>{task_id}</td></tr>
+              <tr><td><strong>Execution Time:</strong></td><td>{execution_date}</td></tr>
+              <tr><td><strong>Log URL:</strong></td><td><a href="{log_url}">{log_url}</a></td></tr>
+            </table>
+            <h3 style="margin-top:20px;">🔍 Log Preview</h3>
+            <pre style="background-color:#f5f5f5; padding:10px; border:1px solid #ccc; max-height:300px; overflow:auto;">
+    {log_preview}
+            </pre>
+            <p style="color:#888; font-size:12px;">Full log available at the link above.</p>
+          </body>
+        </html>
+        """
+        send_email(to=recipients, subject=subject, html_content=html_content)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Merge COSIDAG development branch with cleaned examples and workflow improvements #7

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Merge COSIDAG development branch with cleaned examples and workflow improvements #7

Are you sure you want to change the base?

Uh oh!

Merge COSIDAG development branch with cleaned examples and workflow improvements #7

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!