diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..435169d --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.py[cod] + +# Local secrets / config (see README) +.env +license.jwt +*.jwt +.ruff_cache/ diff --git a/README.md b/README.md index bc2fc59..73b3fc3 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,12 @@ Each service directory is self-contained: its compose file(s), `.env.example`, example client, and `README.md` live together. Run commands from inside the service directory. +Both stacks can also run on serverless GPUs instead of hardware you manage — see +[sync on Modal](sync/README.md#deploying-on-modal-serverless-gpu) and +[streaming on Modal](streaming/README.md#deploying-on-modal-serverless-gpu). +[`bench/`](bench/) holds a load-test harness that points at either deployment +with real audio and measures the concurrency it sustains. + ## Repository layout ``` diff --git a/bench/README.md b/bench/README.md new file mode 100644 index 0000000..525837b --- /dev/null +++ b/bench/README.md @@ -0,0 +1,86 @@ +# Load-test harness + +Points at a running deployment — local compose or Modal — sends **real audio**, +verifies a transcript actually comes back, and reports latency and throughput. +`--ramp` sweeps concurrency to find the level a deployment sustains before it +degrades. + +```bash +python -m venv venv && source venv/bin/activate +pip install -r requirements.txt +``` + +## Correctness check + +One request, verifying the transcript contains an expected word (so an empty +`200` is still counted as a failure): + +```bash +# sync +python harness.py sync \ + --endpoint https://--aai-sync-u3pro-sync-api.modal.run \ + --audio ../sync/example/example_audio_file.wav + +# streaming +python harness.py streaming \ + --endpoint wss://--aai-streaming-u3pro-streaming-api.modal.run \ + --audio ../streaming/example/example_audio_file.wav \ + --speech-model universal-3-5-pro +``` + +Against a local compose stack, use `--endpoint http://localhost:8080` and +`ws://localhost:8080`. + +## Concurrency sweep + +```bash +python harness.py sync --endpoint https://... --audio a.wav --ramp 1,2,4,8,16 +python harness.py streaming --endpoint wss://... --audio a.wav \ + --ramp 1,4,16,32 --max-seconds 20 +``` + +Each level fires N requests simultaneously and reports: + +| column | meaning | +| --- | --- | +| `ok` / `fail` | requests that returned a valid transcript | +| `p50` / `p95` / `max` | per-request wall-clock latency | +| `req/s` | completed requests per second | +| `xRT` | audio seconds transcribed per wall-clock second | + +Throughput plateauing while latency keeps climbing is the saturation point: +past it, requests queue on the GPU rather than being served in parallel. + +## Options + +| flag | purpose | +| --- | --- | +| `--concurrency N` | single level (default 1) | +| `--ramp a,b,c` | sweep several levels in order | +| `--max-seconds N` | truncate the audio; keeps sweeps short | +| `--expect WORD` | substring the transcript must contain; `''` disables | +| `--speech-model` | streaming only, e.g. `universal-3-5-pro` | +| `--speed N` | streaming send rate vs realtime; `2` sends twice as fast | +| `--show-transcript` | print a sample transcript; off by default (see below) | +| `--open-timeout N` | WebSocket handshake wait, default 300s for cold starts | +| `--stop-on-failure` | end a ramp at the first level that fails | + +Exit status is non-zero if any level had a failure, so it works as a CI gate. + +Transcripts are not printed by default. They are produced from whatever audio +you submit and can contain personal data, which you generally do not want in CI +logs. Correctness is still enforced without them: `--expect` fails the run if +the expected substring is missing, and the summary reports word and character +counts. Pass `--show-transcript` when you want to eyeball the text — it is +truncated and newline-collapsed so untrusted output cannot forge log lines. + +## Interpreting Modal results + +- **The first request after idle is a cold start** and is not a latency + measurement. On sync that was ~168 s wall against ~1.8 s of server time. Warm + the endpoint first, or discard the first level. +- **A burst shorter than a cold start does not autoscale.** Modal cannot bring + up a second GPU container within a short ramp, so a sweep measures *one* + container's capacity — which is what you want when sizing a replica. +- **Streaming sessions are realtime-paced by default**, so a level takes at + least the audio's duration. Use `--max-seconds` to keep sweeps short. diff --git a/bench/harness.py b/bench/harness.py new file mode 100644 index 0000000..ea7dbb8 --- /dev/null +++ b/bench/harness.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python3 +"""Load-test harness for the self-hosted stacks, local or on Modal. + +Sends real audio at a chosen concurrency, verifies transcripts actually come +back, and reports latency and throughput. Use --ramp to sweep concurrency and +find the level a deployment sustains before it degrades. + + # correctness check, one request + python harness.py sync --endpoint https://host --audio ../sync/example/example_audio_file.wav + + # concurrency sweep + python harness.py sync --endpoint https://host --audio a.wav --ramp 1,2,4,8,16 + python harness.py streaming --endpoint wss://host --audio a.wav --ramp 1,4,16,32,48 +""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +import wave +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass, field +from typing import Callable +from urllib.parse import urlencode + +# Words expected in the bundled sample; used only as a sanity check that the +# transcript is real output rather than an empty 200. +DEFAULT_EXPECT = "assemblyai" + + +@dataclass +class Result: + ok: bool + seconds: float + detail: str = "" + text: str = "" + extra: dict = field(default_factory=dict) + + +def load_pcm16(path: str, max_seconds: float | None) -> tuple[bytes, int, float]: + """Read a 16-bit PCM WAV, optionally truncated, returning raw frames.""" + with wave.open(path, "rb") as wav: + if wav.getsampwidth() != 2 or wav.getcomptype() != "NONE": + raise SystemExit(f"{path}: must be uncompressed 16-bit PCM WAV") + rate = wav.getframerate() + frames = wav.getnframes() + if max_seconds: + frames = min(frames, int(rate * max_seconds)) + return wav.readframes(frames), rate, frames / rate + + +def wav_bytes(pcm: bytes, rate: int, channels: int = 1) -> bytes: + """Re-wrap raw PCM as a WAV container for the sync API's multipart upload.""" + import io + + buf = io.BytesIO() + with wave.open(buf, "wb") as out: + out.setnchannels(channels) + out.setsampwidth(2) + out.setframerate(rate) + out.writeframes(pcm) + return buf.getvalue() + + +# -------------------------------------------------------------------------- +# sync: one HTTP POST per request +# -------------------------------------------------------------------------- +def sync_once(endpoint: str, audio: bytes, expect: str) -> Result: + import requests + + start = time.perf_counter() + try: + resp = requests.post( + f"{endpoint.rstrip('/')}/transcribe", + files={"audio": ("audio.wav", audio, "audio/wav")}, + data={"config": json.dumps({"language_code": "en"})}, + headers={"Authorization": "harness"}, + timeout=300, + ) + except Exception as exc: # network/timeout + return Result(False, time.perf_counter() - start, detail=repr(exc)) + elapsed = time.perf_counter() - start + + if resp.status_code != 200: + return Result( + False, elapsed, detail=f"HTTP {resp.status_code}: {resp.text[:120]}" + ) + body = resp.json() + text = body.get("text", "") + if expect and expect.lower() not in text.lower(): + return Result( + False, elapsed, detail=f"transcript missing {expect!r}", text=text + ) + return Result( + True, + elapsed, + text=text, + extra={ + "server_ms": body.get("request_time_ms"), + "audio_ms": body.get("audio_duration_ms"), + "words": len(body.get("words", [])), + }, + ) + + +# -------------------------------------------------------------------------- +# streaming: one WebSocket session per request +# -------------------------------------------------------------------------- +def streaming_once( + endpoint: str, + pcm: bytes, + rate: int, + expect: str, + speech_model: str | None, + speed: float, + open_timeout: float, +) -> Result: + from websockets.sync.client import connect + + params = {"sample_rate": rate, "format_turns": "true"} + if speech_model: + params["speech_model"] = speech_model + url = f"{endpoint.rstrip('/')}?{urlencode(params)}" + + # 50 ms of audio per frame, the granularity the example client uses. + frame = int(rate * 0.05) * 2 + chunks = [pcm[i : i + frame] for i in range(0, len(pcm), frame)] + + start = time.perf_counter() + first_turn: float | None = None + turns = 0 + final_text: list[str] = [] + + try: + # Generous open timeout: a cold Modal container can take minutes to + # accept the upgrade, and the default 10s reads as a spurious failure. + with connect( + url, + additional_headers={"Authorization": "harness"}, + open_timeout=open_timeout, + max_size=None, + ) as ws: + + def writer(): + for chunk in chunks: + time.sleep(0.05 / speed) + ws.send(chunk) + ws.send('{"type": "Terminate"}') + + with ThreadPoolExecutor(max_workers=1) as pool: + write_future = pool.submit(writer) + for message in ws: + data = json.loads(message) + kind = data.get("type") + if kind == "Turn": + nonlocal_words = data.get("words") or [] + if nonlocal_words: + if first_turn is None: + first_turn = time.perf_counter() - start + turns += 1 + if data.get("end_of_turn"): + final_text.append( + " ".join(w["text"] for w in nonlocal_words) + ) + elif kind == "Termination": + break + write_future.result() + except Exception as exc: + return Result(False, time.perf_counter() - start, detail=repr(exc)) + + elapsed = time.perf_counter() - start + text = " ".join(final_text) + if expect and expect.lower() not in text.lower(): + return Result( + False, elapsed, detail=f"transcript missing {expect!r}", text=text + ) + return Result( + True, elapsed, text=text, extra={"first_turn_s": first_turn, "turns": turns} + ) + + +# -------------------------------------------------------------------------- +# driver +# -------------------------------------------------------------------------- +def run_level(work: Callable[[], Result], n: int) -> list[Result]: + """Fire n copies of `work` at once and collect every outcome.""" + with ThreadPoolExecutor(max_workers=n) as pool: + futures = [pool.submit(work) for _ in range(n)] + return [f.result() for f in futures] + + +def summarize(results: list[Result], wall: float, audio_seconds: float) -> dict: + ok = [r for r in results if r.ok] + lat = sorted(r.seconds for r in ok) + + def quantile(p: float) -> float: + if not lat: + return float("nan") + return lat[min(int(len(lat) * p), len(lat) - 1)] + + return { + "n": len(results), + "ok": len(ok), + "failed": len(results) - len(ok), + "p50": quantile(0.50), + "p95": quantile(0.95), + "max": lat[-1] if lat else float("nan"), + "wall": wall, + "rps": len(ok) / wall if wall else 0.0, + "audio_x_realtime": (len(ok) * audio_seconds / wall) if wall else 0.0, + } + + +def print_table(rows: list[tuple[int, dict]]) -> None: + print( + f"\n{'conc':>5} {'ok':>5} {'fail':>5} {'p50 s':>8} {'p95 s':>8} " + f"{'max s':>8} {'req/s':>7} {'xRT':>7}" + ) + print("-" * 60) + for conc, s in rows: + print( + f"{conc:>5} {s['ok']:>5} {s['failed']:>5} {s['p50']:>8.2f} " + f"{s['p95']:>8.2f} {s['max']:>8.2f} {s['rps']:>7.2f} " + f"{s['audio_x_realtime']:>7.1f}" + ) + + +def main() -> int: + ap = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + ap.add_argument("mode", choices=["sync", "streaming"]) + ap.add_argument( + "--endpoint", + required=True, + help="https://... for sync, wss://... for streaming", + ) + ap.add_argument("--audio", required=True, help="16-bit PCM WAV") + ap.add_argument("--concurrency", type=int, default=1) + ap.add_argument("--ramp", help="comma-separated concurrency levels, e.g. 1,4,8,16") + ap.add_argument("--max-seconds", type=float, help="truncate audio to N seconds") + ap.add_argument( + "--expect", + default=DEFAULT_EXPECT, + help="substring the transcript must contain ('' to skip)", + ) + ap.add_argument("--speech-model", help="streaming only, e.g. universal-3-5-pro") + ap.add_argument( + "--speed", type=float, default=1.0, help="streaming send rate vs realtime" + ) + ap.add_argument( + "--show-transcript", + action="store_true", + help="print a sample transcript; off by default since audio may contain personal data", + ) + ap.add_argument( + "--open-timeout", + type=float, + default=300.0, + help="seconds to wait for the WebSocket handshake (cold starts are slow)", + ) + ap.add_argument( + "--stop-on-failure", + action="store_true", + help="end a ramp at the first level with failures", + ) + args = ap.parse_args() + + pcm, rate, seconds = load_pcm16(args.audio, args.max_seconds) + print(f"audio: {args.audio} | {seconds:.1f}s @ {rate} Hz | mode={args.mode}") + print(f"endpoint: {args.endpoint}") + + if args.mode == "sync": + payload = wav_bytes(pcm, rate) + work = lambda: sync_once(args.endpoint, payload, args.expect) # noqa: E731 + else: + work = lambda: streaming_once( # noqa: E731 + args.endpoint, + pcm, + rate, + args.expect, + args.speech_model, + args.speed, + args.open_timeout, + ) + + levels = [int(x) for x in args.ramp.split(",")] if args.ramp else [args.concurrency] + + rows = [] + sample_shown = False + for conc in levels: + started = time.perf_counter() + results = run_level(work, conc) + wall = time.perf_counter() - started + stats = summarize(results, wall, seconds) + rows.append((conc, stats)) + + if not sample_shown: + first_ok = next((r for r in results if r.ok), None) + if first_ok: + # Transcripts are user audio and can carry personal data, so the + # default output describes the result without reproducing it. + # Correctness is already asserted by --expect; this is only a + # human sanity check, so it is opt-in. + if args.show_transcript: + # Collapse newlines: transcript text is untrusted input and + # must not be able to forge extra log lines. + sample = " ".join(first_ok.text.split())[:160] + print(f"\ntranscript: {sample}...") + else: + print( + f"\ntranscript: {len(first_ok.text)} chars " + f"(hidden; pass --show-transcript to print it)" + ) + if first_ok.extra: + print(f"detail: {first_ok.extra}") + sample_shown = True + + print( + f"[conc={conc}] ok={stats['ok']}/{stats['n']} " + f"p50={stats['p50']:.2f}s p95={stats['p95']:.2f}s " + f"{stats['audio_x_realtime']:.1f}x realtime" + ) + for r in results: + if not r.ok: + print(f" FAIL: {r.detail[:160]}") + if stats["failed"] and args.stop_on_failure: + print(f"\nstopping: first failures at concurrency {conc}") + break + + if len(rows) > 1: + print_table(rows) + clean = [c for c, s in rows if s["failed"] == 0] + if clean: + print(f"\nhighest fully-successful concurrency tested: {max(clean)}") + else: + print("\nno concurrency level completed without failures") + + return 0 if all(s["failed"] == 0 for _, s in rows) else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bench/requirements.txt b/bench/requirements.txt new file mode 100644 index 0000000..435d5e3 --- /dev/null +++ b/bench/requirements.txt @@ -0,0 +1,2 @@ +requests~=2.32 +websockets~=15.0 diff --git a/streaming/README.md b/streaming/README.md index cbd0722..3f8f137 100644 --- a/streaming/README.md +++ b/streaming/README.md @@ -215,6 +215,148 @@ docker compose -f docker-compose.english-multilang.yml restart streaming-asr-mul docker compose -f docker-compose.universal-3-5-pro.yml restart streaming-asr-universal-3-5-pro ``` +## Deploying on Modal (serverless GPU) + +`modal_app.py` runs this stack on [Modal](https://modal.com). Compose's four +services become three pieces: + +| Compose service | On Modal | Hardware | +|---|---|---| +| `streaming-api` | `streaming_api` function | CPU | +| `license-and-usage-proxy` | `license_proxy` function | CPU | +| `streaming-asr-universal-3-5-pro` | a **Sandbox** | L40S GPU | +| `streaming-asr-lb` (nginx) | dropped | — | + +nginx only routes `X-Model-Version` across several ASR backends and balances +replicas. With one model there is nothing to route. + +The ASR runs in a **Sandbox rather than a Function** because a Modal tunnel's +lifetime is bound to the function *call*: a `@modal.web_server` body returns as +soon as it has started its server, Modal tears down the call's tunnel, and the +port stops answering while the container stays up. A Sandbox's tunnel lives as +long as the Sandbox does. + +### Prerequisites + +Same Modal secrets as the [sync stack](../sync/README.md#store-credentials-as-modal-secrets) +— `aai-ecr` and `aai-license`. Create them once; both stacks share them. + +### Deploy + +```bash +modal run modal_app.py::start_asr # boot the GPU backend (once) +modal deploy modal_app.py # the API + license proxy +``` + +`start_asr` creates the Sandbox, publishes its gRPC address into a +`modal.Dict`, and prints it. The model needs a few minutes to warm; check it +with `grpc_health_probe`, which ships in the image: + +```bash +python -c " +import modal +d = modal.Dict.from_name('aai-streaming-asr-addr') +sb = modal.Sandbox.from_id(d['sandbox_id']) +p = sb.exec('grpc_health_probe', '-addr=:50051'); p.wait() +print(p.stderr.read())" # status: SERVING +``` + +Tear the backend down when you are done — **it holds a GPU for as long as it +runs**: + +```bash +modal run modal_app.py::stop_asr +``` + +### Verify + +```bash +curl -fsS https://--aai-streaming-u3pro-streaming-api.modal.run/v3/ws/health +curl -fsS https://--aai-streaming-u3pro-license-proxy.modal.run/v1/status +``` + +Then stream, using the [bundled example](#running-the-streaming-example) with +`wss://` in place of `ws://localhost:8080`: + +```bash +python example_with_prerecorded_audio_file.py \ + --audio-file example_audio_file.wav \ + --endpoint wss://--aai-streaming-u3pro-streaming-api.modal.run \ + --speech-model universal-3-5-pro +``` + +Or use the [load-test harness](../bench/README.md), which checks correctness +and concurrency in one step: + +```bash +cd ../bench && pip install -r requirements.txt +python harness.py streaming \ + --endpoint wss://--aai-streaming-u3pro-streaming-api.modal.run \ + --audio ../streaming/example/example_audio_file.wav \ + --speech-model universal-3-5-pro --max-seconds 20 +``` + +### Restarting the ASR invalidates the API + +`streaming_api` reads the ASR address from the `modal.Dict` **once, at container +startup**, and passes it to the API process as `AAI_ASR_ENDPOINT`. A new Sandbox +gets a new address, so warm API containers keep dialing the old one and every +session fails with `3005 Session Cancelled` and, in the logs, +`Bad connection. Missing expected server metadata keys`. + +After any `start_asr`, force fresh API containers: + +```bash +modal app stop aai-streaming-u3pro -y && modal deploy modal_app.py +``` + +The startup log line confirms which address a container picked up: + +``` +[startup] ASR=r446.modal.host:39655 proxy=https://... +``` + +### Tear down + +The Sandbox holds an L40S for as long as it runs and does **not** scale to zero, +so tear it down explicitly when you are finished: + +```bash +modal run modal_app.py::stop_asr # releases the GPU +modal app stop aai-streaming-u3pro # the API + license proxy +modal app stop aai-streaming-asr # the Sandbox's owning app +``` + +`stop_asr` also clears the address out of the `modal.Dict`, so a later +`start_asr` starts clean. + +### Security + +`start_asr` exposes the ASR's gRPC port with `unencrypted_ports`, so it is a +**plaintext TCP socket on the public internet carrying audio in the clear** — +where compose keeps that hop on a private bridge network. This is acceptable for +testing only. Before real traffic, either move the hop onto TLS +(`encrypted_ports`, untested here) or run the API and ASR in one container so +the hop stays on localhost. Note also that the API's own `.modal.run` URL is +public and the service does not validate credentials beyond requiring a +non-empty `Authorization` header. + +### Measured behaviour (single L40S) + +From `bench/harness.py` against 15 s clips, one Sandbox: + +| concurrent sessions | outcome | +|---|---| +| 4 – 40 | all succeeded, p50 steady at 18–25 s | +| 64 | 7/64 failed | +| 96 | 3/96 failed, 44× realtime aggregate | + +Roughly **40 concurrent realtime streams** per L40S with headroom. The failures +at 64 were connection-level (`ConnectionClosedOK`, one bad HTTP response), +consistent with the CPU API function still autoscaling rather than the GPU +saturating — throughput was still climbing at 96, so the ASR itself was not the +limit. + ## Production deployment recommendations See the [top-level README](../README.md#production-recommendations-license-and-usage-proxy) diff --git a/streaming/modal_app.py b/streaming/modal_app.py new file mode 100644 index 0000000..8181931 --- /dev/null +++ b/streaming/modal_app.py @@ -0,0 +1,197 @@ +"""Run the self-hosted streaming (WebSocket realtime) stack on Modal. + +Mirrors docker-compose.universal-3-5-pro.yml. Compose's four services collapse +into three Modal functions: + + streaming_api (CPU) -> WebSocket front door, the public entrypoint + license_proxy (CPU) -> license-and-usage-proxy + ASR (L40S) -> a Sandbox, not a function (see below) + +nginx (streaming-asr-lb) is dropped: it exists only to route X-Model-Version +across several ASR backends and to load-balance replicas. With one model, +Modal's autoscaler covers the second job and the first is unnecessary. + +Compose puts the ASR on a private bridge network. Modal has no inter-container +network, so the ASR's gRPC port is published through a tunnel and its address +handed to the API through a modal.Dict. + +The ASR runs in a Sandbox rather than a Function because a tunnel's lifetime is +bound to the function *call*: a @modal.web_server body returns as soon as it has +started its server, Modal tears the call's tunnel down, and the port stops +answering while the container stays up. A Sandbox's tunnel lives as long as the +Sandbox. + +Deploy: + modal run modal_app.py::start_asr # once; boots the GPU backend + modal deploy modal_app.py # the API + license proxy + modal run modal_app.py::stop_asr # tears the backend down +""" + +import os +import subprocess + +import modal + +REGISTRY = "344839248844.dkr.ecr.us-west-2.amazonaws.com" +TAG = "release-v1.0.0" + +# Image ENTRYPOINTs. Modal prepends an image's ENTRYPOINT to its own runtime +# command, so every image clears it with .entrypoint([]) and each server is +# launched explicitly below. +API_BIN = "/opt/assemblyai/engineering/projects/realtime/api_v2/bin" +ASR_BIN = "/opt/assemblyai/engineering/projects/realtime/asr_u3pro/self_hosted_bin" +PROXY_BIN = "/opt/assemblyai/engineering/projects/realtime/license_and_usage_proxy/bin" + +LICENSE_PATH = "/tmp/aai_license.jwt" +ASR_GRPC_PORT = 50051 + +ecr_secret = modal.Secret.from_name("aai-ecr") +license_secret = modal.Secret.from_name("aai-license") + +# Publishes the ASR's tunnel address to streaming_api; Modal gives containers no +# way to address each other directly. +asr_registry = modal.Dict.from_name("aai-streaming-asr-addr", create_if_missing=True) + +app = modal.App("aai-streaming-u3pro") + + +def _vendor_image(repo: str, add_python: str | None = None) -> modal.Image: + """Build a Modal-compatible image from one of the AssemblyAI ECR images. + + Every vendor image needs the same three adjustments, learned from the sync + stack: clear the ENTRYPOINT, make sure Modal can find an interpreter, and + install the Modal client into that interpreter (Modal's runtime-mounted + client dependencies do not land on these images' sys.path). + """ + image = modal.Image.from_aws_ecr( + f"{REGISTRY}/{repo}:{TAG}", secret=ecr_secret, add_python=add_python + ).entrypoint([]) + if add_python: + # The standalone interpreter ships pip. + return image.pip_install(f"modal=={modal.__version__}") + # Wolfi images expose python3 but omit pip; bootstrap it first. + return image.run_commands( + "/usr/bin/python3 -m ensurepip --default-pip", + f"/usr/bin/python3 -m pip install --no-cache-dir --break-system-packages " + f"modal=={modal.__version__}", + ) + + +# asr and api keep their interpreters inside Bazel trees (/opt/python), where +# Modal cannot find them, so both get a standalone one injected into the unused +# /usr/local. Only the proxy image puts python3 on PATH itself. +asr_image = _vendor_image( + "self-hosted-streaming-asr-universal-3-5-pro", add_python="3.12" +).env( + { + "SERVER_PORT": str(ASR_GRPC_PORT), + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "MAX_OPEN_STREAMS": "32", + "VLLM_USE_FLASHINFER_SAMPLER": "0", + } +) +api_image = _vendor_image("self-hosted-streaming-api", add_python="3.12") +proxy_image = _vendor_image("self-hosted-streaming-license-and-usage-proxy") + + +def _write_license() -> None: + """Materialize the license from its secret; Modal has no bind mounts.""" + with open(LICENSE_PATH, "w") as fh: + fh.write(os.environ["AAI_LICENSE_JWT"].strip()) + + +@app.function(image=proxy_image, secrets=[license_secret], timeout=3600) +@modal.web_server(8080, startup_timeout=180) +def license_proxy(): + _write_license() + subprocess.Popen( + [PROXY_BIN], + env={ + **os.environ, + "HTTP_PORT": "8080", + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "LICENSE_FILE_PATH": LICENSE_PATH, + }, + ) + + +@app.local_entrypoint() +def start_asr(): + """Boot the ASR backend in a Sandbox and publish its gRPC address.""" + existing = asr_registry.get("sandbox_id") + if existing: + # A registration only counts if the sandbox is still alive; a stale + # entry from a crashed or terminated sandbox must not block a restart. + if modal.Sandbox.from_id(existing).poll() is None: + print(f"sandbox {existing} is already running; run stop_asr first") + return + print(f"clearing dead sandbox {existing}") + + # Attach to a looked-up (persistent) app, not this ephemeral `modal run` + # app: sandboxes are torn down when their owning app stops. + sandbox_app = modal.App.lookup("aai-streaming-asr", create_if_missing=True) + sandbox = modal.Sandbox.create( + ASR_BIN, + app=sandbox_app, + image=asr_image, + gpu="L40S", + timeout=24 * 60 * 60, + # Raw TCP, matching compose's AAI_USE_SECURE_CHANNEL_TO_ASR_SERVICE=False. + # Verified end to end: a plain grpc.insecure_channel to this address + # completes a health check through the tunnel, trailers included. The + # TLS/h2 tunnel (encrypted_ports) was not re-tested after an unrelated + # stale-address bug was fixed, so it may also work. + # + # SECURITY: this port is on the public internet and carries audio in the + # clear, where compose keeps the hop on a private bridge network. See + # "Security" in the README before running real traffic. + unencrypted_ports=[ASR_GRPC_PORT], + ) + host, port = sandbox.tunnels()[ASR_GRPC_PORT].tcp_socket + asr_registry["address"] = f"{host}:{port}" + asr_registry["sandbox_id"] = sandbox.object_id + print(f"ASR sandbox {sandbox.object_id} -> {host}:{port}") + print("the model takes a few minutes to warm before it accepts streams") + + +@app.local_entrypoint() +def stop_asr(): + """Terminate the ASR sandbox and clear its registration.""" + sandbox_id = asr_registry.get("sandbox_id") + if not sandbox_id: + print("no sandbox registered") + return + modal.Sandbox.from_id(sandbox_id).terminate() + del asr_registry["sandbox_id"] + del asr_registry["address"] + print(f"terminated {sandbox_id}") + + +@app.function(image=api_image, secrets=[license_secret], timeout=3600) +@modal.web_server(8080, startup_timeout=600) +def streaming_api(): + address = asr_registry.get("address") + if not address: + raise RuntimeError( + "no ASR address registered; run `modal run modal_app.py::start_asr`" + ) + + proxy_url = modal.Function.from_name( + "aai-streaming-u3pro", "license_proxy" + ).get_web_url() + print(f"[startup] ASR={address} proxy={proxy_url}", flush=True) + + subprocess.Popen( + [API_BIN], + env={ + **os.environ, + "AAI_WSS_PORT": "8080", + "AAI_LOG_LEVEL": "INFO", + "AAI_USE_STRUCTURED_LOGGING": "False", + "AAI_ASR_ENDPOINT": address, + "AAI_USE_SECURE_CHANNEL_TO_ASR_SERVICE": "False", + "AAI_LICENSE_AND_USAGE_PROXY_ENDPOINT": proxy_url, + }, + ) diff --git a/sync/README.md b/sync/README.md index e5d9fca..34903e5 100644 --- a/sync/README.md +++ b/sync/README.md @@ -121,6 +121,157 @@ python transcribe_file.py # uses the bundled example_audio_fi python transcribe_file.py path/to/audio.wav # or your own 16-bit PCM WAV ``` +## Deploying on Modal (serverless GPU) + +`modal_app.py` runs this same stack on [Modal](https://modal.com) instead of a +GPU box you manage. Compose's two services become two Modal functions, because +Modal runs one image per container and has no sidecars: + +| Compose service | Modal function | Hardware | +|---|---|---| +| `sync-api` | `sync_api` | L40S GPU | +| `license-and-usage-proxy` | `license_proxy` | CPU | + +`sync_api` resolves the proxy's Modal URL at startup and passes it as +`LICENSE_AND_USAGE_PROXY_ENDPOINT`, replacing the compose bridge network. + +### Prerequisites + +```bash +pip install modal && modal setup # authenticate the Modal CLI +``` + +### Store credentials as Modal secrets + +Modal has no bind mounts, so the license travels as a secret and is written to +disk at container startup. Both secrets are read at image-build and run time: + +```bash +# ECR pull credentials. Modal re-mints the 12-hour registry token from these. +modal secret create aai-ecr \ + AWS_ACCESS_KEY_ID="$(aws configure get aws_access_key_id)" \ + AWS_SECRET_ACCESS_KEY="$(aws configure get aws_secret_access_key)" \ + AWS_REGION=us-west-2 + +# The license itself, kept out of any image layer. +modal secret create aai-license AAI_LICENSE_JWT="$(cat license.jwt)" +``` + +If you authenticate with `aws login` or SSO rather than static keys, export the +temporary session credentials instead — note they expire, so image *rebuilds* +need a refresh (deploys of an already-built image do not): + +```bash +eval "$(aws configure export-credentials --format env)" +modal secret create aai-ecr \ + AWS_ACCESS_KEY_ID="$AWS_ACCESS_KEY_ID" \ + AWS_SECRET_ACCESS_KEY="$AWS_SECRET_ACCESS_KEY" \ + AWS_SESSION_TOKEN="$AWS_SESSION_TOKEN" \ + AWS_REGION=us-west-2 --force +``` + +Usage-based licenses also need `USAGE_TRACKING_API_KEY` added to `aai-license` +and passed through in `license_proxy`. Flat-billed licenses need nothing extra. + +### Deploy + +```bash +modal deploy modal_app.py +``` + +The first deploy pulls and converts the ~13.5 GB sync image and takes several +minutes; later deploys reuse the cached image and take seconds. Two public URLs +are printed: + +``` +https://--aai-sync-u3pro-sync-api.modal.run +https://--aai-sync-u3pro-license-proxy.modal.run +``` + +### Verify + +```bash +curl -fsS https://--aai-sync-u3pro-license-proxy.modal.run/v1/status +# {"state":"Connected", ...} + +curl -sS -o /dev/null -w '%{http_code}\n' \ + https://--aai-sync-u3pro-sync-api.modal.run/readyz +# 303 while the container is cold, 200 once the model is warm +``` + +The [load-test harness](../bench/README.md) checks correctness and concurrency +in one step: + +```bash +cd ../bench && pip install -r requirements.txt +python harness.py sync \ + --endpoint https://--aai-sync-u3pro-sync-api.modal.run \ + --audio ../sync/example/example_audio_file.wav +``` + +Or transcribe exactly as documented in [Transcribe](#transcribe), swapping +`http://localhost:8080` for the `sync-api` URL: + +```bash +curl -F 'audio=@example/example_audio_file.wav;type=audio/wav' \ + -F 'config={"language_code":"en"};type=application/json' \ + -H 'Authorization: any value works' \ + https://--aai-sync-u3pro-sync-api.modal.run/transcribe +``` + +### Tear down + +Both functions scale to zero on their own, so an idle deployment holds no GPU. +To remove it entirely: + +```bash +modal app stop aai-sync-u3pro +``` + +### Measured behaviour (single L40S) + +From `bench/harness.py` against the bundled 60 s sample, one warm container: + +| concurrent requests | ok | p50 | p95 | audio x realtime | +|---:|---:|---:|---:|---:| +| 1 | 1/1 | 2.60 s | 2.60 s | 22.6 | +| 2 | 2/2 | 4.41 s | 4.41 s | 27.2 | +| 4 | 4/4 | 6.15 s | 7.88 s | 30.5 | +| 8 | 8/8 | 9.43 s | 14.57 s | 33.0 | +| 16 | 16/16 | 16.73 s | 28.85 s | 33.3 | + +Server-side inference was ~1.8–2.1 s for 60 s of audio. Throughput plateaus +near **33x realtime at concurrency 8**; past that, latency grows roughly +linearly while throughput does not, which is the point where requests queue on +the GPU. Size a replica against that knee. + +Note the first request after idle is a cold start, not a latency measurement — +one measured 168 s wall against 1.8 s of server time. A burst shorter than a +cold start also will not autoscale, so a sweep measures a single container. + +### Modal-specific notes + +- **Cold starts.** A cold `sync_api` container spends roughly 2–4 minutes + pulling the image, loading weights, and capturing CUDA graphs; Modal returns + `303` until the server binds. `scaledown_window=300` keeps a warm container + for 5 minutes after the last request. For latency-sensitive traffic set + `min_containers=1` on `sync_api` — that holds a GPU and bills accordingly. +- **`.entrypoint([])` is required.** Modal prepends an image's `ENTRYPOINT` to + its own runtime command. Left in place, the vendor binary swallows Modal's + arguments, starts with default environment (so `LICENSE_FILE_PATH` reverts to + the compose path and the proxy dies with `License file not found`), and the + Python in `modal_app.py` never executes. +- **Interpreter handling differs per image.** The proxy (Wolfi) already exposes + `python3`, so it must *not* get `add_python`. The sync image's interpreter is + hermetic inside Bazel runfiles and invisible to Modal, so it needs + `add_python="3.12"`. Both images additionally pip-install the Modal client, + because Modal's runtime-mounted client dependencies do not land on these + interpreters' `sys.path` (symptom: `ModuleNotFoundError: grpclib`). +- **Authentication.** As on any other host, the service does not validate + credentials but rejects an empty `Authorization` header with `401`. A + `.modal.run` URL is public — put auth in front of it, or deploy the endpoint + with Modal proxy auth, before exposing it beyond testing. + ## Production deployment recommendations See the [top-level README](../README.md#production-recommendations-license-and-usage-proxy) diff --git a/sync/modal_app.py b/sync/modal_app.py new file mode 100644 index 0000000..cc4f161 --- /dev/null +++ b/sync/modal_app.py @@ -0,0 +1,138 @@ +"""Run the self-hosted sync (full-file HTTP) stack on Modal. + +Mirrors docker-compose.universal-3-5-pro.yml. Compose's two services become two +Modal functions, since Modal runs one image per container with no sidecars: + + license_proxy (CPU) -> license-and-usage-proxy + sync_api (L40S) -> sync-api, reaches the proxy over its Modal URL + +Deploy: modal deploy modal_app.py +""" + +import os +import subprocess + +import modal + +REGISTRY = "344839248844.dkr.ecr.us-west-2.amazonaws.com" +TAG = "release-v1.0.0" + +# ENTRYPOINTs of the two images, started explicitly in each function body. +# Modal prepends an image's ENTRYPOINT to its own runtime command, so both +# images must clear it with .entrypoint([]) — otherwise the vendor binary +# consumes Modal's arguments, starts with default env, and our code never runs. +PROXY_BIN = "/opt/assemblyai/engineering/projects/realtime/license_and_usage_proxy/bin" +SYNC_BIN = ( + "/opt/assemblyai/engineering/projects/realtime/asr_sync_u3pro/self_hosted_bin" +) + +LICENSE_PATH = "/tmp/aai_license.jwt" + +ecr_secret = modal.Secret.from_name("aai-ecr") +license_secret = modal.Secret.from_name("aai-license") + +app = modal.App("aai-sync-u3pro") + +# The proxy image (Wolfi) already has python3 on PATH, so Modal detects it and +# add_python would only shadow a working interpreter. +proxy_image = ( + modal.Image.from_aws_ecr( + f"{REGISTRY}/self-hosted-streaming-license-and-usage-proxy:{TAG}", + secret=ecr_secret, + ) + .entrypoint([]) + # Wolfi base: /usr/bin/python3 is 3.13 but ships no pip, and Modal's + # runtime-mounted client deps do not land on its sys.path. Installing the + # client into the image's own interpreter makes resolution deterministic. + .run_commands( + "/usr/bin/python3 -m ensurepip --default-pip", + f"/usr/bin/python3 -m pip install --no-cache-dir --break-system-packages " + f"modal=={modal.__version__}", + ) +) + +# The sync image's interpreter is hermetic (bundled inside Bazel runfiles), so +# nothing named python3 is on PATH and Modal cannot detect a version. Injecting +# a standalone interpreter is safe here precisely because /usr/local is unused, +# so it shadows nothing the ASR binary depends on. +sync_image = ( + modal.Image.from_aws_ecr( + f"{REGISTRY}/self-hosted-sync-asr-u3-pro:{TAG}", + secret=ecr_secret, + add_python="3.12", + ) + .entrypoint([]) + # As with the proxy, pin Modal's client into the interpreter it will + # actually launch; the runtime-mounted deps do not resolve in these images. + .pip_install(f"modal=={modal.__version__}") +) + + +def _write_license() -> None: + """Materialize the license from the Modal secret onto disk. + + Compose bind-mounts license.jwt into the container; Modal has no bind + mounts, so the JWT travels as a secret and is written at startup. + """ + token = os.environ["AAI_LICENSE_JWT"].strip() + with open(LICENSE_PATH, "w") as fh: + fh.write(token) + print( + f"[startup] wrote {LICENSE_PATH} ({os.path.getsize(LICENSE_PATH)} bytes)", + flush=True, + ) + + +@app.function( + image=proxy_image, + secrets=[license_secret], + timeout=3600, +) +@modal.web_server(8080, startup_timeout=180) +def license_proxy(): + _write_license() + print(f"[startup] launching {PROXY_BIN}", flush=True) + subprocess.Popen( + [PROXY_BIN], + env={ + **os.environ, + "HTTP_PORT": "8080", + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "LICENSE_FILE_PATH": LICENSE_PATH, + # Licence is flat-billed, so no USAGE_TRACKING_API_KEY is required. + }, + ) + + +@app.function( + image=sync_image, + gpu="L40S", + secrets=[license_secret], + timeout=3600, + scaledown_window=300, +) +@modal.web_server(8080, startup_timeout=900) # weights load + engine warmup +def sync_api(): + proxy_url = ( + os.environ.get("PROXY_ENDPOINT") + or modal.Function.from_name("aai-sync-u3pro", "license_proxy").get_web_url() + ) + + subprocess.Popen( + [SYNC_BIN], + env={ + **os.environ, + "HTTP_PORT": "8080", + "AAI_ENV": "production", + "LOGGING_LEVEL": "INFO", + "USE_STRUCTURED_LOGGING": "False", + "GPU_MONITORING_ENABLED": "False", + "LICENSE_AND_USAGE_PROXY_ENDPOINT": proxy_url, + "MAX_AUDIO_DURATION_MS": "120000", + "MIN_AUDIO_DURATION_MS": "80", + "MAX_REQUEST_BYTES": "41943040", + "INFERENCE_TIMEOUT_SECONDS": "30", + "VLLM_USE_FLASHINFER_SAMPLER": "0", + }, + )