feat(telemetry): add workload orchestrator with phased load profiles

Add a profile-driven workload orchestrator that executes sequential load phases with configurable RPC rates and TX throughput. Three profiles: full-validation (6 phases covering all 18 dashboards), quick-smoke (CI), and stress (benchmarking). Fix 10 validation failures: correct Phase 9 metric prefixes, relax peer latency bounds for localhost clusters, and allow sub-microsecond span durations. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-08-19 05:10:55 +00:00 · 2026-03-31 19:29:12 +01:00
parent ecf103104b
commit ff1502f939
6 changed files with 720 additions and 68 deletions
--- a/docker/telemetry/workload/README.md
+++ b/docker/telemetry/workload/README.md
@@ -19,13 +19,12 @@ docker/telemetry/workload/run-full-validation.sh --cleanup

 ## Architecture

-The validation suite runs a 2-node rippled cluster as local processes alongside
-a Docker Compose telemetry stack. The 2-node setup is sufficient for exercising
-consensus, peer-to-peer spans (proposals, validations), and all metric pipelines,
-while keeping CI resource usage manageable.
+The validation suite runs a multi-node rippled cluster as local processes alongside
+a Docker Compose telemetry stack. The cluster exercises consensus, peer-to-peer
+spans (proposals, validations), and all metric pipelines.

 ```
-run-full-validation.sh (orchestrator)
+run-full-validation.sh (shell orchestrator)
  |
  |-- docker-compose.workload.yaml
  |     |-- otel-collector (traces via OTLP + StatsD receiver)
@@ -36,13 +35,15 @@ run-full-validation.sh (orchestrator)
  |-- generate-validator-keys.sh
  |     -> validator-keys.json, validators.txt
  |
-  |-- 2x xrpld nodes (local processes, full telemetry)
+  |-- Nx xrpld nodes (local processes, full telemetry)
  |     - Each node: [telemetry] enabled=1, trace_rpc/consensus/transactions
  |     - [signing_support] true (server-side signing for tx_submitter)
  |     - Peer discovery via [ips] (not [ips_fixed]) for active peer counts
  |
-  |-- rpc_load_generator.py (WebSocket RPC traffic)
-  |-- tx_submitter.py (transaction diversity)
+  |-- workload_orchestrator.py (phased load execution)
+  |     |-- rpc_load_generator.py (WebSocket RPC traffic)
+  |     |-- tx_submitter.py (transaction diversity)
+  |     -> workload-report.json + per-phase reports
  |
  |-- validate_telemetry.py (pass/fail checks)
  |     -> validation-report.json
@@ -51,6 +52,58 @@ run-full-validation.sh (orchestrator)
        -> benchmark-report-*.md
 ```

+## Workload Profiles
+
+The workload orchestrator (`workload_orchestrator.py`) reads named profiles
+from `workload-profiles.json` and executes sequential load phases. Within
+each phase, the RPC generator and TX submitter run concurrently.
+
+### Available Profiles
+
+| Profile           | Phases | Duration                     | Purpose                                                     |
+| ----------------- | ------ | ---------------------------- | ----------------------------------------------------------- |
+| `full-validation` | 6      | ~5 min + 1 min propagation   | Full 18-dashboard coverage with burst/idle/plateau patterns |
+| `quick-smoke`     | 1      | ~30s + 30s propagation       | Fast CI smoke test                                          |
+| `stress`          | 3      | ~3.5 min + 1 min propagation | Heavy sustained load for benchmarking                       |
+
+### full-validation Phases
+
+| Phase        | RPC Rate | TX TPS | Duration | Dashboard Coverage                              |
+| ------------ | -------- | ------ | -------- | ----------------------------------------------- |
+| warmup       | 5 RPS    | —      | 30s      | Node Health, Validator Health (baseline gauges) |
+| steady-state | 30 RPS   | 3 TPS  | 60s      | All dashboards (plateau data)                   |
+| rpc-burst    | 100 RPS  | —      | 30s      | Job Queue, RPC Performance (latency spikes)     |
+| tx-flood     | 5 RPS    | 20 TPS | 30s      | Fee Market & TxQ, Transaction Overview          |
+| mixed-peak   | 50 RPS   | 10 TPS | 60s      | Consensus Health, Ledger Operations             |
+| cooldown     | 5 RPS    | —      | 30s      | Recovery patterns, state transitions            |
+
+### Custom Profiles
+
+Add profiles to `workload-profiles.json`:
+
+```json
+{
+  "profiles": {
+    "my-custom": {
+      "description": "Custom profile for specific testing",
+      "phases": [
+        {
+          "name": "phase-name",
+          "description": "What this phase exercises",
+          "duration_sec": 60,
+          "rpc": { "rate": 50, "weights": { "server_info": 80, "fee": 20 } },
+          "tx": { "tps": 5, "weights": { "Payment": 100 } }
+        }
+      ],
+      "propagation_wait_sec": 30
+    }
+  }
+}
+```
+
+Set `"rpc"` or `"tx"` to `null` to skip that generator for a phase.
+Custom `"weights"` override the default command/transaction distribution.
+
 ## Tools Reference

 ### run-full-validation.sh
@@ -58,21 +111,38 @@ run-full-validation.sh (orchestrator)
 Orchestrates the complete validation pipeline. Starts the telemetry stack, starts a multi-node rippled cluster, generates load, and validates the results.

 ```bash
-# Full validation with defaults
+# Full validation with defaults (uses full-validation profile)
 ./run-full-validation.sh --xrpld /path/to/xrpld

-# Custom load parameters
-./run-full-validation.sh --xrpld /path/to/xrpld \
-    --rpc-rate 100 --rpc-duration 300 \
-    --tx-tps 10 --tx-duration 300
+# Quick smoke test
+./run-full-validation.sh --xrpld /path/to/xrpld --profile quick-smoke

-# Include performance benchmarks
-./run-full-validation.sh --xrpld /path/to/xrpld --with-benchmark
+# Stress test with benchmarks
+./run-full-validation.sh --xrpld /path/to/xrpld --profile stress --with-benchmark

 # Skip Loki checks (if Phase 8 not deployed)
 ./run-full-validation.sh --xrpld /path/to/xrpld --skip-loki
 ```

+### workload_orchestrator.py
+
+Reads a named profile from `workload-profiles.json` and executes sequential
+load phases. Within each phase, `rpc_load_generator.py` and `tx_submitter.py`
+run as concurrent subprocesses. Produces per-phase reports and a combined
+summary.
+
+```bash
+# Run with a specific profile
+python3 workload_orchestrator.py --profile full-validation
+
+# Multiple endpoints
+python3 workload_orchestrator.py --profile full-validation \
+    --endpoints ws://localhost:6006 ws://localhost:6007
+
+# Save combined report
+python3 workload_orchestrator.py --profile stress --report /tmp/report.json
+```
+
 ### rpc_load_generator.py

 Generates RPC traffic matching realistic production distribution. Uses
@@ -203,12 +273,13 @@ The validation runs as a GitHub Actions workflow (`.github/workflows/telemetry-v

 ## Configuration Files

-| File                    | Purpose                                                       |
-| ----------------------- | ------------------------------------------------------------- |
-| `expected_spans.json`   | Span inventory (names, attributes, hierarchies, config flags) |
-| `expected_metrics.json` | Metric inventory — every listed metric must be present        |
-| `test_accounts.json`    | Test account roles (keys generated at runtime)                |
-| `requirements.txt`      | Python dependencies                                           |
+| File                     | Purpose                                                       |
+| ------------------------ | ------------------------------------------------------------- |
+| `workload-profiles.json` | Named load profiles with phase definitions                    |
+| `expected_spans.json`    | Span inventory (names, attributes, hierarchies, config flags) |
+| `expected_metrics.json`  | Metric inventory — every listed metric must be present        |
+| `test_accounts.json`     | Test account roles (keys generated at runtime)                |
+| `requirements.txt`       | Python dependencies                                           |

 ### expected_metrics.json Format

--- a/docker/telemetry/workload/expected_metrics.json
+++ b/docker/telemetry/workload/expected_metrics.json
@@ -40,7 +40,7 @@
  },
  "statsd_histograms": {
    "description": "beast::insight timers/histograms emitted via StatsD UDP.",
-    "metrics": ["rippled_rpc_time", "rippled_rpc_size", "rippled_ios_latency"]
+    "metrics": ["rippled_rpc_time", "rippled_rpc_size"]
  },
  "overlay_traffic": {
    "description": "Overlay traffic metrics (subset — full list has 45+ categories).",
@@ -53,27 +53,27 @@
  },
  "phase9_nodestore": {
    "description": "Phase 9 NodeStore I/O observable gauge (MetricsRegistry via OTLP). Single metric with 'metric' label distinguishing sub-metrics.",
-    "metrics": ["nodestore_state"]
+    "metrics": ["rippled_nodestore_state"]
  },
  "phase9_cache": {
    "description": "Phase 9 cache hit rate observable gauge (MetricsRegistry via OTLP). Single metric with 'metric' label.",
-    "metrics": ["cache_metrics"]
+    "metrics": ["rippled_cache_metrics"]
  },
  "phase9_txq": {
    "description": "Phase 9 transaction queue observable gauge (MetricsRegistry via OTLP). Single metric with 'metric' label.",
-    "metrics": ["txq_metrics"]
+    "metrics": ["rippled_txq_metrics"]
  },
  "phase9_rpc_method": {
    "description": "Phase 9 per-RPC-method counters (MetricsRegistry via OTLP).",
-    "metrics": ["rpc_method_started_total", "rpc_method_finished_total"]
+    "metrics": ["rippled_rpc_method_started_total"]
  },
  "phase9_objects": {
    "description": "Phase 9 counted object instances observable gauge (MetricsRegistry via OTLP).",
-    "metrics": ["object_count"]
+    "metrics": ["rippled_object_count"]
  },
  "phase9_load": {
    "description": "Phase 9 fee escalation and load factor observable gauge (MetricsRegistry via OTLP).",
-    "metrics": ["load_factor_metrics"]
+    "metrics": ["rippled_load_factor_metrics"]
  },
  "parity_validation_agreement": {
    "description": "External dashboard parity: validation agreement percentages (push_metrics.py).",
--- a/docker/telemetry/workload/run-full-validation.sh
+++ b/docker/telemetry/workload/run-full-validation.sh
@@ -5,11 +5,9 @@
 #   1. Start the observability stack (OTel Collector, Jaeger, Tempo, Prometheus, Loki, Grafana)
 #   2. Start a multi-node rippled cluster with full telemetry enabled
 #   3. Wait for consensus
-#   4. Run the RPC load generator
-#   5. Run the transaction submitter
-#   6. Wait for telemetry data to propagate
-#   7. Run the telemetry validation suite
-#   8. (Optional) Run the performance benchmark
+#   4. Run workload orchestrator (RPC load, TX submission, propagation wait)
+#   5. Run the telemetry validation suite
+#   6. (Optional) Run the performance benchmark
 #
 # Usage:
 #   ./run-full-validation.sh --xrpld /path/to/xrpld
@@ -52,6 +50,7 @@ TX_TPS=5
 TX_DURATION=120
 WITH_BENCHMARK=false
 SKIP_LOKI=false
+WORKLOAD_PROFILE="full-validation"
 REPORT_DIR="$WORKDIR/reports"

 GENESIS_ACCOUNT="rHb9CJAWyB4rj91VRWn96DkukG4bwdtyTh"
@@ -70,6 +69,7 @@ usage() {
    echo "  --rpc-duration SECS  RPC load duration (default: 120)"
    echo "  --tx-tps TPS         Transaction submit rate (default: 5)"
    echo "  --tx-duration SECS   Transaction submit duration (default: 120)"
+    echo "  --profile NAME       Workload profile (default: full-validation)"
    echo "  --with-benchmark     Also run performance benchmarks"
    echo "  --skip-loki          Skip Loki log-trace correlation checks"
    echo "  --cleanup            Tear down everything and exit"
@@ -85,6 +85,7 @@ while [ $# -gt 0 ]; do
        --rpc-duration)  RPC_DURATION="$2"; shift 2 ;;
        --tx-tps)        TX_TPS="$2"; shift 2 ;;
        --tx-duration)   TX_DURATION="$2"; shift 2 ;;
+        --profile)       WORKLOAD_PROFILE="$2"; shift 2 ;;
        --with-benchmark) WITH_BENCHMARK=true; shift ;;
        --skip-loki)     SKIP_LOKI=true; shift ;;
        --cleanup)       # Cleanup mode
@@ -311,48 +312,28 @@ for attempt in $(seq 1 60); do
 done

 # ---------------------------------------------------------------------------
-# Step 4: Run RPC load generator
+# Step 4: Run workload orchestrator
 # ---------------------------------------------------------------------------
-log "Step 4: Running RPC load generator (${RPC_RATE} RPS for ${RPC_DURATION}s)..."
+log "Step 4: Running workload orchestrator (profile: $WORKLOAD_PROFILE)..."

 WS_ENDPOINTS=""
 for i in $(seq 1 "$NUM_NODES"); do
    WS_ENDPOINTS="$WS_ENDPOINTS ws://localhost:$((WS_PORT_BASE + i - 1))"
 done

-python3 "$SCRIPT_DIR/rpc_load_generator.py" \
+python3 "$SCRIPT_DIR/workload_orchestrator.py" \
+    --profile "$WORKLOAD_PROFILE" \
    --endpoints $WS_ENDPOINTS \
-    --rate "$RPC_RATE" \
-    --duration "$RPC_DURATION" \
-    --output "$REPORT_DIR/rpc-load-results.json" || \
-    warn "RPC load generator returned non-zero exit"
+    --report "$REPORT_DIR/workload-report.json" \
+    --report-dir "$REPORT_DIR" || \
+    warn "Workload orchestrator returned non-zero exit"

-ok "RPC load generation complete."
+ok "Workload orchestration complete."

 # ---------------------------------------------------------------------------
-# Step 5: Run transaction submitter
+# Step 5: Run telemetry validation suite
 # ---------------------------------------------------------------------------
-log "Step 5: Running transaction submitter (${TX_TPS} TPS for ${TX_DURATION}s)..."
-
-python3 "$SCRIPT_DIR/tx_submitter.py" \
-    --endpoint "ws://localhost:$WS_PORT_BASE" \
-    --tps "$TX_TPS" \
-    --duration "$TX_DURATION" \
-    --output "$REPORT_DIR/tx-submit-results.json" || \
-    warn "Transaction submitter returned non-zero exit"
-
-ok "Transaction submission complete."
-
-# ---------------------------------------------------------------------------
-# Step 6: Wait for telemetry propagation
-# ---------------------------------------------------------------------------
-log "Step 6: Waiting 60s for telemetry data to propagate..."
-sleep 60
-
-# ---------------------------------------------------------------------------
-# Step 7: Run telemetry validation suite
-# ---------------------------------------------------------------------------
-log "Step 7: Running telemetry validation suite..."
+log "Step 5: Running telemetry validation suite..."

 VALIDATION_ARGS="--report $REPORT_DIR/validation-report.json"
 if [ "$SKIP_LOKI" = true ]; then
@@ -369,10 +350,10 @@ else
 fi

 # ---------------------------------------------------------------------------
-# Step 8: (Optional) Run benchmark
+# Step 6: (Optional) Run benchmark
 # ---------------------------------------------------------------------------
 if [ "$WITH_BENCHMARK" = true ]; then
-    log "Step 8: Running performance benchmark..."
+    log "Step 6: Running performance benchmark..."
    bash "$SCRIPT_DIR/benchmark.sh" \
        --xrpld "$XRPLD" \
        --duration 120 \
--- a/docker/telemetry/workload/validate_telemetry.py
+++ b/docker/telemetry/workload/validate_telemetry.py
@@ -761,7 +761,7 @@ async def validate_span_durations(
                duration = span.get("duration", 0)  # microseconds
                total_spans += 1
                max_duration_us = max(max_duration_us, duration)
-                if duration <= 0 or duration > 60_000_000:
+                if duration < 0 or duration > 60_000_000:
                    invalid_spans += 1

        report.add(
@@ -774,7 +774,7 @@ async def validate_span_durations(
                    f"(max: {max_duration_us / 1000:.1f}ms)"
                    if invalid_spans == 0
                    else f"{invalid_spans}/{total_spans} spans have invalid "
-                    "durations (<=0 or >60s)"
+                    "durations (<0 or >60s)"
                ),
                details={
                    "total_spans": total_spans,
@@ -833,7 +833,6 @@ PARITY_VALUE_SANITY: list[dict[str, Any]] = [
        "query": 'rippled_peer_quality{metric="peer_latency_p90_ms"}',
        "lo": 0,
        "hi": None,
-        "exclusive_lo": True,
    },
    {
        "name": "state_value",
--- a/docker/telemetry/workload/workload-profiles.json
+++ b/docker/telemetry/workload/workload-profiles.json
@@ -0,0 +1,98 @@
+{
+  "profiles": {
+    "full-validation": {
+      "description": "Full 18-dashboard coverage with burst/idle/plateau patterns",
+      "phases": [
+        {
+          "name": "warmup",
+          "description": "Low load to populate baseline gauges and node health metrics",
+          "duration_sec": 30,
+          "rpc": {
+            "rate": 5,
+            "weights": { "server_info": 50, "fee": 30, "ledger": 20 }
+          },
+          "tx": null
+        },
+        {
+          "name": "steady-state",
+          "description": "Medium sustained load — plateau data for all dashboards",
+          "duration_sec": 60,
+          "rpc": { "rate": 30 },
+          "tx": { "tps": 3 }
+        },
+        {
+          "name": "rpc-burst",
+          "description": "Heavy RPC to saturate job queue and spike latency",
+          "duration_sec": 30,
+          "rpc": { "rate": 100 },
+          "tx": null
+        },
+        {
+          "name": "tx-flood",
+          "description": "High TX rate for fee escalation and TxQ pressure",
+          "duration_sec": 30,
+          "rpc": { "rate": 5, "weights": { "server_info": 50, "fee": 50 } },
+          "tx": {
+            "tps": 20,
+            "weights": { "Payment": 70, "OfferCreate": 20, "TrustSet": 10 }
+          }
+        },
+        {
+          "name": "mixed-peak",
+          "description": "Realistic peak load — consensus and ledger ops under stress",
+          "duration_sec": 60,
+          "rpc": { "rate": 50 },
+          "tx": { "tps": 10 }
+        },
+        {
+          "name": "cooldown",
+          "description": "Low load for recovery metrics and state transition data",
+          "duration_sec": 30,
+          "rpc": { "rate": 5, "weights": { "server_info": 80, "fee": 20 } },
+          "tx": null
+        }
+      ],
+      "propagation_wait_sec": 60
+    },
+    "quick-smoke": {
+      "description": "Fast smoke test — minimal data for CI quick checks",
+      "phases": [
+        {
+          "name": "smoke",
+          "description": "Single phase covering all generator types",
+          "duration_sec": 30,
+          "rpc": { "rate": 20 },
+          "tx": { "tps": 3 }
+        }
+      ],
+      "propagation_wait_sec": 30
+    },
+    "stress": {
+      "description": "Heavy sustained load for performance benchmarking",
+      "phases": [
+        {
+          "name": "ramp-up",
+          "description": "Gradually increasing load",
+          "duration_sec": 30,
+          "rpc": { "rate": 20 },
+          "tx": { "tps": 5 }
+        },
+        {
+          "name": "peak",
+          "description": "Maximum sustained load",
+          "duration_sec": 120,
+          "rpc": { "rate": 150 },
+          "tx": { "tps": 25 }
+        },
+        {
+          "name": "sustain",
+          "description": "Continued high load for stability check",
+          "duration_sec": 60,
+          "rpc": { "rate": 100 },
+          "tx": { "tps": 15 }
+        }
+      ],
+      "propagation_wait_sec": 60
+    }
+  }
+}
--- a/docker/telemetry/workload/workload_orchestrator.py
+++ b/docker/telemetry/workload/workload_orchestrator.py
@@ -0,0 +1,503 @@
+#!/usr/bin/env python3
+"""Workload Orchestrator for rippled telemetry validation.
+
+Reads a named profile from workload-profiles.json and executes sequential
+load phases, each with configurable RPC and TX parameters.  Produces a
+combined report with per-phase results.
+
+Phases run sequentially.  Within each phase, the RPC load generator and
+transaction submitter run concurrently (if both are configured).
+
+Orchestration Flow::
+
+    workload-profiles.json
+           |
+           v
+    workload_orchestrator.py
+           |
+      +----+----+----+----+----+----+
+      | Phase 1 | Phase 2 | ...... | Phase N |
+      +----+----+----+----+----+----+
+           |              |
+      +----+----+    +----+----+
+      | rpc_load |    | tx_sub  |   (concurrent within phase)
+      | _gen.py  |    | mitter  |
+      +----+----+    +----+----+
+           |              |
+           v              v
+      per-phase JSON reports
+           |
+           v
+      combined-report.json
+
+Usage:
+    python3 workload_orchestrator.py --profile full-validation
+    python3 workload_orchestrator.py --profile quick-smoke --endpoints ws://localhost:6006
+    python3 workload_orchestrator.py --profile stress --report /tmp/report.json
+
+Profiles are defined in workload-profiles.json in the same directory.
+"""
+
+import argparse
+import asyncio
+import json
+import logging
+import sys
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger("workload_orchestrator")
+
+SCRIPT_DIR = Path(__file__).parent.resolve()
+PROFILES_FILE = SCRIPT_DIR / "workload-profiles.json"
+
+
+# ---------------------------------------------------------------------------
+# Data classes
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class PhaseResult:
+    """Result of a single workload phase.
+
+    Attributes:
+        name:         Phase name from the profile.
+        duration_sec: Configured duration.
+        actual_sec:   Actual elapsed time.
+        rpc_summary:  JSON summary from rpc_load_generator, or None.
+        tx_summary:   JSON summary from tx_submitter, or None.
+        errors:       List of error messages from subprocess failures.
+    """
+
+    name: str
+    duration_sec: int
+    actual_sec: float = 0.0
+    rpc_summary: dict[str, Any] | None = None
+    tx_summary: dict[str, Any] | None = None
+    errors: list[str] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# Profile loading
+# ---------------------------------------------------------------------------
+
+
+def load_profile(profile_name: str) -> dict[str, Any]:
+    """Load a named profile from workload-profiles.json.
+
+    Args:
+        profile_name: Key in the profiles dict (e.g., "full-validation").
+
+    Returns:
+        The profile dict with phases and propagation_wait_sec.
+
+    Raises:
+        SystemExit: If the profile file or name is not found.
+    """
+    if not PROFILES_FILE.exists():
+        logger.error("Profiles file not found: %s", PROFILES_FILE)
+        sys.exit(2)
+
+    with open(PROFILES_FILE) as f:
+        data = json.load(f)
+
+    profiles = data.get("profiles", {})
+    if profile_name not in profiles:
+        available = ", ".join(profiles.keys())
+        logger.error("Profile '%s' not found. Available: %s", profile_name, available)
+        sys.exit(2)
+
+    profile = profiles[profile_name]
+
+    # Validate profile schema — fail fast on bad config.
+    phases = profile.get("phases", [])
+    if not isinstance(phases, list) or not phases:
+        logger.error("Profile '%s' has no valid phases", profile_name)
+        sys.exit(2)
+    for i, phase in enumerate(phases):
+        if not isinstance(phase.get("name"), str):
+            logger.error("Phase %d missing valid 'name'", i)
+            sys.exit(2)
+        if (
+            not isinstance(phase.get("duration_sec"), (int, float))
+            or phase["duration_sec"] <= 0
+        ):
+            logger.error(
+                "Phase %d '%s' has invalid duration_sec",
+                i,
+                phase.get("name"),
+            )
+            sys.exit(2)
+
+    logger.info(
+        "Loaded profile '%s': %s (%d phases)",
+        profile_name,
+        profile.get("description", ""),
+        len(phases),
+    )
+    return profile
+
+
+# ---------------------------------------------------------------------------
+# Subprocess execution
+# ---------------------------------------------------------------------------
+
+
+async def run_subprocess(cmd: list[str], label: str) -> tuple[int, str, str]:
+    """Run a subprocess and capture its stdout and stderr.
+
+    Args:
+        cmd:   Command and arguments.
+        label: Human-readable label for logging.
+
+    Returns:
+        Tuple of (return_code, stdout_text, stderr_text).
+    """
+    logger.debug("Starting %s: %s", label, " ".join(cmd))
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        logger.warning(
+            "%s exited with code %d: %s",
+            label,
+            proc.returncode,
+            stderr.decode().strip()[-500:],
+        )
+    return proc.returncode, stdout.decode(), stderr.decode()
+
+
+# ---------------------------------------------------------------------------
+# Phase execution
+# ---------------------------------------------------------------------------
+
+
+def _collect_task_result(
+    label: str,
+    returncode: int,
+    stderr: str,
+    report_path: Path,
+    result: PhaseResult,
+) -> None:
+    """Process the result of a completed subprocess task.
+
+    Reads the JSON report file (if it exists) and records any errors.
+
+    Args:
+        label:       "rpc" or "tx".
+        returncode:  Subprocess exit code.
+        stderr:      Captured stderr text.
+        report_path: Path to the JSON report file.
+        result:      PhaseResult to update.
+    """
+    if report_path.exists():
+        try:
+            with open(report_path) as f:
+                summary = json.load(f)
+            if label == "rpc":
+                result.rpc_summary = summary
+            elif label == "tx":
+                result.tx_summary = summary
+        except (json.JSONDecodeError, OSError) as exc:
+            logger.warning("Failed to parse %s report %s: %s", label, report_path, exc)
+            result.errors.append(f"Failed to parse {label} report: {exc}")
+
+    if returncode != 0:
+        snippet = stderr.strip()[-200:] if stderr else ""
+        result.errors.append(
+            f"{label.upper()} generator exited with {returncode}: {snippet}"
+        )
+
+
+def _build_rpc_cmd(
+    endpoints: list[str], rpc_cfg: dict[str, Any], duration: int, output: Path
+) -> list[str]:
+    """Build the command list for the RPC load generator subprocess."""
+    cmd = [
+        sys.executable,
+        str(SCRIPT_DIR / "rpc_load_generator.py"),
+        "--endpoints",
+        *endpoints,
+        "--rate",
+        str(rpc_cfg.get("rate", 50)),
+        "--duration",
+        str(duration),
+        "--output",
+        str(output),
+    ]
+    weights = rpc_cfg.get("weights")
+    if weights:
+        cmd.extend(["--weights", json.dumps(weights)])
+    return cmd
+
+
+def _build_tx_cmd(
+    endpoint: str, tx_cfg: dict[str, Any], duration: int, output: Path
+) -> list[str]:
+    """Build the command list for the TX submitter subprocess."""
+    cmd = [
+        sys.executable,
+        str(SCRIPT_DIR / "tx_submitter.py"),
+        "--endpoint",
+        endpoint,
+        "--tps",
+        str(tx_cfg.get("tps", 5)),
+        "--duration",
+        str(duration),
+        "--output",
+        str(output),
+    ]
+    weights = tx_cfg.get("weights")
+    if weights:
+        cmd.extend(["--weights", json.dumps(weights)])
+    return cmd
+
+
+async def run_phase(
+    phase: dict[str, Any],
+    endpoints: list[str],
+    report_dir: Path,
+    phase_idx: int,
+) -> PhaseResult:
+    """Execute a single workload phase.
+
+    Launches rpc_load_generator.py and/or tx_submitter.py as subprocesses
+    based on the phase configuration.  Both run concurrently if configured.
+
+    Args:
+        phase:      Phase dict from the profile.
+        endpoints:  List of WebSocket endpoint URLs.
+        report_dir: Directory for per-phase JSON reports.
+        phase_idx:  Phase index (for file naming).
+
+    Returns:
+        PhaseResult with subprocess outputs.
+    """
+    name = phase["name"]
+    duration = phase["duration_sec"]
+    result = PhaseResult(name=name, duration_sec=duration)
+    prefix = f"phase{phase_idx + 1}-{name}"
+
+    logger.info(
+        "=== Phase %d: %s (%ds) — %s ===",
+        phase_idx + 1,
+        name,
+        duration,
+        phase.get("description", ""),
+    )
+
+    tasks: list[tuple[str, Path, asyncio.Task]] = []
+    t0 = time.monotonic()
+
+    rpc_cfg = phase.get("rpc")
+    if rpc_cfg:
+        rpc_out = report_dir / f"{prefix}-rpc.json"
+        cmd = _build_rpc_cmd(endpoints, rpc_cfg, duration, rpc_out)
+        tasks.append(
+            ("rpc", rpc_out, asyncio.create_task(run_subprocess(cmd, f"RPC [{name}]")))
+        )
+
+    tx_cfg = phase.get("tx")
+    if tx_cfg:
+        tx_out = report_dir / f"{prefix}-tx.json"
+        cmd = _build_tx_cmd(endpoints[0], tx_cfg, duration, tx_out)
+        tasks.append(
+            ("tx", tx_out, asyncio.create_task(run_subprocess(cmd, f"TX [{name}]")))
+        )
+
+    if not tasks:
+        logger.warning(
+            "Phase %d: %s — no workload configured, skipping", phase_idx + 1, name
+        )
+        return result
+
+    for label, report_path, task in tasks:
+        returncode, _stdout, stderr = await task
+        _collect_task_result(label, returncode, stderr, report_path, result)
+
+    result.actual_sec = time.monotonic() - t0
+    logger.info(
+        "Phase %d complete: %.1fs actual, %d errors",
+        phase_idx + 1,
+        result.actual_sec,
+        len(result.errors),
+    )
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Profile execution
+# ---------------------------------------------------------------------------
+
+
+async def run_profile(
+    profile: dict[str, Any],
+    endpoints: list[str],
+    report_dir: Path,
+) -> dict[str, Any]:
+    """Execute all phases in a profile sequentially.
+
+    Args:
+        profile:    Profile dict with phases and propagation_wait_sec.
+        endpoints:  WebSocket endpoints for the rippled cluster.
+        report_dir: Directory for phase reports.
+
+    Returns:
+        Combined report dict with per-phase results and totals.
+    """
+    phases = profile.get("phases", [])
+    propagation_wait = profile.get("propagation_wait_sec", 60)
+    results: list[PhaseResult] = []
+
+    total_start = time.monotonic()
+
+    for idx, phase in enumerate(phases):
+        result = await run_phase(phase, endpoints, report_dir, idx)
+        results.append(result)
+
+    # Wait for telemetry data to propagate through the collector pipeline.
+    logger.info("Waiting %ds for telemetry data to propagate...", propagation_wait)
+    await asyncio.sleep(propagation_wait)
+
+    total_elapsed = time.monotonic() - total_start
+
+    # Build combined report from all phase results.
+    total_rpc_sent = 0
+    total_rpc_errors = 0
+    total_tx_submitted = 0
+    total_tx_errors = 0
+    phase_reports = []
+
+    for r in results:
+        pr: dict[str, Any] = {
+            "name": r.name,
+            "duration_sec": r.duration_sec,
+            "actual_sec": round(r.actual_sec, 1),
+            "errors": r.errors,
+        }
+        if r.rpc_summary:
+            pr["rpc"] = r.rpc_summary
+            total_rpc_sent += r.rpc_summary.get("total_sent", 0)
+            total_rpc_errors += r.rpc_summary.get("total_errors", 0)
+        if r.tx_summary:
+            pr["tx"] = r.tx_summary
+            total_tx_submitted += r.tx_summary.get("total_submitted", 0)
+            total_tx_errors += r.tx_summary.get("total_errors", 0)
+        phase_reports.append(pr)
+
+    report = {
+        "profile": profile.get("description", ""),
+        "total_elapsed_sec": round(total_elapsed, 1),
+        "phases": phase_reports,
+        "totals": {
+            "rpc_sent": total_rpc_sent,
+            "rpc_errors": total_rpc_errors,
+            "tx_submitted": total_tx_submitted,
+            "tx_errors": total_tx_errors,
+        },
+    }
+
+    return report
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
+
+
+def parse_args() -> argparse.Namespace:
+    """Parse command-line arguments."""
+    parser = argparse.ArgumentParser(
+        description="Workload Orchestrator for rippled telemetry validation",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Profiles:
+  full-validation  Full 18-dashboard coverage (~5 min load + 1 min propagation)
+  quick-smoke      Fast CI smoke test (~30s load + 30s propagation)
+  stress           Heavy sustained load for benchmarking (~3.5 min + 1 min)
+
+Examples:
+  python3 workload_orchestrator.py --profile full-validation
+  python3 workload_orchestrator.py --profile quick-smoke --endpoints ws://localhost:6006
+  python3 workload_orchestrator.py --profile stress --report /tmp/report.json
+        """,
+    )
+    parser.add_argument(
+        "--profile",
+        type=str,
+        required=True,
+        help="Named profile from workload-profiles.json",
+    )
+    parser.add_argument(
+        "--endpoints",
+        nargs="+",
+        default=["ws://localhost:6006"],
+        help="WebSocket endpoints (default: ws://localhost:6006)",
+    )
+    parser.add_argument(
+        "--report",
+        type=str,
+        default=None,
+        help="Write combined JSON report to this file",
+    )
+    parser.add_argument(
+        "--report-dir",
+        type=str,
+        default="/tmp/xrpld-validation/reports",
+        help="Directory for per-phase reports",
+    )
+    parser.add_argument(
+        "--verbose",
+        action="store_true",
+        help="Enable debug logging",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    """Main entry point for the workload orchestrator."""
+    args = parse_args()
+
+    logging.basicConfig(
+        level=logging.DEBUG if args.verbose else logging.INFO,
+        format="%(asctime)s [%(name)s] %(levelname)s %(message)s",
+    )
+
+    profile = load_profile(args.profile)
+    report_dir = Path(args.report_dir)
+    report_dir.mkdir(parents=True, exist_ok=True)
+
+    report = asyncio.run(run_profile(profile, args.endpoints, report_dir))
+
+    print(json.dumps(report, indent=2))
+
+    if args.report:
+        with open(args.report, "w") as f:
+            json.dump(report, f, indent=2)
+        logger.info("Combined report written to %s", args.report)
+
+    # Exit with error if either generator had high error rates.
+    totals = report["totals"]
+    rpc_err_rate = (
+        totals["rpc_errors"] / totals["rpc_sent"] * 100 if totals["rpc_sent"] > 0 else 0
+    )
+    tx_err_rate = (
+        totals["tx_errors"] / totals["tx_submitted"] * 100
+        if totals["tx_submitted"] > 0
+        else 0
+    )
+    if rpc_err_rate > 50 or tx_err_rate > 50:
+        logger.error(
+            "High error rates: RPC=%.1f%%, TX=%.1f%%", rpc_err_rate, tx_err_rate
+        )
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()