als-computing · davramov · Feb 11, 2025 · Feb 11, 2025 · Feb 11, 2025 · Feb 11, 2025
diff --git a/.env.example b/.env.example
@@ -1,7 +1,15 @@
-GLOBUS_CLIENT_ID=<globus_client_id>
-GLOBUS_CLIENT_SECRET=<globus_client_secret>
-PREFECT_API_URL=<url_of_prefect_server>
-PREFECT_API_KEY=<prefect_client_secret>
-PUSHGATEWAY_URL=<url_of_pushgateway_server>
-JOB_NAME=<jobname_for_pushgateway>
-INSTANCE_LABEL=<label_for_pushgateway>
+GLOBUS_CLIENT_ID=<globus_client_id>                 # For Globus Transfer
+GLOBUS_CLIENT_SECRET=<globus_client_secret>         # For Globus Transfer
+GLOBUS_COMPUTE_CLIENT_ID=<globus_client_id>         # For ALCF Jobs
+GLOBUS_COMPUTE_CLIENT_SECRET=<globus_client_secret> # For ALCF Jobs
+GLOBUS_COMPUTE_ENDPOINT=<globus_compute_endpoint>   # For ALCF Jobs
+PREFECT_API_URL=<url_of_prefect_server>             # For Prefect Flows
+PREFECT_API_KEY=<prefect_client_secret>             # For Prefect Flows
+SCICAT_API_URL=<url_of_scicat_api>                  # For SciCat Ingest
+SCICAT_INGEST_USER=<scicat_ingest_user>             # For SciCat Ingest
+SCICAT_INGEST_PASSWORD=<scicat_ingest_password>     # For SciCat Ingest
+PATH_NERSC_CLIENT_ID=<path_nersc_client_id>         # For NERSC SFAPI
+PATH_NERSC_PRI_KEY=<path_nersc_private_key>         # For NERSC SFAPI
+PUSHGATEWAY_URL=<url_of_pushgateway_server>         # For Grafana Pushgateway
+JOB_NAME=<jobname_for_pushgateway>                  # For Grafana Pushgateway
+INSTANCE_LABEL=<label_for_pushgateway>              # For Grafana Pushgateway
diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml
@@ -12,10 +12,10 @@ jobs:
 
     steps:
     - uses: actions/checkout@v2
-    - name: Set up Python 3.11
+    - name: Set up Python 3.12.5
       uses: actions/setup-python@v5
       with:
-        python-version: 3.11
+        python-version: 3.12.5
         cache: 'pip'
     - name: Install dependencies
       run: |

diff --git a/README.md b/README.md
@@ -22,10 +22,18 @@ $   pip3 install -e .
 Use `.env.example` as a template.
 
 ```
-GLOBUS_CLIENT_ID=<globus_client_id>
-GLOBUS_CLIENT_SECRET=<globus_client_secret>
-PREFECT_API_URL=<url_of_prefect_server>
-PREFECT_API_KEY=<prefect_client_secret>
+GLOBUS_CLIENT_ID=<globus_client_id>                 # For Globus Transfer
+GLOBUS_CLIENT_SECRET=<globus_client_secret>         # For Globus Transfer
+GLOBUS_COMPUTE_CLIENT_ID=<globus_client_id>         # For ALCF Jobs
+GLOBUS_COMPUTE_CLIENT_SECRET=<globus_client_secret> # For ALCF Jobs
+GLOBUS_COMPUTE_ENDPOINT=<globus_compute_endpoint>   # For ALCF Jobs
+PREFECT_API_URL=<url_of_prefect_server>             # For Prefect Flows
+PREFECT_API_KEY=<prefect_client_secret>             # For Prefect Flows
+SCICAT_API_URL=<url_of_scicat_api>                  # For SciCat Ingest
+SCICAT_INGEST_USER=<scicat_ingest_user>             # For SciCat Ingest
+SCICAT_INGEST_PASSWORD=<scicat_ingest_password>     # For SciCat Ingest
+PATH_NERSC_CLIENT_ID=<path_nersc_client_id>         # For NERSC SFAPI, generate on https://iris.nersc.gov/
+PATH_NERSC_PRI_KEY=<path_nersc_private_key>         # For NERSC SFAPI
 ```
 
 ## Current workflow overview and status:

diff --git a/config.yml b/config.yml
@@ -107,6 +107,11 @@ globus:
       client_id: ${GLOBUS_CLIENT_ID}
       client_secret: ${GLOBUS_CLIENT_SECRET}
 
+hpss_alsdev:
+  root_path: /home/a/alsdev/data_mover
+  uri: nersc.gov
+  name: hpss_alsdev
+
 harbor_images832:
   recon_image: tomorecon_nersc_mpi_hdf5@sha256:cc098a2cfb6b1632ea872a202c66cb7566908da066fd8f8c123b92fa95c2a43c
   multires_image: tomorecon_nersc_mpi_hdf5@sha256:cc098a2cfb6b1632ea872a202c66cb7566908da066fd8f8c123b92fa95c2a43c

diff --git a/create_deployment_832_dispatcher.sh b/create_deployment_832_dispatcher.sh
@@ -3,4 +3,14 @@ export $(grep -v '^#' .env | xargs)
 
 prefect work-pool create 'dispatcher_pool'
 prefect deployment build ./orchestration/flows/bl832/dispatcher.py:dispatcher -n run_832_dispatcher -q bl832 -p dispatcher_pool
-prefect deployment apply dispatcher-deployment.yaml
+prefect deployment apply dispatcher-deployment.yaml
+
+prefect work-pool create 'hpss_pool'
+prefect deployment build ./orchestration/flows/bl832/dispatcher.py:archive_832_project_dispatcher -n run_archive_832_project_dispatcher -q hpss_dispatcher_queue -p hpss_pool
+prefect deployment apply archive_832_project_dispatcher-deployment.yaml
+
+prefect deployment build ./orchestration/flows/bl832/dispatcher.py:archive_832_projects_from_previous_cycle_dispatcher -n run_archive_832_projects_from_previous_cycle_dispatcher -q hpss_dispatcher_queue -p hpss_pool
+prefect deployment apply archive_832_projects_from_previous_cycle_dispatcher-deployment.yaml
+
+prefect deployment build ./orchestration/flows/bl832/dispatcher.py:archive_all_832_raw_projects_dispatcher -n run_archive_all_832_raw_projects_dispatcher -q hpss_dispatcher_queue -p hpss_pool
+prefect deployment apply archive_all_832_raw_projects_dispatcher-deployment.yaml
diff --git a/create_deployment_832_hpss.sh b/create_deployment_832_hpss.sh
@@ -0,0 +1,7 @@
+export $(grep -v '^#' .env | xargs)
+
+
+prefect work-pool create 'hpss_pool'
+
+prefect deployment build ./orchestration/flows/bl832/hpss.py:cfs_to_hpss_flow -n cfs_to_hpss_flow -q cfs_to_hpss_queue -p hpss_pool
+prefect deployment apply cfs_to_hpss_flow-deployment.yaml
diff --git a/docs/bl832_ALCF.md b/docs/bl832_ALCF.md
@@ -428,7 +428,7 @@ def  prune_alcf832_raw(relative_path:  str):
 	prune_one_safe(
 		file=relative_path,
 		if_older_than_days=0,
-		tranfer_client=tc,
+		transfer_client=tc,
 		source_endpoint=config.alcf832_raw,
 		check_endpoint=config.nersc832_alsdev_raw,
 		logger=p_logger,

diff --git a/docs/mkdocs/docs/733.md b/docs/mkdocs/docs/733.md
@@ -0,0 +1,88 @@
+# Beamline 7.3.3
+
+
+## Flow Diagram
+```mermaid
+sequenceDiagram
+    participant DET as Detector/<br/>File Watcher
+    participant DISP as Prefect<br/>Dispatcher
+    participant D733 as data733<br/>Storage
+    participant GLOB as Globus<br/>Transfer
+    participant CFS as NERSC<br/>CFS
+    participant CAT as SciCat<br/>Metadata
+    participant SFAPI as SFAPI
+    participant HPC as HPC<br/>Compute
+    participant HPSS as HPSS<br/>Tape
+
+    %% Initial Trigger
+    DET->>DET: Monitor filesystem
+    DET->>DISP: Trigger on new file
+    DISP->>DISP: Coordinate flows
+
+    %% Flow 1: new_file_733
+    rect rgb(220, 230, 255)
+        note over DISP,CAT: FLOW 1: new_file_733
+        DISP->>GLOB: Init transfer
+        activate GLOB
+        GLOB->>D733: Initiate copy
+        activate D733
+        D733-->>GLOB: Copy initiated
+        deactivate D733
+        %% note right of GLOB: Transfer in progress
+        GLOB-->>DISP: Transfer complete
+        deactivate GLOB
+
+        DISP->>CAT: Register metadata
+    end
+
+    %% Flow 2: HPSS Transfer
+    rect rgb(220, 255, 230)
+        note over DISP,CAT: FLOW 2: Scheduled HPSS Transfer
+        DISP->>SFAPI: Submit tape job
+        activate SFAPI
+        SFAPI->>HPSS: Initiate archive
+        activate HPSS
+        HPSS-->>SFAPI: Archive complete
+        deactivate HPSS
+        SFAPI-->>DISP: Job complete
+        deactivate SFAPI
+
+        DISP->>CAT: Update metadata
+    end
+
+    %% Flow 3: HPC Analysis
+    rect rgb(255, 230, 230)
+        note over DISP,HPC: FLOW 3: HPC Downstream Analysis
+        DISP->>SFAPI: Submit compute job
+        activate SFAPI
+        SFAPI->>HPC: Execute job
+        activate HPC
+        HPC->>HPC: Process data
+        HPC-->>SFAPI: Compute complete
+        deactivate HPC
+        SFAPI-->>DISP: Job complete
+        deactivate SFAPI
+
+        DISP->>CAT: Update metadata
+    end
+
+    %% Flow 4: Scheduled Pruning
+    rect rgb(255, 255, 220)
+        note over DISP,CAT: FLOW 4: Scheduled Pruning
+        DISP->>DISP: Scheduled pruning trigger
+
+        DISP->>D733: Prune old files
+        activate D733
+        D733->>D733: Delete expired data
+        D733-->>DISP: Pruning complete
+        deactivate D733
+
+        DISP->>CFS: Prune old files
+        activate CFS
+        CFS->>CFS: Delete expired data
+        CFS-->>DISP: Pruning complete
+        deactivate CFS
+
+        DISP->>CAT: Update metadata
+    end
+  ```
diff --git a/docs/mkdocs/docs/alcf832.md b/docs/mkdocs/docs/alcf832.md
@@ -389,7 +389,7 @@ def  prune_alcf832_raw(relative_path:  str):
     prune_one_safe(
         file=relative_path,
         if_older_than_days=0,
-        tranfer_client=tc,
+        transfer_client=tc,
         source_endpoint=config.alcf832_raw,
         check_endpoint=config.nersc832_alsdev_raw,
         logger=p_logger,

diff --git a/docs/mkdocs/docs/assets/images/sfapi_step1.png b/docs/mkdocs/docs/assets/images/sfapi_step1.png
diff --git a/docs/mkdocs/docs/assets/images/sfapi_step2.png b/docs/mkdocs/docs/assets/images/sfapi_step2.png
diff --git a/docs/mkdocs/docs/assets/images/sfapi_step3.png b/docs/mkdocs/docs/assets/images/sfapi_step3.png
diff --git a/docs/mkdocs/docs/assets/images/sfapi_step4.png b/docs/mkdocs/docs/assets/images/sfapi_step4.png
diff --git a/docs/mkdocs/docs/common_infrastructure.md b/docs/mkdocs/docs/common_infrastructure.md
@@ -0,0 +1,47 @@
+# Common Infrastructure
+
+## Overview
+The common infrastructure for this project includes:
+- **Shared Code**: There are general functions and classes used across beamline workflows to reduce code duplication.
+- **Beamline Specific Implementation Patterns**: We organize each beamline's implementation in a similar way, making it easier to understand and maintain.
+
+## Shared Code
+Shared code is organized into modules that can be imported in beamline specific implementations. Key modules include:
+- **`orchestration/config.py`**
+    - Contains an Abstract Base Class (ABC) called `BeamlineConfig()` which serves as the base for all beamline-specific configuration classes. It uses the `Dynaconf` package to load the configuration file,`config.yml`, which contains information about endpoints, containers, and more.
+- **`orchestration/transfer_endpoints.py`**
+    - Contains an ABC called `TransferEndpoint()`, which is extended by `FileSystemEndpoint`, `HPSSEndpoint` and `GlobusEndpoint`. These definitions are used to enforce typing and ensure the correct transfer and pruning implmentation are used.
+- **`orchestration/transfer_controller.py`**:
+    - Contains an ABC called `TransferController()` with specific implementations for Globus, Local File Systems, and NERSC HPSS.
+- **`orchestration/prune_controller.py`**
+    - This module is responsible for managing the pruning of data off of storage systems. It uses a configurable retention policy to determine when to remove files. It contains an ABC called `PruneController()` that is extended by specific implementations for `FileSystemEndpoint`, `GlobusEndpoint`, and `HPSSEndpoint`.
+- **`orchestration/sfapi.py`**: Create an SFAPI Client to launch remote jobs at NERSC.
+- **`orchestration/flows/scicat/ingest.py`**: Ingests datasets into SciCat, our metadata management system.
+- **`orchestration/hpss.py`**: Schedule a Prefect Flow to copy data between NERSC CFS and HPSS. These call the relevant TransferControllers for HPSS, which handle the underlying tape-safe logic.
+
+
+## Beamline Specific Implementation Patterns
+In order to balance generalizability, maintainability, and scalability of this project to multiple beamlines, we try to organize specific implementations in a similar way. We keep specific implementaqtions in the directory `orchestration/flows/bl{beamline_id}/`, which generally contains a few things:
+- **`config.py`**
+    - Extend `BeamlineConfig()` from `orchestration/config.py` for specific implementations (e.g. `Config832`, `Config733`, etc.) This ensures only the relevant beamline specific configurations are used in each case.
+- **`dispatcher.py`**
+    - This script is the starting point for each beamline's data transfer and analysis workflow. The Prefect Flow it contains is generally invoked by a File Watcher script on the beamline computer. The Dispatcher contains the logic for calling subflows, ensures that steps are completed in the correct order, and prevents subsequent steps from being called if there is a failure along the way.
+- **`move.py`**
+    - This script is usually the first one the Dispatcher calls synchronously, and contains the logic for immediately moving data, scheduling pruning flows, and ingesting into SciCat. Downstream steps typically rely on this action completing first.
+- **`job_controller.py`**
+    - For beamlines that trigger remote analysis workflows, the `JobController()` ABC allows us to define HPC or machine specific implementations, which may differ in how code can be deployed. For example, it can be extended to define how to run tomography reconstruction at ALCF and NERSC.
+- **`{hpc}.py`**
+    - We separate HPC implementations for `JobController()` in their own files.
+- **`ingest.py`**
+    - This is where we define SciCat implementations for each beamline, as each technique will have specific metadata fields that are important to capture.
+
+## Testing
+We write Unit Tests using [pytest](https://pytest.org/) for individual components, which can be found in `orchestration/_tests/`. We run these tests as part of our Github Actions.
+
+## CI/CD
+The project is integrated with [GitHub Actions](https://github.com/features/actions) for continuous integration and deployment. The specifics for these can be found in `.github/workflows/`. The features we support here includes:
+
+- **Automated Test Execution**: All the unit tests are run automatically with every Git Push.
+- **Linting**: `flake8` is used to check for syntax and styling errors.
+- **MkDocs**: The documentation site is automatically updated whenever a Pull Request is merged into the main branch.
+- **Docker**: A Docker image is aumatically created and registered on the Github Container Repository (ghcr.io) when a new release is made.