xGFabric twin as a service (DTaaS / ORBIT)

service/README.md

@@ -0,0 +1,106 @@
+# xGFabric twin, service mode
+
+`../twin_service.py` runs twin.py's graph on a DTaaS broker: the twin
+lives in the ORBIT `dt` plugin, its tasks run on a rhapsody endpoint,
+the sensor is an external channel publisher.  Physics is faked at the
+same seams twin.py already fakes (sensor records, simulation); the
+selection story is real — three surrogate architectures with different
+costs, ranked by profiler-predicted Pi runtime.
+
+## Local run (three terminals + stack)
+
+Stack (from a digital.twins checkout with `./deploy/install.sh` done,
+venv `ve3`/`ve.demo`):
+
+    ./deploy/run-broker.sh $PWD/<venv>
+    ./deploy/run-endpoint.sh dt_inference_ep localhost $PWD/<venv>
+
+Client terminals (each):
+
+    export RADICAL_ORBIT_BROKER_URL=wss://localhost:8000
+    export DT_STREAM_BACKEND=orbit
+
+    <venv>/bin/python service/sensor_publisher.py     # terminal 1
+    <venv>/bin/python twin_service.py --runtime 240   # terminal 2
+
+Dashboard: `https://localhost:8000/broker/dt/ui?live=1` (broker token).
+Heatmaps land in `$XGF_WORKSPACE` (default `~/xgf_twin/`) on the host
+running the endpoint tasks.
+
+Placement: `DT_INFERENCE_ENDPOINT` / `DT_INFERENCE_BACKEND` override
+the defaults (`dt_inference_ep` / `concurrent`).
+
+## Remote run (broker on radical.3, endpoint on Perlmutter)
+
+`service/deploy/` adapts the AmSC dt-complete deploy kit (same debugged
+constraints: dragon launcher requirement, python >= 3.12.1,
+SLURM_EXPORT_ENV, cert staging), pinning rhapsody's
+`fix/dragon-cancel-idempotent` branch (e491cd2-based) on every tier: it
+carries the dragon cancel + traceback fixes and stays compatible with
+the pinned dragonhpc 0.14.1 (rhapsody main's dragon backend passes
+`task_logs=` to `Batch()`, which that dragon rejects).
+
+    # broker host, once
+    service/deploy/setup-broker.sh
+    cd ~/digital_twins && ./deploy/run-broker.sh $PWD/ve.demo
+
+    # Perlmutter login node, once
+    service/deploy/setup-hpc-endpoint.sh <broker-host>
+    # then inside salloc -N1 -C cpu -q interactive -t 2:00:00 -A <account>:
+    service/deploy/run-hpc-endpoint.sh <broker-host>     # registers as 'hpc'
+
+    # client terminals (driver + sensor), each:
+    source service/deploy/client-env.sh <broker-host> remote
+    <ve.demo>/bin/python service/sensor_publisher.py     # terminal 1
+    <ve.demo>/bin/python twin_service.py --runtime 240   # terminal 2
+
+The client venv must be the same Python minor (digital.twins
+`./deploy/install.sh client` + `pip install numpy`); a 3.13 venv is
+rejected at the first verb.
+
+## Real workload (Level A: real training, faked simulation)
+
+The fake `service/*` components run the DTaaS mechanics end to end.  The
+real path swaps in the actual FNO/PINN/PCR investigators (TensorFlow /
+scikit-learn) and the real profiler, still on precalc simulation data
+(`/global/cfs/cdirs/m5290/precalc_sims`) -- real OpenFOAM (Level B,
+`cups_structure.zip`) is a later step.
+
+Only the **endpoint** changes: the trainings need TensorFlow, so it runs
+in a clone of Ben's `cfdaai` conda env (which carries the stack) with our
+runtime installed into the clone -- built by `setup-hpc-endpoint-real.sh`,
+then launched via `run-hpc-endpoint.sh` with `DT_VENV`/`XGF_DIR` set (the
+setup script prints the exact line).  Broker and client stay on `ve.demo`
+and need no TensorFlow.
+
+    # Perlmutter login node, once (clones cfdaai, installs our runtime,
+    # checks out the tasks tree, stages the profiler dataset)
+    service/deploy/setup-hpc-endpoint-real.sh <broker-host>
+    # then in an allocation, per the line it prints:
+    DT_VENV=<clone> XGF_DIR=<xgf checkout> \
+      service/deploy/run-hpc-endpoint.sh <broker-host>
+
+Two things are NOT done yet and gate an actual real run:
+
+- **Lazy TensorFlow imports.**  The real investigator wrappers
+  `import tensorflow` at module load, which would drag TF onto the
+  client and broker just to package/instantiate the class.  The wrappers
+  need their heavy imports deferred into the task bodies (which run on
+  the endpoint, where TF lives).  Until that lands, only the endpoint
+  tier is provisioned; the driver still uses the fake components.
+- **cloudpickle parity.**  The cloned conda env and `ve.demo` may carry
+  different cloudpickle versions; align them (the endpoint install pulls
+  our pinned stack, but verify against the client) or unpickling the
+  shipped classes can fail.
+
+## What maps to what
+
+| twin.py (standalone)             | twin_service.py (DTaaS)              |
+|----------------------------------|--------------------------------------|
+| local WorkflowEngine + backend   | session engine on the broker, tasks on the endpoint (ENGINES config) |
+| DavisWind persistent component   | external `ChannelPublisher` + `add_input` binding |
+| WindFieldAgent + FNO/PINN/PCR    | `ServiceWindFieldAgent` + fake `SurrogateInvestigator`s (same selection logic) |
+| profiler subprocess + Pi learner | `ServiceProfiler` (inline timed run) + `ServicePiPredictor` |
+| CUPS_Sink                        | `ServiceSink` (runtime-resolved workspace) |
+| ZMQ stream                       | ORBIT data plane                     |
+| asyncflow telemetry + reports    | not yet — see the telemetry branch   |

service/__init__.py

@@ -0,0 +1,9 @@
+# Service-mode (DTaaS / ORBIT) variant of the xGFabric twin.
+#
+# Same graph and the same model-selection story as ../twin.py -- a wind
+# sensor feeds a field agent whose three competing surrogates are ranked
+# by profiler-predicted Pi runtime -- but the components here are
+# service-safe: they ship to the broker by value, run their tasks on a
+# rhapsody endpoint, and fake the physics (as twin.py already does for
+# the sensor and the simulation).  The real FNO/PINN/PCR training stacks
+# stay in ../tasks and are NOT imported here.

service/deploy/client-env.sh

@@ -0,0 +1,23 @@
+# Client-side environment for the xGFabric service twin -- source me in
+# EVERY client terminal (driver and sensor):
+#
+#   source service/deploy/client-env.sh <broker-host> [remote]
+#
+# With `remote`, task placement targets the HPC endpoint ('hpc' on
+# dragon_v3, as run-hpc-endpoint.sh launches it); without it the local
+# defaults apply (dt_inference_ep / concurrent).
+#
+# The client venv is digital.twins' `./deploy/install.sh client` plus
+# `pip install numpy` -- same Python minor as broker and endpoint.
+# DT_BROKER_CERT overrides the pinned-cert path when this machine runs
+# a broker of its own.
+BROKER="${1:?usage: source client-env.sh <broker-host> [remote]}"
+
+export RADICAL_ORBIT_BROKER_URL="wss://$BROKER:8000"
+export RADICAL_ORBIT_BROKER_CERT="${DT_BROKER_CERT:-$HOME/.radical/orbit/broker_cert.pem}"
+export DT_STREAM_BACKEND=orbit
+
+if [ "${2:-}" = "remote" ]; then
+    export DT_INFERENCE_ENDPOINT=hpc
+    export DT_INFERENCE_BACKEND=dragon_v3
+fi

service/deploy/run-hpc-endpoint.sh

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# Run the xGFabric twin's HPC endpoint -- INSIDE a compute allocation.
+# Adapted from the AmSC dt-complete deploy kit; the launch constraints
+# (dragon launcher, PATH-by-name helpers, SLURM_EXPORT_ENV) are the
+# debugged ones from there.
+#
+#   ./run-hpc-endpoint.sh <broker-host>
+#
+# The endpoint registers as 'hpc' -- what the client's remote profile
+# (client-env.sh remote) asks for.  DT_DIR as in setup-hpc-endpoint.sh.
+#
+# Two modes, same script:
+#   faked demo   -- DT_VENV unset: uses $DT_DIR/ve.demo (setup-hpc-endpoint.sh)
+#   real workload-- DT_VENV=<conda env>, XGF_DIR=<xGFabric checkout>: the
+#                   real trainings need TensorFlow (Ben's cloned cfdaai env)
+#                   and the profiler/task bodies import from the tasks tree
+#                   (setup-hpc-endpoint-real.sh prints the exact invocation).
+set -euo pipefail
+BROKER="${1:?usage: $0 <broker-host>}"
+DT_DIR="${DT_DIR:-${SCRATCH:-$HOME}/digital_twins}"
+VENV="${DT_VENV:-$DT_DIR/ve.demo}"
+
+# dragon resolves its helpers BY NAME through srun on the task side
+export PATH="$VENV/bin:$PATH"
+export RADICAL_ORBIT_BROKER_URL="wss://$BROKER:8000"
+export RADICAL_ORBIT_BROKER_CERT="$HOME/.radical/orbit/broker_cert.pem"
+export RADICAL_ORBIT_RHAPSODY_BACKEND=dragon_v3
+export RADICAL_ORBIT_RHAPSODY_NOTIFY_WINDOW=0
+export SLURM_EXPORT_ENV=ALL          # inner sruns must not scrub the env
+export DT_STREAM_BACKEND=orbit
+# heatmaps land on scratch where available
+export XGF_WORKSPACE="${XGF_WORKSPACE:-${SCRATCH:-$HOME}/xgf_twin}"
+
+# real workload: the profiler shells out to a script in the xGFabric tasks
+# tree and the (lazy-imported) task bodies import tasks.* -- put the
+# checkout on PYTHONPATH when XGF_DIR names one
+if [ -n "${XGF_DIR:-}" ]; then
+    export PYTHONPATH="$XGF_DIR:${PYTHONPATH:-}"
+fi
+
+"$VENV/bin/dragon" "$VENV/bin/radical-orbit-endpoint.py" -n hpc \
+    2>&1 | tee endpoint.log

service/deploy/setup-broker.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+# DTaaS broker host for the xGFabric service twin (e.g. radical.3).
+# Run there.  Adapted from the AmSC dt-complete deploy kit, which
+# carries the debugged install/launch lore; delta here: numpy for the
+# by-value components.
+#
+# Once per host: broker_cert.pem / broker_key.pem / broker.token in
+# ~/.radical/orbit/.  DT_DIR overrides the checkout+venv location.
+set -euo pipefail
+DT_DIR="${DT_DIR:-$HOME/digital_twins}"
+
+[ -d "$DT_DIR" ] || git clone https://github.com/radical-cybertools/digital.twins.git "$DT_DIR"
+cd "$DT_DIR" && git checkout devel && git pull
+
+./deploy/install.sh broker                        # pinned stack -> ./ve.demo
+./ve.demo/bin/pip install -q numpy
+# match the endpoint's rhapsody exactly (e491cd2-based): main breaks the
+# endpoint's dragon backend (task_logs= to Batch), so both tiers pin the
+# proven branch.  The dragon cancel fix on it is a no-op broker-side; the
+# orbit-backend cancel fix lives only on the main-based #91 branch, so a
+# harmless KeyError-cancel line may appear on teardown here.
+./ve.demo/bin/pip install -q --force-reinstall --no-deps \
+  "rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@fix/dragon-cancel-idempotent"
+
+echo "done.  start the broker with:"
+echo "  cd $DT_DIR && ./deploy/run-broker.sh \$PWD/ve.demo"

service/deploy/setup-hpc-endpoint-real.sh

@@ -0,0 +1,78 @@
+#!/usr/bin/env bash
+# Real-workload HPC endpoint for the xGFabric service twin (Perlmutter).
+# Run on a login node.
+#
+#   ./setup-hpc-endpoint-real.sh <broker-host>
+#
+# Unlike setup-hpc-endpoint.sh (faked components, plain ve.demo), the real
+# FNO/PINN/PCR trainings need TensorFlow + the CFD/ML stack.  Rather than
+# build that, we CLONE Ben's cfdaai conda env (which has it) and install
+# our runtime into the clone -- never into his shared env.
+#
+# Prerequisites (read access): /global/common/software/m5290 (Ben's envs
+# + cups_structure.zip) and /global/cfs/cdirs/m5290/precalc_sims.
+# `conda` must be on PATH (module load, or source Ben's mconda).
+#
+# Overrides: DT_DIR (our checkout), XGF_DIR (xGFabric checkout),
+# DT_ENV (clone location), DT_BASE_ENV (base env name, default cfdaai).
+set -euo pipefail
+BROKER="${1:?usage: $0 <broker-host>}"
+DT_DIR="${DT_DIR:-${SCRATCH:-$HOME}/digital_twins}"
+XGF_DIR="${XGF_DIR:-${SCRATCH:-$HOME}/xGFabric}"
+ENV_PREFIX="${DT_ENV:-${SCRATCH:-$HOME}/dt-endpoint-env}"
+BEN_ENVS="/global/common/software/m5290/bcarter/mconda/envs"
+BASE_ENV="${DT_BASE_ENV:-cfdaai}"
+
+command -v conda >/dev/null || {
+    echo "ERROR: conda not on PATH -- module load python, or source Ben's" >&2
+    echo "       mconda, then re-run." >&2; exit 1; }
+
+# clone Ben's env (never install into his shared copy)
+conda config --add envs_dirs "$BEN_ENVS" 2>/dev/null || true
+if [ ! -d "$ENV_PREFIX" ]; then
+    echo "==> cloning $BASE_ENV -> $ENV_PREFIX (this is large; once)"
+    conda create -y -p "$ENV_PREFIX" --clone "$BEN_ENVS/$BASE_ENV"
+fi
+PY="$ENV_PREFIX/bin/python"
+
+# wire contract: the same Python minor on every tier
+ver="$("$PY" -c 'import sys; print("%d.%d" % sys.version_info[:2])')"
+if [ "$ver" != "3.12" ]; then
+    echo "ERROR: $BASE_ENV is Python $ver; the DT wire contract pins 3.12" >&2
+    echo "       on every host.  Set DT_BASE_ENV to a 3.12 env (try" >&2
+    echo "       xgfabric), or rebuild from its environment.yml on 3.12." >&2
+    exit 1
+fi
+
+# our runtime into the clone -- git-pinned deps first (as deploy/install.sh
+# does), then digitaltwin resolves the rest (asyncflow, orbit) from PyPI
+[ -d "$DT_DIR" ] || git clone https://github.com/radical-cybertools/digital.twins.git "$DT_DIR"
+( cd "$DT_DIR" && git checkout devel && git pull )
+"$PY" -m pip install -q \
+    "rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17"
+"$PY" -m pip install -q --force-reinstall --no-deps \
+    "rhapsody-py[telemetry,dragon] @ git+https://github.com/radical-cybertools/rhapsody@fix/dragon-cancel-idempotent"
+"$PY" -m pip install -q "$DT_DIR"
+
+# xGFabric tasks tree: the profiler shells out to a script here, and the
+# real task bodies import tasks.* at run time (run-hpc-endpoint.sh puts
+# XGF_DIR on PYTHONPATH)
+[ -d "$XGF_DIR" ] || git clone https://github.com/radical-collaboration/xGFabric.git "$XGF_DIR"
+( cd "$XGF_DIR" && git checkout feature/dtaas-twin && git pull )
+# the Pi predictor's clean-slate dataset (Ben: the sample IS the real one)
+cp -n "$XGF_DIR/tasks/profiler/pi_profiler/data.csv.sample" \
+      "$XGF_DIR/tasks/profiler/pi_profiler/data.csv" 2>/dev/null || true
+
+# broker cert + token
+mkdir -p ~/.radical/orbit
+scp "$BROKER:.radical/orbit/broker_cert.pem" "$BROKER:.radical/orbit/broker.token" ~/.radical/orbit/
+
+cat <<EOF
+done.  in an allocation, run the endpoint with the clone + tasks tree:
+
+  DT_VENV=$ENV_PREFIX XGF_DIR=$XGF_DIR \\
+    $XGF_DIR/service/deploy/run-hpc-endpoint.sh $BROKER
+
+precalc sims: /global/cfs/cdirs/m5290/precalc_sims  (config.sh path)
+cups zip (Level B only): /global/common/software/m5290/cups_structure.zip
+EOF

service/deploy/setup-hpc-endpoint.sh

@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+# HPC-endpoint venv for the xGFabric service twin (e.g. Perlmutter).
+# Run on a login node.  Adapted from the AmSC dt-complete deploy kit.
+#
+#   ./setup-hpc-endpoint.sh <broker-host>
+#
+# DT_DIR overrides the checkout+venv location; the default lands on
+# $SCRATCH -- the venv is far too big for a Perlmutter home quota.
+set -euo pipefail
+BROKER="${1:?usage: $0 <broker-host>}"
+DT_DIR="${DT_DIR:-${SCRATCH:-$HOME}/digital_twins}"
+
+# same Python minor as every other host -- the service rejects skew, and
+# exactly 3.12.0 breaks dragon's transport import (needs >= 3.12.1)
+module load python/3.12 2>/dev/null || true
+
+[ -d "$DT_DIR" ] || git clone https://github.com/radical-cybertools/digital.twins.git "$DT_DIR"
+cd "$DT_DIR" && git checkout devel && git pull
+
+./deploy/install.sh endpoint                      # pinned stack -> ./ve.demo
+# the surrogate/sink task bodies unpickle and run here
+./ve.demo/bin/pip install -q numpy matplotlib
+# dragon backend + idempotent cancel + failure-traceback logging.
+# e491cd2-based (NOT main): main's dragon backend passes task_logs= to
+# Batch(), which the pinned dragonhpc 0.14.1 does not accept.  This is
+# the branch the AmSC demo proved on the same dragon.
+./ve.demo/bin/pip install -q --force-reinstall --no-deps \
+  "rhapsody-py[telemetry,dragon] @ git+https://github.com/radical-cybertools/rhapsody@fix/dragon-cancel-idempotent"
+
+mkdir -p ~/.radical/orbit
+scp "$BROKER:.radical/orbit/broker_cert.pem" "$BROKER:.radical/orbit/broker.token" ~/.radical/orbit/
+
+echo "done.  get an allocation (salloc -N1 -C cpu -q interactive -t 2:00:00 -A <account>),"
+echo "then run:  service/deploy/run-hpc-endpoint.sh $BROKER"