scripts.benchmark_pipeline

Reference › scripts › benchmark_pipeline

Performance benchmarks for the PipelineRunner and the infrastructure under it. A human runs it, and tests/scripts/test_benchmark_pipeline.py imports it. It uses the environment bootstrap in env_bootstrap and the graph factory in graph_factory.

Invocation: main loads the dotenv file, then reads --boot-iter (default 5), --intent-iter (10), --throughput-sec (5.0), --workers (4) and --per-worker (10). _run_all executes six benchmarks and _print_summary prints the results: benchmark_boot (runner start-up), benchmark_submit_intent (latency of one intent submission), benchmark_intent_throughput (intents per second over a fixed duration), benchmark_parallel_intents (several worker threads each submitting a batch), benchmark_redis_ping and benchmark_graph_query. Each returns a dictionary of figures, with percentiles from _percentile and times formatted by _fmt_ms.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/benchmark_pipeline.py

#!/usr/bin/env python3
"""Performance benchmarks for PipelineRunner and infrastructure.

Measures boot time, intent submission throughput, end-to-end pipeline
latency, Redis pub/sub routing, and Neo4j KG sync performance.
"""

from __future__ import annotations

import argparse
import os
import statistics
import sys
import time
from collections.abc import Mapping
from concurrent.futures import ThreadPoolExecutor
from typing import Any

from orchestrata_ultimate.core.env_bootstrap import load_env_contract
from orchestrata_ultimate.core.graph_factory import resolve_kg_database, resolve_uri
from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)


def _fmt_ms(seconds: float) -> str:
    """Format seconds as milliseconds with 3 decimal places."""
    return f"{seconds * 1000:.3f}ms"


def _percentile(sorted_times: list[float], p: float) -> float:
    """Compute percentile from sorted list of times (in seconds)."""
    if not sorted_times:
        return 0.0
    idx = int(len(sorted_times) * p / 100)
    idx = min(idx, len(sorted_times) - 1)
    return sorted_times[idx]


def benchmark_boot(iterations: int = 5) -> dict:
    """Benchmark PipelineRunner.boot() performance."""
    from orchestrata_ultimate.kernel.pipeline import PipelineRunner

    times = []
    for i in range(iterations):
        runner = PipelineRunner()
        start = time.perf_counter()
        runner.boot()
        elapsed = time.perf_counter() - start
        times.append(elapsed)
        print(f"  Boot {i + 1}: {_fmt_ms(elapsed)}")

    times.sort()
    return {
        "operation": "boot",
        "iterations": iterations,
        "mean_ms": statistics.mean(times) * 1000,
        "min_ms": times[0] * 1000,
        "max_ms": times[-1] * 1000,
        "p50_ms": _percentile(times, 50) * 1000,
        "p90_ms": _percentile(times, 90) * 1000,
    }


def benchmark_submit_intent(iterations: int = 10, runner: Any | None = None) -> dict:
    """Benchmark PipelineRunner.submit_intent() latency.

    Args:
        iterations: Number of intent submissions to time.
        runner: Optional already-booted ``PipelineRunner`` to reuse.  When
            omitted, a fresh runner is created and booted (CLI behavior).

    Degrades gracefully when the downstream volk/LLM is unavailable: the
    benchmark records the failure and continues so a single external provider
    outage does not abort the whole suite.
    """
    from orchestrata_ultimate.kernel.pipeline import PipelineRunner

    if runner is None:
        runner = PipelineRunner()
        runner.boot()

    times: list[float] = []
    errors: list[str] = []
    for i in range(iterations):
        start = time.perf_counter()
        try:
            runner.submit_intent(f"Benchmark intent {i}")
            elapsed = time.perf_counter() - start
            times.append(elapsed)
            print(f"  Submit intent {i + 1}: {_fmt_ms(elapsed)}")
        except Exception as exc:  # noqa: BLE001 - degrade on external provider failure
            errors.append(str(exc))
            print(f"  Submit intent {i + 1}: FAILED ({exc})")

    times.sort()
    result: dict[str, Any] = {
        "operation": "submit_intent",
        "iterations": iterations,
        "successful": len(times),
        "mean_ms": statistics.mean(times) * 1000 if times else 0.0,
        "min_ms": times[0] * 1000 if times else 0.0,
        "max_ms": times[-1] * 1000 if times else 0.0,
        "p50_ms": _percentile(times, 50) * 1000,
        "p90_ms": _percentile(times, 90) * 1000,
    }
    if errors:
        result["errors"] = errors
    return result


def benchmark_intent_throughput(
    duration_sec: float = 5.0, runner: Any | None = None
) -> dict:
    """Measure intent submission throughput (requests/sec).

    Submits intents as fast as possible for the given duration.

    Args:
        duration_sec: How long to submit intents.
        runner: Optional already-booted ``PipelineRunner`` to reuse.  When
            omitted, a fresh runner is created and booted (CLI behavior).

    Degrades gracefully when the downstream volk/LLM is unavailable.
    """
    from orchestrata_ultimate.kernel.pipeline import PipelineRunner

    if runner is None:
        runner = PipelineRunner()
        runner.boot()

    count = 0
    failures = 0
    start = time.perf_counter()
    while time.perf_counter() - start < duration_sec:
        try:
            runner.submit_intent(f"Throughput test {count}")
            count += 1
        except Exception:  # noqa: BLE001 - degrade on external provider failure
            failures += 1

    elapsed = time.perf_counter() - start
    rps = count / elapsed if elapsed > 0 else 0.0
    status = f"{count} intents"
    if failures:
        status += f", {failures} failures"
    print(f"  Submitted {status} in {elapsed:.2f}s → {rps:.1f} req/s")

    result: dict[str, Any] = {
        "operation": "intent_throughput",
        "duration_sec": elapsed,
        "requests": count,
        "rps": rps,
    }
    if failures:
        result["failures"] = failures
    return result


def benchmark_parallel_intents(
    workers: int = 4, per_worker: int = 10, runner: Any | None = None
) -> dict:
    """Benchmark parallel intent submission with thread pool.

    Args:
        workers: Number of concurrent worker threads.
        per_worker: Intents submitted per worker.
        runner: Optional already-booted ``PipelineRunner`` to reuse.  When
            omitted, a fresh runner is created and booted (CLI behavior).

    Degrades gracefully when the downstream volk/LLM is unavailable.
    """
    from orchestrata_ultimate.kernel.pipeline import PipelineRunner

    if runner is None:
        runner = PipelineRunner()
        runner.boot()
    lock = __import__("threading").Lock()
    times: list[float] = []
    failures = 0

    def _submit_batch(worker_id: int) -> None:
        nonlocal failures
        for i in range(per_worker):
            start = time.perf_counter()
            try:
                with lock:
                    runner.submit_intent(f"Parallel W{worker_id}-{i}")
                elapsed = time.perf_counter() - start
                times.append(elapsed)
            except Exception:  # noqa: BLE001 - degrade on external provider failure
                with lock:
                    failures += 1

    start = time.perf_counter()
    with ThreadPoolExecutor(max_workers=workers) as pool:
        for w in range(workers):
            pool.submit(_submit_batch, w)
    elapsed = time.perf_counter() - start

    times.sort()
    total = workers * per_worker
    succeeded = len(times)
    status = f"{succeeded}/{total} intents"
    if failures:
        status += f", {failures} failures"
    print(f"  {status} from {workers} workers in {elapsed:.2f}s")
    # Throughput counts SUCCESSFUL submissions only — computing it from the
    # nominal total reported positive req/s for fully-failed runs.
    rps = succeeded / elapsed if elapsed > 0 else 0.0
    print(f"  Total throughput: {rps:.1f} req/s")

    result: dict[str, Any] = {
        "operation": "parallel_intents",
        "workers": workers,
        "per_worker": per_worker,
        "total": total,
        "succeeded": succeeded,
        "duration_sec": elapsed,
        "rps": rps,
        "mean_ms": statistics.mean(times) * 1000 if times else 0.0,
        "min_ms": times[0] * 1000 if times else 0.0,
        "max_ms": times[-1] * 1000 if times else 0.0,
        "p50_ms": _percentile(times, 50) * 1000,
        "p90_ms": _percentile(times, 90) * 1000,
    }
    if failures:
        result["failures"] = failures
    return result


def benchmark_redis_ping(
    iterations: int = 100, env: Mapping[str, str] | None = None
) -> dict:
    """Benchmark Redis round-trip latency.

    Args:
        iterations: Number of PING round-trips to time.
        env: Optional env mapping to resolve REDIS_HOST/REDIS_PORT from
            (defaults to ``os.environ``) — lets callers (tests) point this
            at an unreachable host without mutating process env.
    """
    import redis

    env = env if env is not None else os.environ
    host = resolve_redis_host(env)
    port = resolve_redis_port(env)

    client = redis.Redis(
        host=host,
        port=port,
        db=0,
        socket_connect_timeout=5,
        socket_timeout=5,
    )

    times = []
    for _ in range(iterations):
        start = time.perf_counter()
        client.ping()
        elapsed = time.perf_counter() - start
        times.append(elapsed)

    client.close()
    times.sort()

    print(f"  Redis ping x{iterations}: mean={_fmt_ms(statistics.mean(times))}")

    return {
        "operation": "redis_ping",
        "iterations": iterations,
        "mean_ms": statistics.mean(times) * 1000,
        "min_ms": times[0] * 1000,
        "max_ms": times[-1] * 1000,
        "p50_ms": _percentile(times, 50) * 1000,
        "p90_ms": _percentile(times, 90) * 1000,
    }


def benchmark_graph_query(
    iterations: int = 50, env: Mapping[str, str] | None = None
) -> dict:
    """Benchmark Neo4j query latency.

    Args:
        iterations: Number of count queries to time.
        env: Optional env mapping to resolve NEO4J_URI/NEO4J_USERNAME/
            NEO4J_PASSWORD from (defaults to ``os.environ``) — lets callers
            (tests) point this at an unreachable URI without mutating
            process env.
    """
    from neo4j import GraphDatabase

    env = env if env is not None else os.environ
    # Only the ADDRESS is required. This also demanded a username; every
    # remaining graph endpoint negotiates no auth (NEO4J_USERNAME/PASSWORD were
    # retired with the Neo4j containers on 2026-07-27), so the credential half
    # of the guard could only ever raise.
    uri = resolve_uri(env=env)

    driver = GraphDatabase.driver(uri, auth=None)
    driver.verify_connectivity()

    times = []
    with driver.session(database=resolve_kg_database()) as session:
        for _ in range(iterations):
            start = time.perf_counter()
            session.run("MATCH (n) RETURN count(n) AS cnt")
            elapsed = time.perf_counter() - start
            times.append(elapsed)

    driver.close()
    times.sort()

    print(f"  Neo4j count query x{iterations}: mean={_fmt_ms(statistics.mean(times))}")

    return {
        "operation": "graph_query",
        "iterations": iterations,
        "mean_ms": statistics.mean(times) * 1000,
        "min_ms": times[0] * 1000,
        "max_ms": times[-1] * 1000,
        "p50_ms": _percentile(times, 50) * 1000,
        "p90_ms": _percentile(times, 90) * 1000,
    }


def _run_all(
    boot_iter: int = 5,
    intent_iter: int = 10,
    throughput_sec: float = 5.0,
    workers: int = 4,
    per_worker: int = 10,
    env: Mapping[str, str] | None = None,
    runner: Any | None = None,
) -> list[dict]:
    """Run all benchmarks and return results.

    Args:
        env: Optional env mapping for Redis/graph benchmarks.
        runner: Optional already-booted ``PipelineRunner`` to reuse across the
            PipelineRunner-backed benchmarks.  When omitted (CLI/default),
            each benchmark creates and boots its own runner, preserving the
            original per-operation measurement semantics.
    """
    results = []

    print("=== PipelineRunner Benchmark Suite ===\n")

    print("1. Boot benchmark:")
    results.append(benchmark_boot(iterations=boot_iter))
    print()

    print("2. Submit intent latency:")
    results.append(benchmark_submit_intent(iterations=intent_iter, runner=runner))
    print()

    print("3. Intent throughput (5s sustained):")
    results.append(
        benchmark_intent_throughput(duration_sec=throughput_sec, runner=runner)
    )
    print()

    print("4. Parallel intents (4 workers x 10 each):")
    results.append(
        benchmark_parallel_intents(
            workers=workers, per_worker=per_worker, runner=runner
        )
    )
    print()

    print("5. Redis ping latency:")
    try:
        results.append(benchmark_redis_ping(iterations=100, env=env))
    except Exception as exc:
        print(f"  SKIPPED (Redis unavailable): {exc}")
    print()

    print("6. Neo4j query latency:")
    try:
        results.append(benchmark_graph_query(iterations=50, env=env))
    except Exception as exc:
        print(f"  SKIPPED (graph unavailable): {exc}")
    print()

    return results


def _print_summary(results: list[dict]) -> None:
    """Print formatted summary of all benchmark results."""
    print("=" * 60)
    print("SUMMARY")
    print("=" * 60)
    for r in results:
        op = r["operation"]
        print(f"\n{op}:")
        for key, value in r.items():
            if key == "operation":
                continue
            if isinstance(value, float):
                if "ms" in key:
                    print(f"  {key:<20} {value:>10.3f} ms")
                elif "sec" in key:
                    print(f"  {key:<20} {value:>10.3f} s")
                elif "rps" in key:
                    print(f"  {key:<20} {value:>10.1f} req/s")
                else:
                    print(f"  {key:<20} {value:>10.3f}")
            elif isinstance(value, (int, str)):
                print(f"  {key:<20} {value:>10}")
            else:
                # Lists/dicts (e.g. degrade-path error detail) render compactly.
                print(f"  {key:<20} {str(value)}")


def _load_dotenv() -> None:
    """Load the RUL-019 two-phase environment contract.

    Delegates to ``core.env_bootstrap.load_env_contract`` — THE canonical
    loader — instead of parsing ``.env`` here. The hand-rolled copy this
    replaced (byte-identical in four scripts) stopped after phase 1, and
    ``.env`` is the prod-BASE layer: it carries ``CXP_GRPC_HOST=localhost``
    for the host user-units, so a script that loads it and stops resolves
    PRODUCTION endpoints while believing it is on the test fleet. The
    canonical loader then applies the ``ORCHESTRATA_ENV_PROFILE`` overlay,
    which is the half that points the process at the right infrastructure,
    and honours ``declare_owned`` so it cannot reset a test worker's
    per-slot isolation.
    """
    load_env_contract()


def main(argv: list[str] | None = None) -> int:
    """CLI entry point for benchmarks."""
    _load_dotenv()
    parser = argparse.ArgumentParser(description="PipelineRunner benchmark suite")
    parser.add_argument(
        "--boot-iter",
        type=int,
        default=5,
        help="Boot benchmark iterations (default: 5)",
    )
    parser.add_argument(
        "--intent-iter",
        type=int,
        default=10,
        help="Intent submission iterations (default: 10)",
    )
    parser.add_argument(
        "--throughput-sec",
        type=float,
        default=5.0,
        help="Throughput test duration in seconds (default: 5.0)",
    )
    parser.add_argument(
        "--workers",
        type=int,
        default=4,
        help="Parallel worker threads (default: 4)",
    )
    parser.add_argument(
        "--per-worker",
        type=int,
        default=10,
        help="Intents per worker (default: 10)",
    )
    args = parser.parse_args(argv)

    results = _run_all(
        boot_iter=args.boot_iter,
        intent_iter=args.intent_iter,
        throughput_sec=args.throughput_sec,
        workers=args.workers,
        per_worker=args.per_worker,
    )
    _print_summary(results)
    return 0


if __name__ == "__main__":
    sys.exit(main())