Vast.ai Core Workflow B: Multi-Instance & Cost Optimization

Overview

Secondary workflow for Vast.ai: orchestrate multiple GPU instances for distributed training, implement automatic spot interruption recovery with checkpoint-based resume, and analyze spending to reduce per-job cost.

Prerequisites

• Completed vastai-core-workflow-a
• Understanding of distributed training (PyTorch DDP, DeepSpeed)
• Checkpoint-based training pipeline

Instructions

Step 1: Multi-Instance Provisioning

import subprocess, json, time
from concurrent.futures import ThreadPoolExecutor

def provision_cluster(num_nodes, gpu_name="A100", min_vram=80, image=""):
    """Provision multiple GPU instances for distributed training."""
    # Search for matching offers
    query = (f"num_gpus=1 gpu_name={gpu_name} gpu_ram>={min_vram} "
             f"reliability>0.98 inet_down>500 rentable=true")
    result = subprocess.run(
        ["vastai", "search", "offers", query, "--order", "dph_total",
         "--raw", "--limit", str(num_nodes * 3)],
        capture_output=True, text=True, check=True,
    )
    offers = json.loads(result.stdout)
    if len(offers) < num_nodes:
        raise RuntimeError(f"Only {len(offers)} offers, need {num_nodes}")

    # Provision nodes in parallel
    instances = []
    for i, offer in enumerate(offers[:num_nodes]):
        inst_id = provision_single(offer["id"], image, rank=i)
        instances.append({"id": inst_id, "rank": i, "offer": offer})

    # Wait for all to be running
    for inst in instances:
        info = wait_for_running(inst["id"])
        inst.update({"ssh_host": info["ssh_host"], "ssh_port": info["ssh_port"]})

    return instances

Step 2: Spot Interruption Recovery

class SpotRecoveryManager:
    """Monitor instances and replace preempted spot instances."""

    def __init__(self, client, checkpoint_dir="/workspace/checkpoints"):
        self.client = client
        self.checkpoint_dir = checkpoint_dir

    def monitor_and_recover(self, instances, image, poll_interval=60):
        """Poll instance status; replace any that are destroyed/error."""
        while True:
            for inst in instances:
                result = subprocess.run(
                    ["vastai", "show", "instance", str(inst["id"]), "--raw"],
                    capture_output=True, text=True,
                )
                info = json.loads(result.stdout)
                status = info.get("actual_status", 

Vast.ai Cost Tuning

Overview

Minimize Vast.ai GPU cloud costs by choosing the right GPU for your workload, leveraging interruptible (spot) instances, eliminating idle compute, and implementing auto-destroy safeguards. Vast.ai pricing is dynamic and varies significantly: RTX 4090 ($0.15-0.30/hr), A100 80GB ($1.00-2.00/hr), H100 SXM ($2.50-4.00/hr).

Prerequisites

• Vast.ai account with billing history
• Understanding of your workload's GPU requirements
• vastai CLI installed

Instructions

Step 1: GPU Selection by Cost-Efficiency

# Compare cost-per-TFLOP across GPU types
GPU_SPECS = {
    "RTX_4090":  {"fp16_tflops": 82.6,  "vram": 24},
    "A100":      {"fp16_tflops": 77.97, "vram": 80},
    "H100_SXM":  {"fp16_tflops": 267,   "vram": 80},
    "RTX_3090":  {"fp16_tflops": 35.6,  "vram": 24},
    "A6000":     {"fp16_tflops": 38.7,  "vram": 48},
}

def cost_per_tflop(gpu_name, dph):
    specs = GPU_SPECS.get(gpu_name, {"fp16_tflops": 1})
    return dph / specs["fp16_tflops"]

# Often RTX 4090 is the best value for inference
# A100 is best for training large models needing >24GB VRAM
# H100 is best only when wall-clock time justifies 10x price premium

Step 2: Spot vs On-Demand Analysis

# Interruptible (spot) instances are 30-60% cheaper
vastai search offers 'num_gpus=1 gpu_name=RTX_4090 rentable=true' \
  --order dph_total --limit 5
# Compare interruptible vs on-demand pricing
# Use interruptible for: batch inference, checkpointed training
# Use on-demand for: final training epochs, production inference

Step 3: Auto-Destroy Safeguards

import time, subprocess, json

def auto_destroy_after(instance_id, max_hours=4):
    """Destroy instance after max_hours to prevent cost overruns."""
    max_seconds = max_hours * 3600
    time.sleep(max_seconds)
    subprocess.run(["vastai", "destroy", "instance", str(instance_id)], check=True)
    print(f"Instance {instance_id} auto-destroyed after {max_hours}h")

# Run in background thread when provisioning
import threading
watchdog = threading.Thread(target=auto_destroy_after, args=(inst_id, 4), daemon=True)
watchdog.start()

Step 4: Idle Instance Detection

#!/bin/bash
# Find and destroy idle instances (GPU util < 10% for >10 min)
vastai show instances --raw | python3 -c "
import sys, json
for inst in json.load(sys.stdin):
    if inst.get('actual_status') == 'running':
        gpu_util = inst.get('gpu_util', 0)
        if gpu_util < 10:
            print(

Vast.ai Data Handling

Overview

Manage training data and model artifacts securely on Vast.ai GPU instances. Covers data transfer, encryption, checkpoint management, and cleanup. Critical consideration: Vast.ai instances run on shared hardware operated by third-party hosts.

Prerequisites

• Vast.ai instance with SSH access
• Cloud storage (S3, GCS) for persistent artifacts
• Understanding of data sensitivity classification

Instructions

Step 1: Data Transfer Patterns

# Small datasets (<5GB): Direct SCP
scp -P $PORT -r ./data/ root@$HOST:/workspace/data/

# Large datasets (5-50GB): Compressed transfer
tar czf - ./data/ | ssh -p $PORT root@$HOST "tar xzf - -C /workspace/"

# Very large datasets (>50GB): Cloud storage staging
# Upload to S3/GCS first, then download on instance
ssh -p $PORT root@$HOST "aws s3 sync s3://bucket/dataset/ /workspace/data/"

Step 2: Encrypted Data Transfer

import subprocess, os

def encrypt_and_upload(local_path, host, port, remote_path, passphrase):
    """Encrypt data before transferring to Vast.ai instance."""
    encrypted = f"{local_path}.enc"
    # Encrypt with AES-256
    subprocess.run([
        "openssl", "enc", "-aes-256-cbc", "-salt", "-pbkdf2",
        "-in", local_path, "-out", encrypted,
        "-pass", f"pass:{passphrase}",
    ], check=True)

    # Transfer encrypted file
    subprocess.run([
        "scp", "-P", str(port), encrypted,
        f"root@{host}:{remote_path}.enc",
    ], check=True)

    # Decrypt on instance
    subprocess.run([
        "ssh", "-p", str(port), f"root@{host}",
        f"openssl enc -aes-256-cbc -d -pbkdf2 "
        f"-in {remote_path}.enc -out {remote_path} "
        f"-pass pass:{passphrase} && rm {remote_path}.enc"
    ], check=True)

    os.remove(encrypted)

Step 3: Checkpoint to Cloud Storage

import torch, boto3, os

class CloudCheckpointManager:
    def __init__(self, s3_bucket, prefix, save_every=500):
        self.s3 = boto3.client("s3")
        self.bucket = s3_bucket
        self.prefix = prefix
        self.save_every = save_every

    def save(self, model, optimizer, step, loss):
        if step % self.save_every != 0:
            return
        local_path = f"/tmp/ckpt-{step}.pt"
        torch.save({
            "step": step, "loss": loss,
            "model": model.state_dict(),
            "optimizer": optimizer.state_dict(),
        }, local_path)
        self.s3.upload_file(local_path, self.bucket,
                           f"{self.pre

Vast.ai Debug Bundle

Current State

!vastai --version 2>/dev/null || echo 'vastai CLI not installed'

!python3 --version 2>/dev/null || echo 'Python not available'

Overview

Collect comprehensive diagnostic information for Vast.ai GPU instance issues. Covers account verification, instance inspection, log collection, GPU diagnostics, and network testing.

Prerequisites

• Vast.ai CLI installed and authenticated
• Access to the problematic instance (if still running)

Instructions

Step 1: Account and Auth Diagnostics

#!/bin/bash
set -euo pipefail
echo "=== Vast.ai Debug Bundle ==="
echo "Timestamp: $(date -u +%Y-%m-%dT%H:%M:%SZ)"

echo -e "\n--- Account Info ---"
vastai show user --raw | python3 -c "
import sys, json
u = json.load(sys.stdin)
print(f'Username: {u.get(\"username\", \"?\")}')
print(f'Balance: \${u.get(\"balance\", 0):.2f}')
print(f'API Key (first 8): {u.get(\"api_key\", \"?\")[:8]}...')
"

Step 2: Instance Status Collection

echo -e "\n--- All Instances ---"
vastai show instances --raw | python3 -c "
import sys, json
instances = json.load(sys.stdin)
for i in instances:
    print(f'ID: {i[\"id\"]} | Status: {i.get(\"actual_status\", \"?\")} | '
          f'GPU: {i.get(\"gpu_name\", \"?\")} | '
          f'\$/hr: {i.get(\"dph_total\", 0):.3f} | '
          f'SSH: {i.get(\"ssh_host\", \"?\")}:{i.get(\"ssh_port\", \"?\")}')
"

Step 3: Instance Log Collection

# Collect logs from a specific instance
INSTANCE_ID="${1:-}"
if [ -n "$INSTANCE_ID" ]; then
    echo -e "\n--- Instance $INSTANCE_ID Logs ---"
    vastai logs "$INSTANCE_ID" --tail 100 2>/dev/null || echo "No logs available"

    echo -e "\n--- Instance $INSTANCE_ID Details ---"
    vastai show instance "$INSTANCE_ID" --raw | python3 -c "
import sys, json
i = json.load(sys.stdin)
for key in ['actual_status', 'status_msg', 'gpu_name', 'gpu_ram',
            'cuda_max_good', 'disk_space', 'ssh_host', 'ssh_port',
            'image_uuid', 'onstart', 'reliability2']:
    print(f'{key}: {i.get(key, \"?\")}')
"
fi

Step 4: Remote GPU Diagnostics (if SSH accessible)

if [ -n "$SSH_HOST" ] && [ -n "$SSH_PORT" ]; then
    echo -e "\n--- GPU Diagnostics (remote) ---"
    ssh -p "$SSH_PORT" -o StrictHostKeyC

Vast.ai Deploy Integration

Overview

Deploy ML training jobs and inference services on Vast.ai GPU cloud. Covers Docker image optimization, automated provisioning scripts, data transfer strategies, and deployment automation.

Prerequisites

• Vast.ai CLI authenticated
• Docker image published to a registry
• Training/inference code tested locally

Instructions

Step 1: Optimized Docker Image

# Dockerfile.vastai — optimized for fast pulls on Vast.ai
FROM pytorch/pytorch:2.2.0-cuda12.1-cudnn8-runtime

# Install dependencies in a single layer
COPY requirements.txt /tmp/
RUN pip install --no-cache-dir -r /tmp/requirements.txt && rm /tmp/requirements.txt

# Copy application code
COPY src/ /workspace/src/
COPY scripts/ /workspace/scripts/

WORKDIR /workspace
CMD ["python", "src/train.py"]

# Build and push
docker build -t ghcr.io/yourorg/training:v1 -f Dockerfile.vastai .
docker push ghcr.io/yourorg/training:v1

Step 2: Automated Deployment Script

#!/usr/bin/env python3
"""deploy.py — Automated Vast.ai deployment with monitoring."""
import subprocess, json, time, argparse, sys

def deploy(args):
    # Search for matching offer
    query = (f"num_gpus={args.gpus} gpu_name={args.gpu} "
             f"reliability>{args.reliability} dph_total<={args.max_price} "
             f"disk_space>={args.disk} rentable=true")

    offers = json.loads(subprocess.run(
        ["vastai", "search", "offers", query, "--order", "dph_total",
         "--raw", "--limit", "5"],
        capture_output=True, text=True, check=True).stdout)

    if not offers:
        print(f"ERROR: No offers matching: {query}", file=sys.stderr)
        sys.exit(1)

    offer = offers[0]
    print(f"Selected: {offer['gpu_name']} ${offer['dph_total']:.3f}/hr "
          f"(ID: {offer['id']})")

    # Create instance
    cmd = ["vastai", "create", "instance", str(offer["id"]),
           "--image", args.image, "--disk", str(args.disk)]
    if args.onstart:
        cmd.extend(["--onstart-cmd", args.onstart])

    result = json.loads(subprocess.run(
        cmd, capture_output=True, text=True, check=True).stdout)
    instance_id = result["new_contract"]
    print(f"Instance {instance_id} provisioning...")

    # Wait for running
    for _ in range(30):
        info = json.loads(subprocess.run(
            ["vastai", "show", "instance", str(instance_id), "--raw"],
            capture_output=True, text=True).stdout)
        if info.get("

Vast.ai Enterprise RBAC

Overview

Control access to Vast.ai GPU instances and spending through API key management, team-level budgets, and GPU allocation policies. Vast.ai uses a marketplace model with per-GPU-hour pricing (RTX 4090 ~$0.20/hr, A100 ~$1.50/hr, H100 ~$3.00/hr).

Prerequisites

• Vast.ai account(s) with API keys
• Understanding of team GPU usage patterns
• Budget allocation per team/project

Instructions

Step 1: Team API Key Strategy

# Separate API keys per team for billing isolation
# Option A: Separate Vast.ai accounts per team
# Option B: Single account with application-level controls

TEAM_CONFIGS = {
    "ml-research": {
        "api_key_env": "VASTAI_KEY_RESEARCH",
        "gpu_whitelist": ["A100", "H100_SXM"],
        "max_instances": 8,
        "daily_budget": 200.00,
        "max_dph": 4.00,
    },
    "ml-engineering": {
        "api_key_env": "VASTAI_KEY_ENGINEERING",
        "gpu_whitelist": ["RTX_4090", "A100"],
        "max_instances": 4,
        "daily_budget": 50.00,
        "max_dph": 2.00,
    },
    "data-science": {
        "api_key_env": "VASTAI_KEY_DATASCIENCE",
        "gpu_whitelist": ["RTX_4090", "RTX_3090"],
        "max_instances": 2,
        "daily_budget": 10.00,
        "max_dph": 0.30,
    },
}

Step 2: Policy Enforcement Layer

class VastPolicyEnforcer:
    def __init__(self, team_config):
        self.config = team_config
        self.client = VastClient(api_key=os.environ[team_config["api_key_env"]])

    def can_provision(self, gpu_name, num_gpus=1):
        """Check if provisioning is allowed by team policy."""
        if gpu_name not in self.config["gpu_whitelist"]:
            return False, f"GPU {gpu_name} not in team whitelist"

        running = len([i for i in self.client.show_instances()
                      if i.get("actual_status") == "running"])
        if running >= self.config["max_instances"]:
            return False, f"Instance limit reached ({running}/{self.config['max_instances']})"

        return True, "OK"

    def provision_with_policy(self, gpu_name, image, disk_gb=20):
        allowed, reason = self.can_provision(gpu_name)
        if not allowed:
            raise PermissionError(f"Policy violation: {reason}")

        offers = self.client.search_offers({
            "gpu_name": {"eq": gpu_name},
            "dph_total": {"lte": self.config["max_dph"]},
            "

Vast.ai Hello World

Overview

Rent your first GPU instance on Vast.ai, run a PyTorch workload, and destroy the instance when done. Demonstrates the full lifecycle: search offers, create instance, connect via SSH, run a job, and tear down.

Prerequisites

• Completed vastai-install-auth setup
• Vast.ai account with credits ($1+ recommended for testing)
• SSH key uploaded to Vast.ai (cloud.vast.ai > Account > SSH Keys)

Instructions

Step 1: Search for Available GPUs (CLI)

# Find cheap single-GPU offers sorted by price
vastai search offers 'num_gpus=1 gpu_ram>=8 inet_down>100 reliability>0.95' \
  --order 'dph_total' --limit 5

# Output columns: ID, GPU, VRAM, $/hr, DLPerf, Reliability, Location

Step 2: Search for Available GPUs (REST API)

curl -s -H "Authorization: Bearer $VASTAI_API_KEY" \
  "https://cloud.vast.ai/api/v0/bundles/?q=%7B%22num_gpus%22%3A%7B%22eq%22%3A1%7D%2C%22gpu_ram%22%3A%7B%22gte%22%3A8%7D%2C%22reliability2%22%3A%7B%22gte%22%3A0.95%7D%2C%22rentable%22%3A%7B%22eq%22%3Atrue%7D%7D&order=dph_total&limit=5" \
  | jq '.offers[:3] | .[] | {id, gpu_name, num_gpus, gpu_ram, dph_total, reliability2}'

Step 3: Create an Instance (CLI)

# Replace OFFER_ID with the ID from search results
vastai create instance OFFER_ID \
  --image pytorch/pytorch:2.2.0-cuda12.1-cudnn8-runtime \
  --disk 20 \
  --onstart-cmd "echo 'Instance ready'"

Step 4: Create an Instance (Python)

from vastai_client import VastClient

client = VastClient()

# Search for affordable RTX 4090 offers
offers = client.search_offers({
    "num_gpus": {"eq": 1},
    "gpu_name": {"eq": "RTX_4090"},
    "reliability2": {"gte": 0.95},
    "rentable": {"eq": True},
})

# Pick the cheapest offer
best = sorted(offers["offers"], key=lambda o: o["dph_total"])[0]
print(f"Best offer: {best['gpu_name']} at ${best['dph_total']:.3f}/hr (ID: {best['id']})")

# Create instance with PyTorch image
instance = client.create_instance(
    offer_id=best["id"],
    image="pytorch/pytorch:2.2.0-cuda12.1-cudnn8-runtime",
    disk_gb=20,
    onstart="nvidia-smi && python -c 'import torch; print(torch.cuda.is_available())'",
)
print(f"Instance created: {instance}")

Step 5: Monitor and Connect

# Check instance status (wait for 'running')
vastai show instances --raw | jq '.[] | {id, actual_status, ssh_host, ssh_port}'

# Connect via SSH once running
ssh -p SSH_PORT root@SSH_HOST

# On the instanc

Vast.ai Incident Runbook

Overview

Rapid incident response procedures for Vast.ai GPU instance failures. Covers triage, mitigation, recovery, and postmortem for common incident types: spot preemption, instance crashes, GPU failures, and billing issues.

Prerequisites

• Vast.ai CLI access
• SSH access to instances (if still running)
• Checkpoint storage accessible (S3/GCS)

Instructions

Triage: Assess Impact (< 2 minutes)

#!/bin/bash
set -euo pipefail
echo "=== INCIDENT TRIAGE ==="
echo "Time: $(date -u)"

# 1. Check all instances
echo -e "\n--- Instance Status ---"
vastai show instances --raw | python3 -c "
import sys, json
for inst in json.load(sys.stdin):
    status = inst.get('actual_status', '?')
    flag = 'ALERT' if status in ('error', 'exited', 'offline') else 'OK'
    print(f'  [{flag}] ID:{inst[\"id\"]} Status:{status} '
          f'GPU:{inst.get(\"gpu_name\",\"?\")} \${inst.get(\"dph_total\",0):.3f}/hr')
"

# 2. Check if affected instance has recent logs
echo -e "\n--- Recent Logs (last 20 lines) ---"
vastai logs ${INSTANCE_ID:-0} --tail 20 2>/dev/null || echo "No logs available"

# 3. Check account balance
echo -e "\n--- Account ---"
vastai show user --raw | python3 -c "import sys,json; u=json.load(sys.stdin); print(f'Balance: \${u.get(\"balance\",0):.2f}')"

Incident Type 1: Spot Preemption

Symptoms: Instance status changes from running to exited or offline without user action.

# 1. Verify preemption (not user error)
vastai show instance $ID --raw | python3 -c "
import sys, json; i=json.load(sys.stdin)
print(f'Status: {i.get(\"actual_status\")}')
print(f'Status msg: {i.get(\"status_msg\", \"none\")}')
"

# 2. Check if checkpoint was saved
# (depends on your checkpoint storage — S3, GCS, etc.)
aws s3 ls s3://bucket/checkpoints/ --recursive | tail -5

# 3. Provision replacement instance
vastai search offers "gpu_name=${GPU_NAME} reliability>0.98 rentable=true" \
  --order dph_total --limit 3

# 4. Create replacement and resume from checkpoint
vastai create instance $NEW_OFFER_ID --image $IMAGE --disk 50

Incident Type 2: Training Job Crash

Symptoms: Instance running but training process exited with error.

# 1. SSH in and check logs
ssh -p $PORT root@$HOST "tail -100 /workspace/train.log 2>/dev/null || echo 'No log file'"

# 2. Common causes
ssh -p $PORT root@$HOST << 'CHECK'
# GPU memory issue?
nvidia-smi | grep -i "out of memo

Vast.ai Install & Auth

Overview

Set up the Vast.ai CLI and REST API access for renting GPU compute instances. Vast.ai is a marketplace where individual hosts and data centers list GPU machines at prices significantly below hyperscaler providers.

Prerequisites

• Python 3.8+
• Vast.ai account at https://cloud.vast.ai
• Credit card or credits loaded for GPU rental

Instructions

Step 1: Install the CLI

set -euo pipefail
pip install vastai
vastai --version

Step 2: Get Your API Key

1. Log in at https://cloud.vast.ai
2. Navigate to Account > API Keys (or visit https://cloud.vast.ai/cli/)
3. Copy your API key (a long hexadecimal string)

Step 3: Configure Authentication

# Save API key to ~/.vast_api_key
vastai set api-key YOUR_API_KEY_HERE

# Verify authentication
vastai show user

For programmatic use, set the environment variable:

export VASTAI_API_KEY="your-api-key-here"
echo 'VASTAI_API_KEY=your-api-key' >> .env

Step 4: Verify with REST API

# Direct REST API call — base URL is cloud.vast.ai/api/v0
curl -s -H "Authorization: Bearer $VASTAI_API_KEY" \
  "https://cloud.vast.ai/api/v0/users/current" | jq '{id, username, balance}'

Step 5: Python Client Setup

# vastai_client.py
import os
import requests
from typing import Optional, Dict, Any, List

class VastClient:
    BASE_URL = "https://cloud.vast.ai/api/v0"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ.get("VASTAI_API_KEY")
        if not self.api_key:
            # Fall back to ~/.vast_api_key
            key_file = os.path.expanduser("~/.vast_api_key")
            if os.path.exists(key_file):
                self.api_key = open(key_file).read().strip()
        if not self.api_key:
            raise ValueError("No API key found. Run: vastai set api-key YOUR_KEY")

        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Accept": "application/json",
        })

    def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        url = f"{self.BASE_URL}{endpoint}"
        resp = self.session.request(method, url, **kwargs)
        resp.raise_for_status()
        return resp.json()

    def search_offers(self, query: Dict[str, Any]) -> List[Dict]:
        return self._request("GET", "/bundles/", params={"q": str(query)})

    def create_instance(self, offer_id: int, image: str, disk_gb:

Vast.ai Local Dev Loop

Overview

Set up a fast, reproducible local development workflow for Vast.ai GPU workloads. Test Docker images locally, mock API responses for CI, and minimize cloud GPU costs during development.

Prerequisites

• Completed vastai-install-auth setup
• Docker installed locally
• Python 3.8+ with pytest

Instructions

Step 1: Project Structure

vastai-project/
  src/
    vastai_client.py      # API client wrapper
    job_runner.py          # Job orchestration logic
    instance_manager.py   # Instance lifecycle management
  docker/
    Dockerfile            # GPU workload image
    requirements.txt      # Python dependencies for GPU job
  tests/
    test_client.py        # Unit tests with mocked API
    test_job_runner.py    # Integration tests
    conftest.py           # Shared fixtures and mocks
  scripts/
    test-connection.sh    # Quick API verification
    benchmark-gpu.py      # GPU benchmark script
  .env.development        # Dev API key (low spending limit)
  .env.production         # Prod API key (gitignored)

Step 2: Mock the Vast.ai API for Testing

# tests/conftest.py
import pytest
from unittest.mock import MagicMock

@pytest.fixture
def mock_vast_client():
    client = MagicMock()
    client.search_offers.return_value = {
        "offers": [
            {"id": 12345, "gpu_name": "RTX_4090", "gpu_ram": 24,
             "dph_total": 0.22, "reliability2": 0.99,
             "inet_down": 500, "ssh_host": "test.host", "ssh_port": 22},
        ]
    }
    client.create_instance.return_value = {"new_contract": 67890}
    client.show_instances.return_value = [
        {"id": 67890, "actual_status": "running",
         "ssh_host": "test.host", "ssh_port": 22}
    ]
    return client

Step 3: Test Docker Images Locally

# Build and test your GPU image locally (CPU mode)
docker build -t my-training:dev -f docker/Dockerfile .
docker run --rm my-training:dev python -c "import torch; print('OK')"

# Test training script in CPU mode
docker run --rm -v $(pwd)/data:/workspace/data my-training:dev \
  python train.py --epochs 1 --batch-size 4 --device cpu --dry-run

Step 4: Quick Connection Test Script

#!/bin/bash
set -euo pipefail
echo "Testing Vast.ai connection..."
vastai show user 2>/dev/null && echo "  CLI auth: OK" || echo "  CLI auth: FAIL"
BALANCE=$(vastai show user --raw 2>/dev/null | python3 -c "import sys,json; print(json.load(sys.stdin).get('balance',0))")
echo "  Balance: \$$BALANCE"
echo "Connec

Vast.ai Migration Deep Dive

Current State

!vastai --version 2>/dev/null || echo 'vastai CLI not installed'

!pip show vastai 2>/dev/null | grep Version || echo 'N/A'

Overview

Migrate GPU workloads to Vast.ai from hyperscaler providers (AWS, GCP, Azure) or other GPU clouds (Lambda, RunPod, CoreWeave). Also covers migrating between GPU types on Vast.ai and the reverse migration away from Vast.ai.

Prerequisites

• Existing GPU workload with Docker image
• Understanding of current GPU costs and utilization
• Checkpoint-based training pipeline (for training migrations)

Instructions

Step 1: Cost Comparison Analysis

# Compare your current GPU costs against Vast.ai marketplace prices
PROVIDER_COSTS = {
    "aws_p4d.24xlarge":      {"gpu": "A100 40GB", "gpus": 8, "hourly": 32.77},
    "aws_p3.2xlarge":        {"gpu": "V100 16GB", "gpus": 1, "hourly": 3.06},
    "gcp_a2-highgpu-1g":     {"gpu": "A100 40GB", "gpus": 1, "hourly": 3.67},
    "azure_NC24ads_A100_v4": {"gpu": "A100 80GB", "gpus": 1, "hourly": 3.67},
    "lambda_1xA100":         {"gpu": "A100",      "gpus": 1, "hourly": 1.25},
}

VASTAI_TYPICAL = {
    "RTX_4090":  0.20,
    "A100":      1.50,
    "H100_SXM":  3.00,
}

def savings_analysis(current_provider, current_hourly, vastai_gpu, vastai_hourly):
    monthly_current = current_hourly * 730  # hours/month
    monthly_vastai = vastai_hourly * 730
    savings = monthly_current - monthly_vastai
    pct = (savings / monthly_current) * 100
    print(f"Current ({current_provider}): ${monthly_current:,.0f}/mo")
    print(f"Vast.ai ({vastai_gpu}): ${monthly_vastai:,.0f}/mo")
    print(f"Savings: ${savings:,.0f}/mo ({pct:.0f}%)")

savings_analysis("AWS p3.2xlarge", 3.06, "RTX_4090", 0.20)
# Output: Savings: $2,088/mo (93%)

Step 2: Docker Image Migration

# Most Docker images work unchanged on Vast.ai
# Key differences:
# - Vast.ai instances run as root
# - /workspace is the default working directory
# - SSH access (not IAM roles) for authentication

# Adapt your existing Dockerfile
cat << 'DOCKERFILE' > Dockerfile.vastai
FROM your-existing-image:latest

# Vast.ai instances use /workspace by default
WORKDIR /workspace

# Install any Vast.ai-specific tools
RUN pip install boto3  # for S3 checkpoint uploads

# Copy training code
COPY src/ /workspace/src/
COPY configs/ /workspace/configs/

CMD ["python", "src/train.py"]
DOCKERFILE

docker build -t ghcr.io/org/training:vastai -f Dockerfil

Vast.ai Multi-Environment Setup

Overview

Configure separate Vast.ai environments for development, staging, and production by using different API keys, GPU profiles, and spending limits. Vast.ai does not have built-in environment isolation, so you implement it through configuration.

Prerequisites

• Vast.ai accounts or API keys per environment
• Secrets manager for key storage
• Understanding of GPU profile requirements per tier

Instructions

Step 1: Environment Configuration

# config.py — environment-specific Vast.ai settings
import os
from dataclasses import dataclass

@dataclass
class VastEnvConfig:
    name: str
    api_key: str
    max_dph: float           # Maximum $/hr per instance
    max_instances: int       # Concurrent instance limit
    max_daily_spend: float   # Daily budget cap
    gpu_whitelist: list      # Allowed GPU types
    reliability_min: float   # Minimum reliability score
    auto_destroy_hours: int  # Auto-destroy timeout

ENVIRONMENTS = {
    "development": VastEnvConfig(
        name="development",
        api_key=os.environ.get("VASTAI_DEV_KEY", ""),
        max_dph=0.25,
        max_instances=2,
        max_daily_spend=5.00,
        gpu_whitelist=["RTX_3090", "RTX_4090"],
        reliability_min=0.90,
        auto_destroy_hours=2,
    ),
    "staging": VastEnvConfig(
        name="staging",
        api_key=os.environ.get("VASTAI_STAGING_KEY", ""),
        max_dph=2.00,
        max_instances=4,
        max_daily_spend=50.00,
        gpu_whitelist=["RTX_4090", "A100"],
        reliability_min=0.95,
        auto_destroy_hours=12,
    ),
    "production": VastEnvConfig(
        name="production",
        api_key=os.environ.get("VASTAI_PROD_KEY", ""),
        max_dph=4.00,
        max_instances=16,
        max_daily_spend=500.00,
        gpu_whitelist=["A100", "H100_SXM"],
        reliability_min=0.98,
        auto_destroy_hours=48,
    ),
}

def get_config(env=None):
    env = env or os.environ.get("VASTAI_ENV", "development")
    return ENVIRONMENTS[env]

Step 2: Environment-Aware Client

class EnvAwareVastClient:
    def __init__(self, env="development"):
        self.config = get_config(env)
        self.client = VastClient(api_key=self.config.api_key)

    def search_offers(self, **overrides):
        query = {
            "rentable": {"eq": True},
            "reliability2": {"gte": self.config.reliability_min},
            "dph_total": {"lte": overrides.get("max_dph", self.config.max_dph)},
        }
        gpu = overrides.get("gpu_name", self.config.gpu_whitelist[0])
 

Vast.ai Observability

Overview

Monitor Vast.ai GPU instance health, utilization, and costs. Key metrics: GPU utilization (idle GPUs waste $0.20-$4.00/hr), instance uptime, training progress, cost accumulation, and spot preemption events.

Prerequisites

• Vast.ai account with active instances
• vastai CLI installed
• Optional: Prometheus, Grafana, or Datadog for dashboarding

Instructions

Step 1: Instance Metrics Collector

import subprocess, json, time
from datetime import datetime

class VastMetricsCollector:
    def __init__(self, output_file="vast_metrics.jsonl"):
        self.output_file = output_file

    def collect(self):
        result = subprocess.run(
            ["vastai", "show", "instances", "--raw"],
            capture_output=True, text=True)
        instances = json.loads(result.stdout)

        metrics = {
            "timestamp": datetime.utcnow().isoformat(),
            "total_instances": len(instances),
            "running": 0, "total_hourly_cost": 0,
            "instances": [],
        }

        for inst in instances:
            status = inst.get("actual_status", "unknown")
            dph = inst.get("dph_total", 0)
            if status == "running":
                metrics["running"] += 1
                metrics["total_hourly_cost"] += dph

            metrics["instances"].append({
                "id": inst["id"],
                "gpu": inst.get("gpu_name"),
                "status": status,
                "dph": dph,
                "gpu_util": inst.get("gpu_util", 0),
                "gpu_temp": inst.get("gpu_temp", 0),
            })

        with open(self.output_file, "a") as f:
            f.write(json.dumps(metrics) + "\n")

        return metrics

    def run(self, interval=60):
        while True:
            m = self.collect()
            print(f"[{m['timestamp']}] Running: {m['running']} | "
                  f"Cost: ${m['total_hourly_cost']:.3f}/hr")
            time.sleep(interval)

Step 2: Alert Conditions

def check_alerts(metrics):
    alerts = []

    # Idle GPU alert (running but <10% utilization)
    for inst in metrics["instances"]:
        if inst["status"] == "running" and inst["gpu_util"] < 10:
            alerts.append(f"IDLE: Instance {inst['id']} GPU util={inst['gpu_util']}% "
                         f"(wasting ${inst['dph']:.3f}/hr)")

    # High temperature alert
    for inst in metrics["instan

Vast.ai Performance Tuning

Overview

Optimize GPU instance selection, startup time, and training throughput on Vast.ai. Key levers: Docker image caching, GPU selection by dlperf score, data pipeline optimization, and multi-GPU scaling.

Prerequisites

• Vast.ai account with active or planned instances
• Understanding of GPU compute bottlenecks
• Profiling tools (nvidia-smi, torch.profiler)

Instructions

Step 1: Optimize Instance Selection by Performance

# Sort by dlperf (deep learning performance benchmark) instead of price
vastai search offers 'num_gpus=1 gpu_ram>=24 reliability>0.95' \
  --order 'dlperf-' --limit 10

# The dlperf field measures actual GPU compute throughput
# Higher dlperf = faster training even at same GPU model
# Variance within same GPU model can be 20-30%

def select_by_performance_per_dollar(offers):
    """Select the offer with best performance per dollar."""
    for o in offers:
        o["perf_per_dollar"] = o.get("dlperf", 0) / max(o["dph_total"], 0.01)
    return max(offers, key=lambda o: o["perf_per_dollar"])

Step 2: Reduce Instance Startup Time

# Use smaller, pre-cached Docker images
# FAST: nvidia/cuda:12.1.1-runtime-ubuntu22.04 (~2GB, widely cached)
# MEDIUM: pytorch/pytorch:2.2.0-cuda12.1-cudnn8-runtime (~4GB)
# SLOW: custom-image:latest with pip install at build (~10GB+)

# Pre-install deps in the image, not in onstart
# BAD (slow startup):
vastai create instance $ID --image pytorch/pytorch:latest \
  --onstart-cmd "pip install transformers datasets wandb"

# GOOD (fast startup):
# Build custom image with all deps pre-installed

Step 3: Data Pipeline Optimization

# Profile GPU utilization on the instance
# SSH into instance and run:
"""
watch -n 1 nvidia-smi  # Check if GPU util is <80% → data bottleneck

# Common fixes for low GPU utilization:
# 1. Increase DataLoader num_workers
# 2. Use pin_memory=True
# 3. Pre-fetch data to local SSD (not NFS)
# 4. Use WebDataset or FFCV for streaming datasets
"""

# Optimize PyTorch DataLoader
from torch.utils.data import DataLoader

loader = DataLoader(
    dataset,
    batch_size=32,
    num_workers=4,       # Match CPU cores on instance
    pin_memory=True,     # Faster GPU transfer
    prefetch_factor=2,   # Pre-load 2 batches per worker
    persistent_workers=True,  # Don't respawn workers each epoch
)

Step 4: GPU Memory Optimization

# Check available VRAM before selecting batch size
import torch

def optimal_batch_size(model, sample_input, gpu_memory_gb):
    """Binary search for largest batch size that fits in

Vast.ai Production Checklist

Overview

Complete checklist for running production GPU workloads on Vast.ai, covering account setup, instance selection, data safety, monitoring, and cost controls.

Prerequisites

• Vast.ai account with sufficient credits
• Docker images tested and published to registry
• Checkpoint-based training pipeline

Instructions

Account & Authentication

• [ ] API key stored in secrets manager (not in code or env files)
• [ ] Dedicated SSH key pair for Vast.ai (not shared with other services)
• [ ] Account balance sufficient for planned workload duration + 50% buffer
• [ ] Billing alerts configured at cloud.vast.ai

Instance Selection

• [ ] GPU type validated for workload (VRAM, compute capability)
• [ ] Reliability filter set to >= 0.98 for production jobs
• [ ] Internet speed filter set to inet_down >= 200 for data transfer
• [ ] Disk allocation includes room for checkpoints + data + 20% overhead
• [ ] CUDA version on host matches Docker image requirements

Data Safety

• [ ] Training data encrypted before upload to instances
• [ ] Checkpoint saving every N steps (not just per epoch)
• [ ] Checkpoints uploaded to persistent storage (S3/GCS) periodically
• [ ] Instance cleanup script removes data before destruction
• [ ] No sensitive data (API keys, PII) embedded in Docker images

Spot Instance Protection

• [ ] Spot preemption handler implemented (save checkpoint on SIGTERM)
• [ ] Auto-recovery: detect destroyed instance, provision replacement, resume
• [ ] On-demand fallback configured for critical final training stages
• [ ] Checkpoint integrity verification after recovery

Monitoring & Alerting

• [ ] GPU utilization monitoring (alert if < 50% for > 10 min)
• [ ] Instance health polling every 60 seconds
• [ ] Cost accumulation tracking with budget threshold alerts
• [ ] Training loss/metrics logged to external service (W&B, MLflow)
• [ ] Dead instance detection (auto-destroy stuck instances)

Cost Controls

• [ ] Maximum dph_total set in search queries
• [ ] Auto-destroy timeout for all instances (e.g., 24h max)
• [ ] Daily spending limit configured
• [ ] Cost-per-job tracking for budget reporting

Verification Script

#!/bin/bash
set -euo pipefail
echo "Vast.ai Production Readiness Check"

# 1. Auth
vastai show user --raw | python3 -c "
import sys, json; u=json.load(sys.stdin)
balance = u.get('balance', 0)
print(f'  Auth: OK | Balance: \${balance:.2f}')
assert balance >= 10, f'Balance too low: \${balance:.2f}'
" && echo "  

Vast.ai Rate Limits

Overview

Handle Vast.ai REST API rate limits gracefully. The API at cloud.vast.ai/api/v0 returns HTTP 429 when request limits are exceeded. Most operations (search, show) are read-heavy and rarely hit limits, but automated scripts doing rapid provisioning or polling can trigger throttling.

Prerequisites

• Vast.ai CLI or REST API client
• Understanding of exponential backoff

Instructions

Step 1: Rate-Limited HTTP Client

import requests
import time

class RateLimitedVastClient:
    BASE_URL = "https://cloud.vast.ai/api/v0"

    def __init__(self, api_key, min_delay=0.5, max_retries=5):
        self.session = requests.Session()
        self.session.headers["Authorization"] = f"Bearer {api_key}"
        self.min_delay = min_delay
        self.max_retries = max_retries
        self.last_request = 0

    def request(self, method, endpoint, **kwargs):
        # Enforce minimum delay between requests
        elapsed = time.time() - self.last_request
        if elapsed < self.min_delay:
            time.sleep(self.min_delay - elapsed)

        for attempt in range(self.max_retries):
            self.last_request = time.time()
            resp = self.session.request(method, f"{self.BASE_URL}{endpoint}", **kwargs)

            if resp.status_code == 429:
                retry_after = int(resp.headers.get("Retry-After", 2 ** attempt))
                print(f"Rate limited. Waiting {retry_after}s (attempt {attempt+1})")
                time.sleep(retry_after)
                continue

            resp.raise_for_status()
            return resp.json()

        raise RuntimeError("Max retries exceeded due to rate limiting")

Step 2: Polling with Adaptive Backoff

def poll_instance_status(client, instance_id, target="running", timeout=300):
    """Poll instance status with increasing intervals."""
    start = time.time()
    interval = 5  # Start at 5s, increase to max 30s

    while time.time() - start < timeout:
        info = client.request("GET", f"/instances/{instance_id}/")
        status = info.get("actual_status", "unknown")

        if status == target:
            return info
        if status in ("error", "offline"):
            raise RuntimeError(f"Instance {instance_id} failed: {status}")

        time.sleep(interval)
        interval = min(interval * 1.5, 30)

    raise TimeoutError(f"Instance did not reach '{target}' within {timeout}s")

Step 3: Batch Search with Throttling

def batch_search(client, gpu_configs):
    """Search for multiple GPU types with rate-limit-safe delays.""&quo

Vast.ai Reference Architecture

Overview

Production architecture for GPU compute workflows on Vast.ai. Covers the three-tier pattern (orchestrator, GPU workers, artifact storage), job queue design, and fault-tolerant training pipelines.

Prerequisites

• Vast.ai account with CLI
• Cloud storage (S3, GCS, or MinIO) for artifacts
• Understanding of ML training pipelines

Instructions

Architecture: Three-Tier GPU Compute

┌─────────────────────────────────────────────────┐
│  ORCHESTRATOR (your server / CI / cloud function) │
│  - Job queue management                          │
│  - Instance provisioning via Vast.ai API         │
│  - Status monitoring and auto-recovery           │
│  - Cost tracking and budget enforcement          │
└───────────────┬─────────────────────────────────┘
                │ Vast.ai REST API
┌───────────────▼─────────────────────────────────┐
│  GPU WORKERS (Vast.ai rented instances)          │
│  - Training / inference execution                │
│  - Checkpoint saving to cloud storage            │
│  - Health reporting back to orchestrator         │
│  - Graceful shutdown on SIGTERM (spot preemption)│
└───────────────┬─────────────────────────────────┘
                │ S3 / GCS / MinIO
┌───────────────▼─────────────────────────────────┐
│  ARTIFACT STORAGE (persistent)                   │
│  - Model checkpoints                             │
│  - Training logs and metrics                     │
│  - Dataset cache                                 │
│  - Final model artifacts                         │
└─────────────────────────────────────────────────┘

Project Structure

ml-pipeline/
  orchestrator/
    job_queue.py         # Job definition and scheduling
    provisioner.py       # Vast.ai instance lifecycle
    monitor.py           # Status polling and auto-recovery
    cost_tracker.py      # Budget enforcement
  worker/
    Dockerfile           # GPU worker image
    train.py             # Training entry point
    checkpoint.py        # Cloud storage checkpoint manager
    health.py            # Report status back to orchestrator
  config/
    gpu_profiles.yaml    # GPU selection criteria per job type
    budgets.yaml         # Cost limits per team/project
  scripts/
    deploy.py            # CLI for launching jobs
    cost_report.py       # Spending analysis

GPU Profile Configuration

# config/gpu_profiles.yaml
profiles:
  dev-test:
    gpu_name: RTX_4090
    num_gpus: 1
    max_dph: 0.25
    reliability_min: 0.90
    max_duration_hours: 2

  training-standard:
    gpu_name: A100
    num_gpus: 1
    max_dph: 2.00
    reliability_min: 0.98
    max_duration_hours: 24

  training-distributed:
    gpu_name: H100_SXM
    num_gpus: 4
    max_dph: 4.00
    reliability_min: 0.99
    max_duration_hours: 48

  inference-batch:
    gpu_name: RTX_4090
    num_

Vast.ai SDK Patterns

Overview

Production-ready patterns for the Vast.ai CLI, Python SDK, and REST API at cloud.vast.ai/api/v0. Covers typed search queries, instance lifecycle management, offer scoring, and error handling.

Prerequisites

• Completed vastai-install-auth setup
• Python 3.8+ with requests
• Familiarity with the Vast.ai marketplace model

Instructions

Pattern 1: Typed Search Query Builder

from dataclasses import dataclass
from typing import Optional

@dataclass
class GPUQuery:
    num_gpus: int = 1
    gpu_name: Optional[str] = None
    gpu_ram_min: Optional[float] = None
    reliability_min: float = 0.95
    max_dph: Optional[float] = None

    def to_filter(self) -> dict:
        f = {"rentable": {"eq": True}, "num_gpus": {"eq": self.num_gpus},
             "reliability2": {"gte": self.reliability_min}}
        if self.gpu_name:
            f["gpu_name"] = {"eq": self.gpu_name}
        if self.gpu_ram_min:
            f["gpu_ram"] = {"gte": self.gpu_ram_min}
        if self.max_dph:
            f["dph_total"] = {"lte": self.max_dph}
        return f

Pattern 2: Context-Managed Instance Lifecycle

from contextlib import contextmanager

@contextmanager
def managed_instance(client, offer_id, image, disk_gb=20, timeout=300):
    """Auto-destroy instance on exit or exception."""
    inst = client.create_instance(offer_id, image, disk_gb)
    instance_id = inst["new_contract"]
    try:
        info = client.poll_until_running(instance_id, timeout)
        yield info
    finally:
        client.destroy_instance(instance_id)

# Usage
with managed_instance(client, offer["id"], "pytorch/pytorch:latest") as inst:
    ssh_exec(inst["ssh_host"], inst["ssh_port"], "python train.py")

Pattern 3: Offer Scoring

def score_offer(offer, weights=None):
    w = weights or {"cost": 0.4, "reliability": 0.3, "perf": 0.3}
    return (w["cost"] * (1.0 / max(offer["dph_total"], 0.01)) +
            w["reliability"] * offer.get("reliability2", 0) * 100 +
            w["perf"] * offer.get("dlperf", 0))

best = max(offers, key=score_offer)

Pattern 4: Retry with Backoff

import time
from functools import wraps

def retry(max_attempts=3, backoff=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except

Vast.ai Security Basics

Overview

Security best practices for Vast.ai API keys, SSH access to GPU instances, data protection on rented hardware, and credential management. Vast.ai instances run as root on shared hardware, requiring careful attention to data lifecycle.

Prerequisites

• Vast.ai account with API key
• Understanding of SSH key management
• Secrets manager available (optional but recommended)

Instructions

Step 1: API Key Management

# Never commit API keys to git
echo '.vast_api_key' >> .gitignore
echo '.env' >> .gitignore

# Use environment variables, not files in repos
export VASTAI_API_KEY="$(vault kv get -field=api_key secret/vastai)"

# Rotate keys periodically at cloud.vast.ai > Account > API Keys

# Fail fast on missing credentials
import os

def get_api_key():
    key = os.environ.get("VASTAI_API_KEY")
    if not key:
        key_file = os.path.expanduser("~/.vast_api_key")
        if os.path.exists(key_file):
            key = open(key_file).read().strip()
    if not key:
        raise ValueError("VASTAI_API_KEY not set and ~/.vast_api_key not found")
    return key

Step 2: SSH Key Security

# Generate a dedicated key pair for Vast.ai instances
ssh-keygen -t ed25519 -f ~/.ssh/vastai_key -C "vastai-instances" -N ""

# Upload public key at cloud.vast.ai > Account > SSH Keys

# Use the dedicated key for connections
ssh -i ~/.ssh/vastai_key -p PORT root@HOST

Step 3: Data Protection on Shared Hardware

def secure_cleanup(instance_id, ssh_host, ssh_port):
    """Securely wipe data before destroying an instance."""
    import subprocess
    # Overwrite sensitive files before instance destruction
    subprocess.run([
        "ssh", "-p", str(ssh_port), "-o", "StrictHostKeyChecking=no",
        f"root@{ssh_host}",
        "rm -rf /workspace/data /workspace/checkpoints /root/.ssh/authorized_keys; "
        "history -c"
    ], check=True)
    # Then destroy
    subprocess.run(["vastai", "destroy", "instance", str(instance_id)], check=True)

Step 4: Network Security

• Use SSH tunnels for any services exposed on instances
• Never expose ports with sensitive data to the public internet
• Transfer data over SCP/SFTP, not unencrypted HTTP
• Encrypt training data before upload; decrypt on-instance

Step 5: Credential Rotation Checklist

• [ ] API key rotated every 90 days
• [ ] SSH keys dedicated to Vast.ai (not shared with production)
• [ ] Old SSH keys removed from cloud.vas

Vast.ai Webhooks & Events

Overview

Build event-driven workflows around Vast.ai GPU instance lifecycle. Vast.ai does not provide traditional webhooks, so event detection relies on polling the REST API at cloud.vast.ai/api/v0 and reacting to instance status transitions (loading, running, exited, error, offline).

Prerequisites

• Vast.ai CLI authenticated
• Understanding of instance lifecycle states
• Python 3.8+ for event loop implementation

Instructions

Step 1: Instance Status Poller

import time, json, subprocess
from typing import Callable, Dict, List

class InstanceEventPoller:
    """Poll Vast.ai API and emit events on status transitions."""

    def __init__(self, api_key: str, poll_interval: int = 30):
        self.api_key = api_key
        self.poll_interval = poll_interval
        self.previous_states: Dict[int, str] = {}
        self.handlers: Dict[str, List[Callable]] = {}

    def on(self, event: str, handler: Callable):
        self.handlers.setdefault(event, []).append(handler)

    def poll_once(self):
        result = subprocess.run(
            ["vastai", "show", "instances", "--raw"],
            capture_output=True, text=True)
        instances = json.loads(result.stdout)

        for inst in instances:
            inst_id = inst["id"]
            status = inst.get("actual_status", "unknown")
            prev = self.previous_states.get(inst_id)

            if prev and prev != status:
                event = f"{prev}_to_{status}"
                for handler in self.handlers.get(event, []):
                    handler(inst)
                for handler in self.handlers.get("any_change", []):
                    handler(inst, prev, status)

            self.previous_states[inst_id] = status

    def run(self):
        print(f"Polling every {self.poll_interval}s...")
        while True:
            self.poll_once()
            time.sleep(self.poll_interval)

Step 2: Event Handlers

def on_instance_running(instance):
    print(f"Instance {instance['id']} is RUNNING")
    print(f"  SSH: ssh -p {instance['ssh_port']} root@{instance['ssh_host']}")
    # Trigger: start training job, send notification, etc.

def on_instance_exited(instance):
    print(f"Instance {instance['id']} EXITED")
    # Trigger: collect results, check for errors, notify team

def on_spot_preemption(instance, old_status, new_status):
    if old_status == "running" and new_status in ("exited", "offline"):
        print(f"ALERT: Instance {instance['id']} may have been preempted")
        # Trigger: auto-recovery, provision replacement

# Wire up handlers
poll