From 3bf9b0c926f42d5c1081057866ef549e0752aa0e Mon Sep 17 00:00:00 2001 From: "guthrie@vast.ai" Date: Wed, 9 Sep 2026 17:57:39 -0700 Subject: [PATCH] docs(examples): add webhook automation tutorials for instance downtime Two new examples under Notifications showing webhooks as an automation hook rather than a chat feed: - Migrate an Instance Before Scheduled Maintenance: react to client:upcoming_downtime, look up the maintenance window, copy state off the instance, rent a replacement on a different machine, destroy the original after instance_started. - Route Around Offline Instances with a Simple Load Balancer: react to client:instance_offline (plural payload), drain, blacklist the failed machine, rent a replacement elsewhere, re-admit only after the client's own readiness probe. Shared setup (receiver, webhook creation, local event simulator) lives in snippets/notifications/webhook-automation-setup.mdx and is imported by both pages. Co-Authored-By: Claude Fable 5.1 --- docs.json | 4 +- .../notifications/maintenance-migration.mdx | 205 +++++++++++ .../offline-aware-load-balancer.mdx | 317 ++++++++++++++++++ .../webhook-automation-setup.mdx | 245 ++++++++++++++ 4 files changed, 770 insertions(+), 1 deletion(-) create mode 100644 examples/notifications/maintenance-migration.mdx create mode 100644 examples/notifications/offline-aware-load-balancer.mdx create mode 100644 snippets/notifications/webhook-automation-setup.mdx diff --git a/docs.json b/docs.json index 8e13e72b..dfd80145 100644 --- a/docs.json +++ b/docs.json @@ -706,7 +706,9 @@ "icon": "bell", "pages": [ "examples/notifications/slack-webhook", - "examples/notifications/google-chat-webhook" + "examples/notifications/google-chat-webhook", + "examples/notifications/maintenance-migration", + "examples/notifications/offline-aware-load-balancer" ] }, { diff --git a/examples/notifications/maintenance-migration.mdx b/examples/notifications/maintenance-migration.mdx new file mode 100644 index 00000000..4e7cd84b --- /dev/null +++ b/examples/notifications/maintenance-migration.mdx @@ -0,0 +1,205 @@ +--- +title: "Migrate an Instance Before Scheduled Maintenance" +slug: "vast-notifications-maintenance-migration" +createdAt: "Wed Sep 09 2026 00:00:00 GMT+0000 (Coordinated Universal Time)" +updatedAt: "Wed Sep 09 2026 00:00:00 GMT+0000 (Coordinated Universal Time)" +--- + +import WebhookAutomationSetup from '/snippets/notifications/webhook-automation-setup.mdx'; + +Use this example to move a job off an instance before its host machine goes down for scheduled maintenance. Vast.ai sends an **Upcoming instance downtime** notification with the affected instance and machine; the handler below looks up the window, copies your work off the instance, starts a replacement on a different machine, and destroys the original once the replacement is running. + +This page builds on the receiver pattern from [Send Notifications to Slack](/examples/notifications/slack-webhook). For the companion example that reacts to unplanned machine loss, see [Route Around Offline Instances](/examples/notifications/offline-aware-load-balancer). + + + +## What Happens Without This + +A host schedules maintenance on the machine your instance runs on. Vast.ai sends you an **Upcoming instance downtime** notification with the window. If nothing acts on it, your instance goes down when the window opens, and whatever was on its disk is unavailable until the host brings the machine back. + +This tutorial catches that notice in code, copies your checkpoint or working directory off the instance, starts a replacement on a different machine, and destroys the original once the replacement is running. Maintenance becomes a move you schedule instead of an outage you discover. + +## How the Events Fit Together + +| Event | `data` fields | Your action | +| --- | --- | --- | +| `upcoming_downtime` | `instance_id`, `machine_id`, `maintenance_id` | Look up the window, then migrate before it opens | +| `instance_started` | `instance_id`, `machine_id` | Replacement is running: destroy the original | + + +Today the `upcoming_downtime` payload identifies the maintenance but does not include its start time or duration. You look those up with one API call, shown below. The payload is being extended to carry the window directly; when it does, the lookup becomes a fallback rather than a required step. + + +## Look Up the Maintenance Window + +Query maintenance windows by machine ID and match on the `maintenance_id` from the payload. A machine can have more than one window scheduled. + +```bash +curl -sS -X POST "https://console.vast.ai/api/v0/machines/maintenances" \ + -H "Authorization: Bearer $VAST_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{"machine_ids": [447823]}' +``` + +```json +[ + { + "id": 88213, + "machine_id": 447823, + "start_time": 1757649600, + "duration_hours": 6, + "maintenance_category": "software" + } +] +``` + +`start_time` is a Unix timestamp in seconds. The response has no end time; compute it as `start_time + 3600 * duration_hours`. + +## Write the Migration Handler + +Create `migrate.py`. It registers two handlers with the shared receiver and owns one piece of state: which new instance is replacing which old one. + +```python +#!/usr/bin/env python3 +import os +import threading +import time + +import requests +from vastai import VastAI + +from receiver import on, serve + +API_KEY = os.environ["VAST_API_KEY"] +vast = VastAI(api_key=API_KEY) +API = "https://console.vast.ai/api/v0" +AUTH = {"Authorization": f"Bearer {API_KEY}"} + +MIGRATE_LEAD_SECONDS = 2 * 3600 # start the move this long before the window opens +CHECKPOINT_DIR = "/workspace/checkpoints" +CLOUD_CONNECTION_ID = os.environ.get("VAST_CLOUD_CONNECTION_ID") # from Cloud Sync in the console + +REPLACING: dict[int, int] = {} # new_instance_id -> old_instance_id + + +def maintenance_window(machine_id: int, maintenance_id: int) -> dict | None: + r = requests.post(f"{API}/machines/maintenances", headers=AUTH, + json={"machine_ids": [machine_id]}, timeout=10) + r.raise_for_status() + for w in r.json(): + if w["id"] == maintenance_id: + w["end_time"] = w["start_time"] + 3600 * w["duration_hours"] + return w + return None + + +def save_checkpoint(instance_id: int) -> None: + """Get state off the instance. Replace with your own rsync/s3 sync if you already have one.""" + vast.cloud_copy( + src=CHECKPOINT_DIR, + dst=f"/vast-checkpoints/{instance_id}", + instance=str(instance_id), + connection=CLOUD_CONNECTION_ID, + transfer="Instance to Cloud", + ) + + +def place_replacement(old: dict, exclude_machines: list[int]) -> int: + """Rent the same GPU and image on a different machine. Returns the new instance ID.""" + offers = vast.search_offers( + query=f"gpu_name={old['gpu_name'].replace(' ', '_')} num_gpus={old['num_gpus']} " + f"rentable=true dph<={old['dph_total'] * 1.25:.3f}", + order="dph", + limit=20, + ) + offer = next(o for o in offers if o["machine_id"] not in exclude_machines) + result = vast.create_instance( + id=offer["id"], + image=old["image_uuid"], + disk=old["disk_space"], + label=old.get("label"), + env=old.get("extra_env"), + onstart_cmd=os.environ.get("RESUME_CMD"), # your job's "resume from checkpoint" command + ) + return result["new_contract"] + + +def migrate(instance_id: int, exclude_machines: list[int]) -> None: + old = vast.show_instance(id=instance_id) + print(f"migrating instance {instance_id} off machine {old['machine_id']}", flush=True) + save_checkpoint(instance_id) + new_id = place_replacement(old, exclude_machines) + REPLACING[new_id] = instance_id + print(f"replacement instance {new_id} created; waiting for it to start", flush=True) + + +@on("upcoming_downtime") +def on_upcoming_downtime(evt: dict) -> None: + d = evt.get("data") or {} + if evt.get("data_truncated") or "instance_id" not in d: + print("payload incomplete; check instances manually", flush=True) + return + window = maintenance_window(d["machine_id"], d["maintenance_id"]) + if window is None: + print(f"maintenance {d['maintenance_id']} no longer scheduled; nothing to do", flush=True) + return + + starts_in = window["start_time"] - time.time() + delay = max(0, starts_in - MIGRATE_LEAD_SECONDS) + print(f"maintenance on machine {d['machine_id']} starts in {starts_in/3600:.1f}h " + f"for {window['duration_hours']}h; migrating in {delay/3600:.1f}h", flush=True) + threading.Timer(delay, migrate, args=(d["instance_id"], [d["machine_id"]])).start() + + +@on("instance_started") +def on_instance_started(evt: dict) -> None: + new_id = (evt.get("data") or {}).get("instance_id") + old_id = REPLACING.pop(new_id, None) + if old_id is None: + return # not one of ours + print(f"instance {new_id} running; destroying original {old_id}", flush=True) + vast.destroy_instance(id=old_id) + + +if __name__ == "__main__": + serve() +``` + +A few points about the design: + +- **The move starts on your schedule, not the host's.** `MIGRATE_LEAD_SECONDS` decides how long before the window the migration begins. Set it longer than your checkpoint copy plus the replacement's image pull. +- **The replacement excludes the machine under maintenance.** `place_replacement` skips any offer on `machine_id`. Without that, the cheapest offer is often another slot on the same machine. +- **The original is destroyed only after the replacement reports `instance_started`.** That event fires once, when a new rental first runs. Until then you pay for both instances, which is the price of not losing the job. +- **The checkpoint copy is the part only you can own.** Vast.ai cannot know when a file is fully written. If your job already pushes checkpoints to your own storage on a schedule, replace `save_checkpoint` with a no-op and point `RESUME_CMD` at that location. + +## Run It + +```bash +export VAST_API_KEY VAST_WEBHOOK_SECRET +export VAST_CLOUD_CONNECTION_ID= +export RESUME_CMD='cd /workspace && python train.py --resume-from /workspace/checkpoints/latest' +python3 migrate.py +``` + +Then simulate a downtime notice against a real instance you own. Set `MIGRATE_LEAD_SECONDS` high, or the event will schedule the move for later rather than now: + +```bash +python3 simulate.py upcoming_downtime \ + '{"instance_id": , "machine_id": , "maintenance_id": 0}' +``` + +Because `maintenance_id` `0` does not exist, the handler logs "no longer scheduled" and stops, which confirms the lookup path works. To exercise the full migration, temporarily change `maintenance_window` to return a fake window a few hours out. + +## What This Does Not Cover + +- **Multi-instance jobs.** Each affected instance gets its own `upcoming_downtime` event. A distributed job needs a coordinator that moves all ranks together. +- **Windows that fire while the receiver is down.** Timers live in process memory. For production, persist the pending migrations or poll the maintenance endpoint for your instances' machines on startup. +- **Contract expiry.** A rental that reaches its end date is deleted, not paused. The **Instance expiring soon** notification gives three days of notice and pairs naturally with the same `save_checkpoint` step. + +## Production Notes + +- Run the receiver on an always-on host with a stable HTTPS URL, not on a rented GPU instance. +- Persist any state that must survive a restart: pending migrations in Tutorial 1, the pool and blacklist in Tutorial 2. +- Keep a periodic reconciliation against `GET /api/v0/instances/` as a backstop. Webhook delivery is at-least-once and can be delayed by retries. +- Rotate the webhook secret if it is ever exposed, and update `VAST_WEBHOOK_SECRET` before the next delivery. +- Before depending on any `data` field, confirm it with a test delivery or a simulated event. The set of fields per event is documented in [Notification Webhooks](/guides/reference/notification-webhooks) and may grow over time. diff --git a/examples/notifications/offline-aware-load-balancer.mdx b/examples/notifications/offline-aware-load-balancer.mdx new file mode 100644 index 00000000..d56a36c7 --- /dev/null +++ b/examples/notifications/offline-aware-load-balancer.mdx @@ -0,0 +1,317 @@ +--- +title: "Route Around Offline Instances with a Simple Load Balancer" +slug: "vast-notifications-offline-aware-load-balancer" +createdAt: "Wed Sep 09 2026 00:00:00 GMT+0000 (Coordinated Universal Time)" +updatedAt: "Wed Sep 09 2026 00:00:00 GMT+0000 (Coordinated Universal Time)" +--- + +import WebhookAutomationSetup from '/snippets/notifications/webhook-automation-setup.mdx'; + +Use this example to keep a serving pool healthy when a machine fails. Vast.ai sends an **Instance offline** notification when a host machine stops reporting; the handler below drops the affected instances from a minimal HTTP load balancer immediately, rents a replacement on a different machine, and re-admits it only after your own readiness probe passes. + +This page builds on the receiver pattern from [Send Notifications to Slack](/examples/notifications/slack-webhook). For the companion example that handles scheduled host maintenance, see [Migrate an Instance Before Scheduled Maintenance](/examples/notifications/maintenance-migration). + + + +## What Happens Without This + +You serve a model from several instances behind a load balancer. One machine fails or its host takes it offline. Your balancer keeps sending it traffic until a health check eventually times out, and every request routed there in the meantime fails. + +Vast.ai sends an **Instance offline** notification when the host machine stops reporting. This tutorial uses that push signal to drop the dead instance from rotation immediately, rent a replacement on a different machine, and add it back to the pool only when your own readiness probe says the model is loaded. + +The balancer here is a minimal HTTP proxy in Python, kept small so the event handling is visible. In production you would apply the same handlers to whatever proxy you already run. + +## How the Events Fit Together + +| Event | `data` fields | Your action | +| --- | --- | --- | +| `instance_offline` | `instance_ids[]`, `machine_ids[]` | Drain every listed instance; avoid the listed machines when replacing | +| `instance_online` | `instance_id`, `machine_id` | The offline instance recovered: probe it, then restore | +| `instance_started` | `instance_id`, `machine_id` | A new rental is running for the first time: probe it, then admit | +| `instance_resumed` | `instance_id`, `machine_id`, `actual_status` | Left "Scheduling"; only probe if `actual_status` is `running` | +| `instance_stopped` | `instance_id` | Drain | +| `instance_deleted` | `instance_id` | Remove from the pool | + +Two details matter here: + +- **`instance_offline` is plural.** When a machine goes down, every instance you had on it is reported in one delivery. That is why the payload carries `machine_ids` as well: the failure is correlated, and a replacement rented onto the same machine will fail the same way. +- **No event means "your model is ready."** `instance_online` and `instance_started` mean the platform can see the container. Gate re-admission on your own readiness probe. + +## Write the Load Balancer + +Create `balancer.py`. It holds the pool, runs a proxy on port 9000, and registers handlers with the shared receiver. + +```python +#!/usr/bin/env python3 +import itertools +import os +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import requests +from vastai import VastAI + +from receiver import on, serve + +vast = VastAI(api_key=os.environ["VAST_API_KEY"]) + +POOL_LABEL = os.environ.get("POOL_LABEL", "inference") # label on every replica instance +MODEL_PORT = os.environ.get("MODEL_PORT", "8000") # port your model server listens on +READY_PATH = os.environ.get("READY_PATH", "/health") +PROXY_PORT = int(os.environ.get("PROXY_PORT", "9000")) +POOL_SIZE = int(os.environ.get("POOL_SIZE", "3")) +BLACKLIST_TTL = 3600 + +POOL: dict[int, dict] = {} # instance_id -> {"url": str, "machine_id": int, "healthy": bool} +BLACKLIST: dict[int, float] = {} # machine_id -> expires_at +_lock = threading.Lock() +_rr = itertools.count() + + +# ---- pool bookkeeping ------------------------------------------------------- + +def instance_url(inst: dict) -> str | None: + ports = (inst.get("ports") or {}).get(f"{MODEL_PORT}/tcp") or [] + if not inst.get("public_ipaddr") or not ports: + return None + return f"http://{inst['public_ipaddr']}:{ports[0]['HostPort']}" + + +def ready(url: str) -> bool: + try: + return requests.get(url + READY_PATH, timeout=3).status_code == 200 + except requests.RequestException: + return False + + +def drain(instance_id: int, reason: str) -> None: + with _lock: + if instance_id in POOL: + POOL[instance_id]["healthy"] = False + print(f"drained {instance_id}: {reason}", flush=True) + + +def remove(instance_id: int) -> None: + with _lock: + POOL.pop(instance_id, None) + print(f"removed {instance_id}", flush=True) + + +def admit_when_ready(instance_id: int, attempts: int = 60, interval: int = 10) -> None: + """Probe your readiness endpoint until it passes, then put the instance in rotation.""" + for _ in range(attempts): + inst = vast.show_instance(id=instance_id) + url = instance_url(inst) + if inst.get("actual_status") == "running" and url and ready(url): + with _lock: + POOL[instance_id] = {"url": url, "machine_id": inst["machine_id"], "healthy": True} + print(f"admitted {instance_id} at {url}", flush=True) + return + time.sleep(interval) + print(f"gave up waiting for {instance_id} to become ready", flush=True) + + +def healthy_urls() -> list[str]: + with _lock: + return [r["url"] for r in POOL.values() if r["healthy"]] + + +def active_machines() -> list[int]: + now = time.time() + with _lock: + expired = [m for m, t in BLACKLIST.items() if t < now] + for m in expired: + del BLACKLIST[m] + return list(BLACKLIST) + [r["machine_id"] for r in POOL.values()] + + +# ---- capacity --------------------------------------------------------------- + +def rent_replica(exclude_machines: list[int]) -> None: + """Rent one more replica, never on a blacklisted machine or one already in the pool.""" + offers = vast.search_offers( + query=f"{os.environ['OFFER_QUERY']} rentable=true", order="dph", limit=20 + ) + offer = next((o for o in offers if o["machine_id"] not in exclude_machines), None) + if offer is None: + print("no eligible offers; pool stays short", flush=True) + return + result = vast.create_instance( + id=offer["id"], + image=os.environ["MODEL_IMAGE"], + disk=float(os.environ.get("MODEL_DISK_GB", "40")), + label=POOL_LABEL, + direct=True, + env=f"-p {MODEL_PORT}:{MODEL_PORT}", + onstart_cmd=os.environ.get("MODEL_START_CMD"), + ) + print(f"rented replacement {result['new_contract']} on machine {offer['machine_id']}", flush=True) + # admission happens when instance_started arrives, or via the periodic probe + + +def bootstrap_pool() -> None: + """On startup, adopt existing replicas and top the pool up to POOL_SIZE.""" + for inst in vast.show_instances(): + if inst.get("label") == POOL_LABEL: + threading.Thread(target=admit_when_ready, args=(inst["id"],), daemon=True).start() + with _lock: + short = POOL_SIZE - sum(1 for i in vast.show_instances() if i.get("label") == POOL_LABEL) + for _ in range(max(0, short)): + rent_replica(active_machines()) + + +def periodic_probe() -> None: + """Backstop: webhooks are at-least-once, not instant. Re-check the pool every 30s.""" + while True: + time.sleep(30) + with _lock: + snapshot = list(POOL.items()) + for iid, r in snapshot: + ok = ready(r["url"]) + with _lock: + if iid in POOL: + POOL[iid]["healthy"] = ok + + +# ---- webhook handlers ------------------------------------------------------- + +@on("instance_offline") +def on_instance_offline(evt: dict) -> None: + d = evt.get("data") or {} + lost = [i for i in d.get("instance_ids", []) if i in POOL] + for iid in lost: + drain(iid, "instance_offline") + with _lock: + for mid in d.get("machine_ids", []): + BLACKLIST[mid] = time.time() + BLACKLIST_TTL + for _ in lost: + rent_replica(active_machines()) + + +@on("instance_online") +def on_instance_online(evt: dict) -> None: + iid = (evt.get("data") or {}).get("instance_id") + if iid in POOL: + threading.Thread(target=admit_when_ready, args=(iid,), daemon=True).start() + + +@on("instance_started") +def on_instance_started(evt: dict) -> None: + iid = (evt.get("data") or {}).get("instance_id") + inst = vast.show_instance(id=iid) + if inst.get("label") == POOL_LABEL: + threading.Thread(target=admit_when_ready, args=(iid,), daemon=True).start() + + +@on("instance_resumed") +def on_instance_resumed(evt: dict) -> None: + d = evt.get("data") or {} + if d.get("actual_status") != "running": # may have left Scheduling into stopped/exited + return + if d.get("instance_id") in POOL: + threading.Thread(target=admit_when_ready, args=(d["instance_id"],), daemon=True).start() + + +@on("instance_stopped") +def on_instance_stopped(evt: dict) -> None: + iid = (evt.get("data") or {}).get("instance_id") + if iid in POOL: + drain(iid, "instance_stopped") + + +@on("instance_deleted") +def on_instance_deleted(evt: dict) -> None: + iid = (evt.get("data") or {}).get("instance_id") + if iid in POOL: + remove(iid) + + +# ---- the proxy -------------------------------------------------------------- + +class Proxy(BaseHTTPRequestHandler): + def do_POST(self): + body = self.rfile.read(int(self.headers.get("Content-Length", "0") or "0")) + urls = healthy_urls() + if not urls: + self.send_response(503); self.end_headers(); return + target = urls[next(_rr) % len(urls)] + try: + r = requests.post(target + self.path, data=body, timeout=120, + headers={"Content-Type": self.headers.get("Content-Type", "application/json")}) + except requests.RequestException: + for iid, rec in list(POOL.items()): # mark the replica we just failed on + if rec["url"] == target: + drain(iid, "request failed") + self.send_response(502); self.end_headers(); return + self.send_response(r.status_code) + self.send_header("Content-Type", r.headers.get("Content-Type", "application/json")) + self.send_header("Content-Length", str(len(r.content))) + self.end_headers() + self.wfile.write(r.content) + + def log_message(self, fmt, *args): + pass + + +if __name__ == "__main__": + bootstrap_pool() + threading.Thread(target=periodic_probe, daemon=True).start() + threading.Thread( + target=lambda: ThreadingHTTPServer(("0.0.0.0", PROXY_PORT), Proxy).serve_forever(), + daemon=True, + ).start() + print(f"proxy on :{PROXY_PORT}", flush=True) + serve() +``` + +What each part does: + +- **Drain first, replace second.** On `instance_offline` the handler removes the listed instances from rotation before it does anything else. Renting the replacement is slower and happens after. +- **The blacklist is keyed by machine, not instance.** `rent_replica` skips any offer on a machine that failed in the last hour and any machine already hosting a replica. Two replicas on one machine are not redundant. +- **Re-admission always goes through `admit_when_ready`.** Every "it is back" event, from any source, ends in your readiness probe. The event decides *when to look*; the probe decides *whether to serve*. +- **The periodic probe is a backstop, not the primary signal.** It catches the case where a delivery was delayed or the receiver was restarted. + +## Run It + +Set the environment for your model server. This example assumes a vLLM-style server on port 8000 with a `/health` route: + +```bash +export VAST_API_KEY VAST_WEBHOOK_SECRET +export OFFER_QUERY='gpu_name=RTX_4090 num_gpus=1 dph<=0.6 inet_down>=500' +export MODEL_IMAGE='vllm/vllm-openai:latest' +export MODEL_START_CMD='python3 -m vllm.entrypoints.openai.api_server --model --port 8000' +export POOL_SIZE=3 +python3 balancer.py +``` + +Send a request through the proxy: + +```bash +curl -sS http://127.0.0.1:9000/v1/completions \ + -H "Content-Type: application/json" \ + -d '{"model": "", "prompt": "Hello", "max_tokens": 8}' +``` + +Then simulate a machine loss for one of your replicas. Use the instance and machine IDs from `vastai show instances`: + +```bash +python3 simulate.py instance_offline \ + '{"instance_ids": [], "machine_ids": []}' +``` + +The log should show `drained`, then `rented replacement`, and, once the new replica's `instance_started` delivery arrives and the probe passes, `admitted`. Requests through the proxy keep succeeding throughout. Destroy the drained instance yourself when you are done testing; the handler does not delete instances the platform says are offline, because they may come back. + +## What This Does Not Cover + +- **Streaming responses.** The proxy buffers the whole response. Use it as a model for the event handling, not as your production proxy. +- **Pool size after recovery.** If a drained instance comes back online and passes the probe, the pool is temporarily one replica over `POOL_SIZE`. Decide whether to keep the extra capacity or destroy the newer one. +- **Scheduled maintenance.** `upcoming_downtime` gives you the same machine ID hours in advance. Pair this tutorial with [Migrate an Instance Before Scheduled Maintenance](/examples/notifications/maintenance-migration): rent the replacement before the window, then drain the old replica gracefully. + +## Production Notes + +- Run the receiver on an always-on host with a stable HTTPS URL, not on a rented GPU instance. +- Persist any state that must survive a restart: pending migrations in Tutorial 1, the pool and blacklist in Tutorial 2. +- Keep a periodic reconciliation against `GET /api/v0/instances/` as a backstop. Webhook delivery is at-least-once and can be delayed by retries. +- Rotate the webhook secret if it is ever exposed, and update `VAST_WEBHOOK_SECRET` before the next delivery. +- Before depending on any `data` field, confirm it with a test delivery or a simulated event. The set of fields per event is documented in [Notification Webhooks](/guides/reference/notification-webhooks) and may grow over time. diff --git a/snippets/notifications/webhook-automation-setup.mdx b/snippets/notifications/webhook-automation-setup.mdx new file mode 100644 index 00000000..52c41b22 --- /dev/null +++ b/snippets/notifications/webhook-automation-setup.mdx @@ -0,0 +1,245 @@ +## Before You Start + +### Prerequisites + +- A Vast.ai API key from [Keys](https://cloud.vast.ai/manage-keys/) +- Python 3.10 or newer with `requests` and the `vastai` SDK installed +- A public HTTPS URL for your receiver. For local testing use a tunnel such as Tunnelmole, ngrok, or Cloudflare Tunnel; for production deploy the receiver on an always-on host + +```bash +mkdir vast-webhook-automation +cd vast-webhook-automation +python3 -m venv .venv +. .venv/bin/activate +pip install requests vastai +``` + + +Run the receiver somewhere other than the GPU instances it manages. Its job is to outlive them. + + +### The Event Payload + +Every delivery is a signed JSON `POST`. The fields these tutorials use are `notif_type`, `event_id`, and `data`: + +```json +{ + "event_id": "7e9a2c4e6f9e4a24a53b77c2d8e3f0aa", + "user_id": 123, + "notif_type": "upcoming_downtime", + "subject": "Upcoming Instance Downtime", + "message": "Your instance 33502355 on machine 447823 is scheduled for maintenance ...", + "data": { "instance_id": 33502355, "machine_id": 447823, "maintenance_id": 88213 }, + "data_truncated": false, + "timestamp": 1757433600.123 +} +``` + +`data` is always a JSON object or `null`. If the platform had to drop it for size, `data_truncated` is `true` and you should fall back to reading the instance from the API. Signing, headers, and retry rules are documented in [Notification Webhooks](/guides/reference/notification-webhooks). + +Three rules apply to every receiver in this guide: + +1. **Verify the signature** over the raw request body before doing anything else. +2. **Return `2xx` within 10 seconds.** Put the event on a queue and do the work in a background thread. Any `4xx` other than `408` or `429` is treated as a permanent failure and the event is not retried. +3. **Deduplicate on `event_id`.** Delivery is at-least-once. + +### Shared Receiver + +Both downtime tutorials plug handlers into the same receiver. Create `receiver.py`: + +```python +#!/usr/bin/env python3 +"""Verify, deduplicate, enqueue, acknowledge. Handlers run on a worker thread.""" +import hashlib +import hmac +import json +import os +import queue +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +VAST_WEBHOOK_SECRET = os.environ["VAST_WEBHOOK_SECRET"].encode("utf-8") +PORT = int(os.environ.get("PORT", "8787")) +MAX_SIGNATURE_AGE_SECONDS = 300 + +EVENTS: "queue.Queue[dict]" = queue.Queue() +HANDLERS: dict[str, callable] = {} # notif_type -> function(evt) +_seen: dict[str, float] = {} # event_id -> first-seen time +_seen_lock = threading.Lock() + + +def on(notif_type: str): + """Decorator: register a handler for a notification type slug.""" + def register(fn): + HANDLERS[notif_type] = fn + return fn + return register + + +def verify_vast_signature(headers, raw_body: bytes) -> bool: + timestamp = headers.get("X-Vast-Timestamp", "") + signature = headers.get("X-Vast-Signature-256", "") + if not timestamp or not signature.startswith("sha256="): + return False + try: + if abs(time.time() - int(timestamp)) > MAX_SIGNATURE_AGE_SECONDS: + return False + except ValueError: + return False + digest = hmac.new( + VAST_WEBHOOK_SECRET, timestamp.encode("utf-8") + b"." + raw_body, hashlib.sha256 + ).hexdigest() + return hmac.compare_digest(signature, f"sha256={digest}") + + +def first_time_seen(event_id: str) -> bool: + now = time.time() + with _seen_lock: + for k, t in list(_seen.items()): # expire after 24h + if now - t > 86400: + del _seen[k] + if event_id in _seen: + return False + _seen[event_id] = now + return True + + +class Handler(BaseHTTPRequestHandler): + def _reply(self, status: int): + self.send_response(status) + self.send_header("Content-Length", "0") + self.end_headers() + + def do_GET(self): + self._reply(200 if self.path == "/health" else 404) + + def do_POST(self): + raw = self.rfile.read(int(self.headers.get("Content-Length", "0") or "0")) + if not verify_vast_signature(self.headers, raw): + self._reply(401) + return + try: + evt = json.loads(raw.decode("utf-8") or "{}") + except json.JSONDecodeError: + self._reply(400) + return + if not isinstance(evt, dict) or "event_id" not in evt: + self._reply(400) + return + if first_time_seen(evt["event_id"]): + EVENTS.put(evt) + self._reply(202) # acknowledge before doing any work + + def log_message(self, fmt, *args): + print(f"{self.address_string()} - {fmt % args}", flush=True) + + +def worker(): + while True: + evt = EVENTS.get() + fn = HANDLERS.get(evt.get("notif_type")) + if fn is None: + print(f"ignored notif_type={evt.get('notif_type')}", flush=True) + continue + try: + fn(evt) + except Exception as e: # never let one event kill the loop + print(f"handler error notif_type={evt.get('notif_type')}: {e!r}", flush=True) + + +def serve(): + threading.Thread(target=worker, daemon=True).start() + print(f"listening on http://127.0.0.1:{PORT}", flush=True) + ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever() +``` + + +A `401` on a bad signature discards that event permanently. That is the right behavior for a request that is not from Vast.ai, but it means a wrong secret in `VAST_WEBHOOK_SECRET` silently loses real events. Always confirm with the test delivery before relying on the receiver. + + +### Create the Webhook + +Create a webhook subscribed to the events you need. The list below covers both downtime tutorials; trim it if you only follow one. The response includes the signing secret, which is returned only at creation and on rotation. + +```bash +read -rsp "Vast.ai API key: " VAST_API_KEY; export VAST_API_KEY; echo + +cat > create-webhook.json <