xGFabric twin as a service: real workload + demo kit (supersedes #13, #14)

demo/Sep_04_broker.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+# xGFabric demo -- DTaaS broker.  Run on the broker host (radical.3).
+# Sets up (install + fixes) then starts the broker.  Idempotent.
+#
+#   demo/Sep_04_broker.sh [broker-ip]
+#
+# Assumes broker_cert.pem / broker_key.pem / broker.token in ~/.radical/orbit/.
+set -euo pipefail
+
+BROKER_IP="${1:-95.217.193.116}"
+
+# --- demo/site specific ----------------------------------------------------
+DT_DIR="${DT_DIR:-$HOME/digital_twins}"
+RH="rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@fix/dragon-cancel-idempotent"
+# ---------------------------------------------------------------------------
+
+echo "--------------------------"
+echo "xGFabric Demo September 04"
+echo "Broker on $BROKER_IP ($(hostname -f))"
+echo "--------------------------"
+
+# checkout + install the pinned stack, then pin the dragon-compatible
+# rhapsody branch (extras backfilled)
+[ -d "$DT_DIR" ] || git clone https://github.com/radical-cybertools/digital.twins.git "$DT_DIR"
+( cd "$DT_DIR" && git checkout devel && git pull )
+( cd "$DT_DIR"
+  ./deploy/install.sh broker
+  # install.sh compares by version string and skips when it is unchanged,
+  # so it can leave stale digitaltwin code (e.g. missing record_output).
+  # Force the checked-out devel tip over whatever it left.
+  ./ve.demo/bin/pip install -q --force-reinstall --no-deps "$DT_DIR"
+  ./ve.demo/bin/pip install -q numpy pandas    # the agent runs broker-side
+  ./ve.demo/bin/pip install -q --force-reinstall --no-deps \
+      "rhapsody-py @ git+https://github.com/radical-cybertools/rhapsody@fix/dragon-cancel-idempotent"
+  ./ve.demo/bin/pip install -q "$RH" )
+
+echo "starting broker (wss://0.0.0.0:8000) ..."
+exec "$DT_DIR/deploy/run-broker.sh" "$DT_DIR/ve.demo"

demo/Sep_04_client.sh

@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+# xGFabric demo -- client driver.  Run on the client.  Ensures the light
+# client-side deps, then drives the real twin against the HPC endpoint.
+#
+#   demo/Sep_04_client.sh [broker-ip]
+set -euo pipefail
+
+BROKER_IP="${1:-95.217.193.116}"
+
+# --- demo/site specific ----------------------------------------------------
+VENV="${DT_VENV:-$HOME/radical/digital_twins/ve.demo}"
+DT_DIR="${DT_DIR:-$HOME/radical/digital_twins}"
+HERE="$(cd "$(dirname "$0")/.." && pwd)"     # xGFabric checkout root
+RUNTIME="${RUNTIME:-600}"
+# default to the fast fake-surrogate driver (CPU-friendly: sleeps + numpy,
+# no TF/xgboost, heatmaps in seconds, learning lane built in).  RUN_REAL=1
+# switches to the real TF workload (needs a GPU endpoint to be timely).
+DRIVER="twin_service.py"
+[ "${RUN_REAL:-}" = 1 ] && DRIVER="twin_service_real.py"
+# ---------------------------------------------------------------------------
+
+echo "--------------------------"
+echo "xGFabric Demo September 04"
+echo "Client $DRIVER ($(hostname -f)) -> broker $BROKER_IP, endpoint HPC 'hpc'"
+echo "--------------------------"
+
+cd "$HERE"
+
+# client-side deps the real components import at packaging time (TF and
+# matplotlib stay lazy / endpoint-only); pyspot submodule for the sensor.
+"$VENV/bin/pip" install -q python-dotenv numpy pandas
+git submodule update --init --recursive
+# keep the client digitaltwin current (record_output etc.)
+( cd "$DT_DIR" && git checkout devel && git pull ) && "$VENV/bin/pip" install -q "$DT_DIR"
+
+# placement: run the twin's tasks on the HPC endpoint (dragon)
+export RADICAL_ORBIT_BROKER_URL="wss://$BROKER_IP:8000"
+export RADICAL_ORBIT_BROKER_CERT="$HOME/.radical/orbit/broker_cert.pem"
+export DT_STREAM_BACKEND=orbit
+export DT_INFERENCE_ENDPOINT=hpc
+export DT_INFERENCE_BACKEND=dragon_v3
+export DT_LEARNING_ENDPOINT=hpc
+export DT_LEARNING_BACKEND=concurrent
+
+echo "driving twin via $DRIVER (runtime ${RUNTIME}s) ..."
+echo "dashboard: https://$BROKER_IP:8000/broker/dt/ui?live=1"
+exec "$VENV/bin/python" "$DRIVER" --runtime "$RUNTIME"

demo/Sep_04_endpoint.sh

@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+# xGFabric demo -- Perlmutter endpoint RUN.  Run INSIDE the compute
+# allocation (no pip here; Sep_04_endpoint_setup.sh did that on the login
+# node).  Launches the rhapsody endpoint under the dragon launcher.
+#
+#   demo/Sep_04_endpoint.sh [broker-ip] [endpoint-name]
+set -euo pipefail
+
+BROKER_IP="${1:-95.217.193.116}"
+EP="${2:-hpc}"
+
+# --- demo/site specific ----------------------------------------------------
+ENV_PREFIX="${DT_ENV:-$SCRATCH/dt-endpoint-env}"
+XGF_DIR="${XGF_DIR:-$SCRATCH/xgfabric}"
+PLAYGROUND_DIR="${PLAYGROUND_DIR:-$SCRATCH/xgf_playground}"
+# ---------------------------------------------------------------------------
+
+echo "--------------------------"
+echo "xGFabric Demo September 04"
+echo "Endpoint HPC '$EP' on $(hostname -f) -> broker $BROKER_IP"
+echo "--------------------------"
+
+# dragon resolves its helpers BY NAME via srun on the task side
+export PATH="$ENV_PREFIX/bin:$PATH"
+export PYTHONPATH="$XGF_DIR:${PYTHONPATH:-}"
+export RADICAL_ORBIT_BROKER_URL="wss://$BROKER_IP: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
+export XGF_WORKSPACE="$PLAYGROUND_DIR"
+# keep the orbit log off HOME (tiny NERSC quota) and modest
+export RADICAL_ORBIT_LOG_LVL="${RADICAL_ORBIT_LOG_LVL:-WARNING}"
+export RADICAL_ORBIT_LOG_FILE="$SCRATCH/orbit-logs/$EP.log"
+mkdir -p "$SCRATCH/orbit-logs" "$PLAYGROUND_DIR"
+rm -f "$HOME/.radical/orbit/logs/"*.log 2>/dev/null || true
+
+# --- GPU (opt-in: RUN_GPU=1) -----------------------------------------------
+# Only for a real GPU allocation.  Prepending /usr/lib64 to LD_LIBRARY_PATH
+# can shadow the conda env's own libs and break the dragon task launch, so
+# it stays OFF by default -- the CPU/fake demo path must not touch the env.
+if [ "${RUN_GPU:-}" = 1 ]; then
+  module load cudatoolkit 2>/dev/null || true
+  export LD_LIBRARY_PATH="/usr/lib64:$ENV_PREFIX/lib:${LD_LIBRARY_PATH:-}"
+  echo "GPU check (nvidia-smi -L):"
+  nvidia-smi -L 2>&1 | sed 's/^/  /' || echo "  no GPU visible (need salloc -C gpu)"
+fi
+
+echo "launching endpoint '$EP' under dragon ..."
+exec "$ENV_PREFIX/bin/dragon" "$ENV_PREFIX/bin/radical-orbit-endpoint.py" -n "$EP" \
+    2>&1 | tee "$SCRATCH/orbit-logs/$EP.console.log"

demo/Sep_04_endpoint_setup.sh

@@ -0,0 +1,79 @@
+#!/usr/bin/env bash
+# xGFabric demo -- Perlmutter endpoint SETUP.  Run on a LOGIN NODE (it
+# pip-installs and clones; the allocation cannot).  Idempotent.
+#
+#   demo/Sep_04_endpoint_setup.sh [broker-ip]   (broker-ip only for the banner)
+#
+# Assumes keys/token already staged in ~/.radical/orbit/.
+set -euo pipefail
+
+BROKER_IP="${1:-95.217.193.116}"
+EP=hpc
+
+# --- demo/site specific ----------------------------------------------------
+BRANCH=feature/dtaas-twin-real
+DT_DIR="${DT_DIR:-$SCRATCH/digital_twins}"
+XGF_DIR="${XGF_DIR:-$SCRATCH/xgfabric}"
+ENV_PREFIX="${DT_ENV:-$SCRATCH/dt-endpoint-env}"
+PLAYGROUND_DIR="${PLAYGROUND_DIR:-$SCRATCH/xgf_playground}"
+BEN_ENVS="/global/common/software/m5290/bcarter/mconda/envs"
+BASE_ENV=cfdaai
+RH="rhapsody-py[telemetry,dragon] @ git+https://github.com/radical-cybertools/rhapsody@fix/dragon-cancel-idempotent"
+ROSE="rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17"
+# ---------------------------------------------------------------------------
+
+echo "--------------------------"
+echo "xGFabric Demo September 04 -- ENDPOINT SETUP (login node)"
+echo "Broker $BROKER_IP | Endpoint '$EP' | env $ENV_PREFIX | playground $PLAYGROUND_DIR"
+echo "--------------------------"
+
+module load python/3.12 2>/dev/null || true
+command -v conda >/dev/null || { echo "ERROR: conda not on PATH" >&2; exit 1; }
+
+# checkouts
+[ -d "$DT_DIR" ]  || git clone https://github.com/radical-cybertools/digital.twins.git "$DT_DIR"
+( cd "$DT_DIR"  && git checkout devel  && git pull )
+[ -d "$XGF_DIR" ] || git clone https://github.com/radical-collaboration/xGFabric.git "$XGF_DIR"
+( cd "$XGF_DIR" && git checkout "$BRANCH" && git pull && git submodule update --init --recursive )
+
+# conda env: a clone of Ben's cfdaai (TensorFlow + CFD/ML stack), never his
+conda config --add envs_dirs "$BEN_ENVS" 2>/dev/null || true
+[ -d "$ENV_PREFIX" ] || conda create -y -p "$ENV_PREFIX" --clone "$BEN_ENVS/$BASE_ENV"
+PY="$ENV_PREFIX/bin/python"
+
+ver="$("$PY" -c 'import sys;print("%d.%d"%sys.version_info[:2])')"
+[ "$ver" = "3.12" ] || { echo "ERROR: $BASE_ENV is Python $ver, need 3.12" >&2; exit 1; }
+
+# our runtime into the clone.  orbit force-reinstalled (a clone may carry a
+# stale one that hides the endpoint script), then deps backfilled.
+echo "==> installing runtime into $ENV_PREFIX"
+"$PY" -m pip install -q "$ROSE"
+"$PY" -m pip install -q --force-reinstall --no-deps "radical.orbit>=0.7"
+"$PY" -m pip install -q "radical.orbit>=0.7"
+"$PY" -m pip install -q "$DT_DIR"
+# rhapsody LAST so the dragon-compatible branch wins over what digitaltwin
+# pulled (main passes task_logs= to Batch(), which the pinned dragonhpc
+# rejects); extras step backfills opentelemetry + dragonhpc.
+"$PY" -m pip install -q --force-reinstall --no-deps \
+    "rhapsody-py @ git+https://github.com/radical-cybertools/rhapsody@fix/dragon-cancel-idempotent"
+"$PY" -m pip install -q "$RH"
+# endpoint_trainer.py / endpoint_eval.py need these; cfdaai lacks xgboost
+"$PY" -m pip install -q xgboost scikit-learn
+
+# data dirs: seed the Pi-predictor dataset at the RUNTIME datastore (where
+# endpoint_trainer.py reads it)
+mkdir -p "$PLAYGROUND_DIR/profiler/pi_profiler"
+cp -n "$XGF_DIR/tasks/profiler/pi_profiler/data.csv.sample" \
+      "$PLAYGROUND_DIR/profiler/pi_profiler/data.csv" 2>/dev/null || true
+
+# sanity
+echo "==> verify"
+"$PY" -c "import radical.orbit, digitaltwin, rhapsody, opentelemetry; print('  imports ok')"
+"$PY" -c "import rhapsody.backends.execution.dragon as d; n=open(d.__file__).read().count('task_logs'); print('  dragon task_logs (want 0):', n)"
+[ -x "$ENV_PREFIX/bin/radical-orbit-endpoint.py" ] && echo "  endpoint script ok" || echo "  WARN: endpoint script missing"
+[ -x "$ENV_PREFIX/bin/dragon" ] && echo "  dragon ok" || echo "  WARN: dragon missing"
+
+echo "--------------------------"
+echo "done.  get an allocation, then run:"
+echo "  $XGF_DIR/demo/Sep_04_endpoint.sh $BROKER_IP $EP"
+echo "--------------------------"

demo/Sep_04_sensor.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+# xGFabric demo -- external wind sensor (fake Davis).  Run on the client.
+# Publishes to the twin's input channel over the ORBIT data plane.
+#
+#   demo/Sep_04_sensor.sh [broker-ip]
+set -euo pipefail
+
+BROKER_IP="${1:-95.217.193.116}"
+
+# --- demo/site specific ----------------------------------------------------
+VENV="${DT_VENV:-$HOME/radical/digital_twins/ve.demo}"
+HERE="$(cd "$(dirname "$0")/.." && pwd)"     # xGFabric checkout root
+# ---------------------------------------------------------------------------
+
+echo "--------------------------"
+echo "xGFabric Demo September 04"
+echo "Sensor davis-wind ($(hostname -f)) -> broker $BROKER_IP"
+echo "--------------------------"
+
+export RADICAL_ORBIT_BROKER_URL="wss://$BROKER_IP:8000"
+export RADICAL_ORBIT_BROKER_CERT="$HOME/.radical/orbit/broker_cert.pem"
+export DT_STREAM_BACKEND=orbit
+
+cd "$HERE"
+echo "publishing davis-wind to channel xgf/davis ..."
+exec "$VENV/bin/python" service/sensor_publisher.py

service/README.md

@@ -0,0 +1,134 @@
+# 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>
+
+Driver: `twin_service_real.py` wires the real components (done, but
+experimental -- not yet validated end to end; it is the on-Perlmutter
+starting point).  The lazy-import refactor is in place, so the real
+investigators import TF-free on the client/broker; TF loads only in the
+task bodies on the endpoint.
+
+Client/broker still need the light deps the real components carry:
+
+    git submodule update --init --recursive          # pyspot (sensor)
+    <ve.demo>/bin/pip install python-dotenv numpy pandas
+
+Two items remain open before a real run is trustworthy:
+
+- **Shared filesystem.**  The real components write to
+  `config['PLAYGROUND_DIR']` from both main_loops (broker) and tasks
+  (endpoint); those line up only on a shared filesystem, so run the
+  **broker on Perlmutter too** for the real workload (PLAYGROUND_DIR on
+  `$SCRATCH`).  A broker on radical.3 splits the playground across hosts.
+- **cloudpickle parity.**  The cloned conda env and `ve.demo` may carry
+  different cloudpickle versions; align them 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    | endpoint-side rhapsody telemetry + `collect_reports.py` |
+
+## Telemetry and reports
+
+Nothing to enable: the rhapsody plugin on the endpoint records every
+task plus a resource poll on its own (the `[telemetry]` extra), into
+`telemetry-output/session.*.telemetry.jsonl` under the endpoint's
+working directory.  After (or during) a run:
+
+    <venv>/bin/python service/collect_reports.py [telemetry-dir]
+
+renders twin.py's report set — task waterfall, dependency wait, stage
+timers, swimlane, concurrency/resources, and the gantt — next to the
+jsonl.  For a remote endpoint, scp the jsonl files first.
+
+Known gap: per-workflow grouping in the gantt needs
+`asyncflow.workflow_id`, which only `workflow_scope()` stamps — the
+service engine does not run one.  That is an engine-side telemetry
+feature for the DT service (digital.twins), tracked there; every other
+report is complete without it.

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.