diff --git a/docker/telemetry/workload/README.md b/docker/telemetry/workload/README.md index 5977b643a5..579b88f35c 100644 --- 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 diff --git a/docker/telemetry/workload/expected_metrics.json b/docker/telemetry/workload/expected_metrics.json index 318d346580..f55d2feb6c 100644 --- 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).", diff --git a/docker/telemetry/workload/run-full-validation.sh b/docker/telemetry/workload/run-full-validation.sh index 91cca9fc69..39581fa119 100755 --- 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 \ diff --git a/docker/telemetry/workload/validate_telemetry.py b/docker/telemetry/workload/validate_telemetry.py index d4ba030782..ba117539b9 100644 --- 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", diff --git a/docker/telemetry/workload/workload-profiles.json b/docker/telemetry/workload/workload-profiles.json new file mode 100644 index 0000000000..040be9073d --- /dev/null +++ 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 + } + } +} diff --git a/docker/telemetry/workload/workload_orchestrator.py b/docker/telemetry/workload/workload_orchestrator.py new file mode 100755 index 0000000000..6a679f5b54 --- /dev/null +++ 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()