vLLM Monitoring And Troubleshooting For MangaAssist

graph TD
    subgraph "vLLM Engine"
        VM1["Built-in Prometheus metrics\n(:9090/metrics)"]
    end

    subgraph "vLLM Gateway (Application)"
        AM1["Custom inference metrics"]
        AM2["Trace attributes (MLflow)"]
    end

    subgraph "Collection Layer"
        PROM["Prometheus\n(sidecar or remote)"]
        CWA["CloudWatch Agent"]
    end

    subgraph "Storage and Query"
        CW["CloudWatch Metrics"]
        MLF["MLflow Tracking Server"]
    end

    subgraph "Consumption"
        DASH["CloudWatch Dashboard:\nMangaAssist Inference"]
        ALERT["CloudWatch Alarms"]
        GRAFANA["Grafana\n(optional, for Prometheus native)"]
    end

    VM1 --> PROM
    AM1 --> CWA
    AM2 --> MLF
    PROM --> CW
    CWA --> CW
    CW --> DASH
    CW --> ALERT
    PROM --> GRAFANA

"""
Custom inference metrics for the MangaAssist vLLM gateway.
Emitted to CloudWatch via the CloudWatch Agent.
"""

import time
from dataclasses import dataclass, field

import boto3


@dataclass
class InferenceMetrics:
    namespace: str = "MangaAssist/Inference"
    _cw_client: object = field(default=None, repr=False)

    def __post_init__(self) -> None:
        self._cw_client = boto3.client("cloudwatch")

    def emit_request_metrics(
        self,
        request_id: str,
        queue_wait_ms: float,
        ttft_ms: float,
        generation_ms: float,
        input_tokens: int,
        output_tokens: int,
        adapter_id: str,
        backend_version: str,
        prefix_cache_hit: bool,
        token_budget_trimmed: bool,
    ) -> None:
        metrics = [
            self._metric("QueueWaitMs", queue_wait_ms, "Milliseconds"),
            self._metric("TTFTMs", ttft_ms, "Milliseconds"),
            self._metric("GenerationMs", generation_ms, "Milliseconds"),
            self._metric("InputTokens", input_tokens, "Count"),
            self._metric("OutputTokens", output_tokens, "Count"),
            self._metric("PrefixCacheHit", 1.0 if prefix_cache_hit else 0.0, "Count"),
            self._metric("TokenBudgetTrimmed", 1.0 if token_budget_trimmed else 0.0, "Count"),
        ]

        dimensions = [
            {"Name": "AdapterId", "Value": adapter_id},
            {"Name": "BackendVersion", "Value": backend_version},
        ]

        self._cw_client.put_metric_data(
            Namespace=self.namespace,
            MetricData=[{**m, "Dimensions": dimensions} for m in metrics],
        )

    def emit_oom_caught(self, adapter_id: str) -> None:
        self._cw_client.put_metric_data(
            Namespace=self.namespace,
            MetricData=[{
                "MetricName": "OOMCaught",
                "Value": 1.0,
                "Unit": "Count",
                "Dimensions": [{"Name": "AdapterId", "Value": adapter_id}],
            }],
        )

    def emit_admission_rejected(self, reason: str) -> None:
        self._cw_client.put_metric_data(
            Namespace=self.namespace,
            MetricData=[{
                "MetricName": "AdmissionRejected",
                "Value": 1.0,
                "Unit": "Count",
                "Dimensions": [{"Name": "Reason", "Value": reason}],
            }],
        )

    @staticmethod
    def _metric(name: str, value: float, unit: str) -> dict:
        return {"MetricName": name, "Value": value, "Unit": unit}

import mlflow


def trace_inference_request(
    request_id: str,
    session_id: str,
    adapter_id: str,
    backend_version: str,
    prompt_version: str,
    input_tokens: int,
    output_tokens: int,
    queue_wait_ms: float,
    ttft_ms: float,
    generation_ms: float,
    prefix_cache_hit: bool,
    finish_reason: str,
) -> None:
    with mlflow.start_span(name="vllm_inference") as span:
        span.set_attributes({
            "request_id": request_id,
            "session_id": session_id,
            "adapter_id": adapter_id,
            "backend_version": backend_version,
            "prompt_version": prompt_version,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "queue_wait_ms": queue_wait_ms,
            "ttft_ms": ttft_ms,
            "generation_ms": generation_ms,
            "prefix_cache_hit": prefix_cache_hit,
            "finish_reason": finish_reason,
        })

import boto3

cw_client = boto3.client("cloudwatch")


def create_critical_alarms() -> None:
    # Queue overflow alarm
    cw_client.put_metric_alarm(
        AlarmName="mangaassist-vllm-queue-overflow",
        Namespace="MangaAssist/Inference",
        MetricName="vllm:num_requests_waiting",
        Statistic="Maximum",
        Period=60,
        EvaluationPeriods=2,
        Threshold=80,
        ComparisonOperator="GreaterThanThreshold",
        AlarmActions=["arn:aws:sns:us-east-1:ACCOUNT:mangaassist-critical"],
        TreatMissingData="notBreaching",
    )

    # OOM alarm
    cw_client.put_metric_alarm(
        AlarmName="mangaassist-vllm-oom-detected",
        Namespace="MangaAssist/Inference",
        MetricName="OOMCaught",
        Statistic="Sum",
        Period=300,
        EvaluationPeriods=1,
        Threshold=0,
        ComparisonOperator="GreaterThanThreshold",
        AlarmActions=["arn:aws:sns:us-east-1:ACCOUNT:mangaassist-critical"],
        TreatMissingData="notBreaching",
    )

    # High TTFT alarm
    cw_client.put_metric_alarm(
        AlarmName="mangaassist-vllm-high-ttft",
        Namespace="MangaAssist/Inference",
        MetricName="TTFTMs",
        ExtendedStatistic="p95",
        Period=300,
        EvaluationPeriods=1,
        Threshold=500,
        ComparisonOperator="GreaterThanThreshold",
        AlarmActions=["arn:aws:sns:us-east-1:ACCOUNT:mangaassist-warning"],
        TreatMissingData="notBreaching",
    )

    # Prefix cache degradation
    cw_client.put_metric_alarm(
        AlarmName="mangaassist-vllm-prefix-cache-degradation",
        Namespace="MangaAssist/Inference",
        MetricName="PrefixCacheHit",
        Statistic="Average",
        Period=900,  # 15 min
        EvaluationPeriods=1,
        Threshold=0.5,
        ComparisonOperator="LessThanThreshold",
        AlarmActions=["arn:aws:sns:us-east-1:ACCOUNT:mangaassist-warning"],
        TreatMissingData="notBreaching",
    )

Monthly error budget at 99.9% availability:
  Total minutes:    43,200 (30 days × 24 hours × 60 min)
  Budget:           43.2 minutes of downtime
  Burn rate alert:  If we consume 1% of budget (0.43 min = 26 seconds) in 1 hour,
                    we are on pace to exhaust the month's budget in 4.2 days.

Symptoms:
  - Users report "the bot takes a while to start answering"
  - TTFT P95 alarm fires

Diagnostic Steps:
  1. Check prefix cache hit rate
     → If < 50%: A prompt template change likely broke cache determinism
        → Check recent prompt version deployments
        → Verify no timestamps/random IDs in the cacheable prefix
        → Fix: Revert prompt change or move volatile fields below the prefix boundary

  2. If cache hit rate is normal, check queue wait time
     → If queue_wait_ms P95 > 200ms: The engine is saturated
        → Check active_sequences vs max_num_seqs
        → Check if traffic spiked (admission controller should be rejecting)
        → Fix: Scale out (add instances) or strengthen admission control

  3. If queue and cache are normal, check input token distribution
     → If average input_tokens has increased: Prompts got bigger
        → Check if retrieval is returning more chunks than expected
        → Check if conversation history is not being properly summarized
        → Fix: Review context budgeting settings, enforce token limits

  4. If all above are normal, check GPU health
     → Run: nvidia-smi -q | grep -i "ecc\|retired\|performance"
     → If GPU reports degraded performance: Replace instance

Symptoms:
  - OOMCaught alarm fires
  - Some responses return degraded/fallback content

Diagnostic Steps:
  1. Check gpu_cache_usage_perc at time of OOM
     → If consistently > 95%: Memory configuration too aggressive
        → Consider: Reduce max_num_seqs from 128 to 112
        → Consider: Reduce gpu_memory_utilization from 0.92 to 0.90

  2. Check input_token distribution at time of OOM
     → If spiky (occasional very large prompts): Context budgeting bypass
        → Check if request_budgeter is being called before engine
        → Check if a new code path bypasses the budgeter
        → Fix: Ensure all paths go through token budget allocation

  3. Check adapter_id correlation
     → If OOMs correlate with a specific adapter: Adapter too large
        → Check adapter rank and weight size
        → Fix: Reduce LoRA rank or use a smaller adapter

  4. If no single cause: Check for memory leak
     → Monitor gpu_cache_usage_perc over hours
     → If it trends upward without traffic increase: Possible block leak
        → Fix: Rolling restart of instances (schedule during low traffic)

Symptoms:
  - PrefixCacheHit alarm fires (< 50% for 15 min)
  - Efficiency drops, TTFT may increase
  - No obvious errors or failures

Diagnostic Steps:
  1. Check recent deployments
     → Check prompt_version in recent traces
     → If prompt version changed: Verify new prompt structure
        → Compare prefix bytes of old vs new prompt
        → Look for: timestamps, request IDs, A/B test flags in the prefix

  2. If no deployment, check prompt construction code
     → Look for: new personalization fields injected before the prefix boundary
     → Look for: dynamic policy blocks that change frequently
     → Fix: Move volatile content after the deterministic prefix boundary

  3. Check traffic pattern
     → If traffic is very low (< 10 req/min): Cache entries may be evicted
        → This is normal during off-peak hours
        → Cache hit rate should recover when traffic increases

  4. Check vLLM version
     → vLLM 0.4.1 → 0.4.2 changed eviction behavior (known issue)
     → Fix: Upgrade to 0.4.3 which fixed the eviction bug

Symptoms:
  - One adapter's P95 latency is 2×+ its historical baseline
  - Other adapters perform normally

Diagnostic Steps:
  1. Check if adapter was recently updated
     → Compare adapter version in traces vs the last known-good version
     → If updated: Run offline evaluation against the previous version
        → If regression confirmed: Rollback adapter in the registry
        → If evaluation is clean: Issue is traffic-pattern, not adapter

  2. Check adapter weight size
     → If new adapter is larger (higher rank): More memory per request
        → Check if preemption rate increased for this adapter
        → Fix: Reduce rank or increase GPU headroom

  3. Check request pattern for this adapter
     → If this adapter handles longer prompts than others: Expected
        → Verify context budgeting is applied equally across adapters

Symptoms:
  - Scaling alarm fires but instance count does not increase
  - Or new instances start but /ping fails for > 5 minutes

Diagnostic Steps:
  1. Check SageMaker endpoint events
     → Look for: InsufficientInstanceCapacity
        → If capacity issue: Try a different availability zone
        → If persistent: Add ml.g5.2xlarge as a fallback instance type

  2. Check warm pool status
     → If warm pool empty and cold start required: Normal 5-8 min wait
        → Fix: Ensure warm pool has at least 1 instance pre-provisioned

  3. If instances start but fail health checks
     → Check container logs for startup errors
        → Common: Model file corruption, CUDA version mismatch
        → Fix: Rebuild and re-push Docker image

Symptoms:
  - After upgrading vLLM version, throughput or latency regressed
  - No configuration changes made

Diagnostic Steps:
  1. Run benchmark comparison: new version vs old version on identical traffic
     → Use the shadow comparison job in eval/shadow_compare.py
     → Compare: TTFT, generation_ms, queue_wait_ms, prefix_cache_hit_rate

  2. Check for breaking changes in vLLM release notes
     → Prefix caching: Eviction behavior changed in 0.4.2
     → Multi-LoRA: Adapter loading semantics changed in 0.4.3
     → Block size: Internal allocation changed in 0.5.0

  3. Check CUDA graph behavior
     → If enforce_eager=False, new version may capture different graphs
     → Try: Set enforce_eager=True temporarily to test
     → If performance recovers with eager mode: Graph capture issue

  4. Rollback procedure
     → Revert Docker image tag to previous version
     → Redeploy endpoint (takes ~5 min with warm pool)
     → Monitor for 30 min to confirm regression is gone

# GPU status and utilization
nvidia-smi
nvidia-smi --query-gpu=utilization.gpu,utilization.memory,memory.used,memory.total --format=csv -l 1

# Check for GPU errors
nvidia-smi -q | grep -i "ecc\|retired\|performance\|error"

# vLLM engine status
curl http://localhost:8080/health
curl http://localhost:9090/metrics | grep vllm

# Active sequences and queue depth
curl -s http://localhost:9090/metrics | grep "num_requests"

# GPU cache usage
curl -s http://localhost:9090/metrics | grep "gpu_cache_usage"

# Prefix cache hit rate (custom metric)
curl -s http://localhost:9090/metrics | grep "prefix_cache"

# CUDA memory summary (from Python)
python3 -c "import torch; print(torch.cuda.memory_summary())"

# Check model loading status
curl -s http://localhost:8080/v1/models | python3 -m json.tool

# Test a single inference request
curl -X POST http://localhost:8080/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "manga_domain_v3",
    "messages": [{"role": "user", "content": "test"}],
    "max_tokens": 5
  }'

"""
Load test for MangaAssist vLLM endpoint.
Simulates realistic chatbot traffic patterns.
"""

import asyncio
import time
import random
import statistics
from dataclasses import dataclass

import httpx


@dataclass
class LoadTestConfig:
    endpoint: str
    target_rps: float           # Requests per second
    duration_seconds: int       # Total test duration
    warmup_seconds: int = 30    # Gradual ramp-up period
    max_concurrent: int = 100   # Max concurrent requests


@dataclass
class RequestResult:
    ttft_ms: float
    total_ms: float
    input_tokens: int
    output_tokens: int
    status_code: int
    adapter_id: str


# Realistic MangaAssist request distribution
REQUESTS = [
    # Short factual (30% of traffic)
    {"weight": 0.30, "adapter": "manga_domain_v3", "msg": "Is Spy x Family volume 12 available in English?", "max_tokens": 30},
    # Medium recommendation (40% of traffic)
    {"weight": 0.40, "adapter": "manga_domain_v3", "msg": "I liked Vinland Saga. What should I read next?", "max_tokens": 200},
    # Long grounded answer (20% of traffic)
    {"weight": 0.20, "adapter": "manga_domain_v3", "msg": "Compare all editions of Berserk volume 1.", "max_tokens": 400},
    # Support question (10% of traffic)
    {"weight": 0.10, "adapter": "general_support_v2", "msg": "Where is my order?", "max_tokens": 100},
]

SYSTEM_PROMPT = "You are a manga shopping assistant for the JP Manga Store."


def pick_request() -> dict:
    r = random.random()
    cumulative = 0.0
    for req in REQUESTS:
        cumulative += req["weight"]
        if r <= cumulative:
            return req
    return REQUESTS[-1]


async def send_request(client: httpx.AsyncClient, endpoint: str) -> RequestResult:
    req = pick_request()
    start = time.monotonic()

    payload = {
        "model": req["adapter"],
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": req["msg"]},
        ],
        "max_tokens": req["max_tokens"],
        "stream": True,
    }

    ttft = None
    total_tokens = 0

    async with client.stream("POST", endpoint, json=payload) as response:
        async for line in response.aiter_lines():
            if ttft is None and line.startswith("data:") and '"delta"' in line:
                ttft = (time.monotonic() - start) * 1000
            if '"finish_reason"' in line:
                break
            if line.startswith("data:") and '"delta"' in line:
                total_tokens += 1

    total_ms = (time.monotonic() - start) * 1000

    return RequestResult(
        ttft_ms=ttft or total_ms,
        total_ms=total_ms,
        input_tokens=len(SYSTEM_PROMPT.split()) + len(req["msg"].split()),
        output_tokens=total_tokens,
        status_code=response.status_code,
        adapter_id=req["adapter"],
    )


async def run_load_test(config: LoadTestConfig) -> None:
    results: list[RequestResult] = []
    errors = 0

    async with httpx.AsyncClient(timeout=30.0) as client:
        sem = asyncio.Semaphore(config.max_concurrent)

        async def bounded_request() -> None:
            nonlocal errors
            async with sem:
                try:
                    result = await send_request(client, f"{config.endpoint}/v1/chat/completions")
                    results.append(result)
                except Exception:
                    errors += 1

        tasks = []
        start_time = time.monotonic()

        for i in range(int(config.target_rps * config.duration_seconds)):
            elapsed = time.monotonic() - start_time
            target_time = i / config.target_rps

            # Gradual ramp-up during warmup period
            if elapsed < config.warmup_seconds:
                ramp_factor = elapsed / config.warmup_seconds
                target_time = target_time / max(ramp_factor, 0.1)

            if target_time > elapsed:
                await asyncio.sleep(target_time - elapsed)

            tasks.append(asyncio.create_task(bounded_request()))

        await asyncio.gather(*tasks)

    # Report
    if results:
        ttfts = [r.ttft_ms for r in results]
        totals = [r.total_ms for r in results]
        print(f"\n{'='*60}")
        print(f"Load Test Results ({len(results)} successful, {errors} errors)")
        print(f"{'='*60}")
        print(f"TTFT   P50: {statistics.median(ttfts):.0f} ms")
        print(f"TTFT   P95: {sorted(ttfts)[int(len(ttfts)*0.95)]:.0f} ms")
        print(f"TTFT   P99: {sorted(ttfts)[int(len(ttfts)*0.99)]:.0f} ms")
        print(f"Total  P50: {statistics.median(totals):.0f} ms")
        print(f"Total  P95: {sorted(totals)[int(len(totals)*0.95)]:.0f} ms")
        print(f"Total  P99: {sorted(totals)[int(len(totals)*0.99)]:.0f} ms")
        print(f"Error rate: {errors / (len(results) + errors) * 100:.2f}%")