#!/usr/bin/env python3
"""
O11Y Cluster Inspection - Generate a comprehensive health report.

Usage:
    python skills/o11y-data/o11y-inspect.py <cluster_id> --provider <aws|gcp|alicloud>
        [--hours N] [--region REGION] [--output FILE]

Requires:
    - GITHUB_PAT in environment, .env, or ~/.observability/.github
    - boto3, requests, duckdb (pip install -r requirements.txt)

The script will:
    1. Authenticate via GitHub PAT with provider-specific credentials
    2. Locate the cluster across regions/bizTypes
    3. Collect Prometheus metrics (QPS, latency, connections, TiKV CPU, regions)
    4. Download and analyze log-stats (ERROR/WARN counts, top errors)
    5. Query Delta Lake slowlogs (top slow queries, P99, count)
    6. Query Loki for recent log entries
    7. Generate a JSON report + human-readable summary
"""
import argparse
import json
import os
import re
import subprocess
import sys
import tempfile
import sqlite3
import time
from datetime import datetime, timezone, timedelta
import requests

REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

DS_AUTH_URL = "https://www.ds.us-west-2.aws.observability.tidbcloud.com/client/auth"
GS_API_BASE = "https://www.gs.us-west-2.aws.observability.tidbcloud.com/api/v1/clusters/"

VENDOR_REGIONS = {
    "aws": ["us-east-1", "us-west-2", "ap-east-1", "ap-northeast-1", "ap-southeast-1", "eu-central-1", "eu-west-1"],
    "gcp": ["us-east4", "asia-northeast1", "asia-southeast1", "us-west1"],
    "alicloud": ["us-east-1", "ap-northeast-1", "ap-southeast-1", "cn-hongkong", "ap-southeast-5"],
}
SCAN_BIZ_TYPES = ["tidbcloud", "prod-tidbcloud-nextgen", "prod-devtier"]
DISCOVERY_BIZ_TYPES = [
    "tidbcloud",
    "staging-tidbcloud",
    "dev-tidbcloud",
    "prod-tidbcloud-nextgen",
    "staging-tidbcloud-nextgen",
    "dev-tidbcloud-nextgen",
    "tidbcloud-nextgen",
    "prod-devtier",
    "staging-devtier",
    "dev-devtier",
    "devtier",
]


def infer_deploy_type(biz_type):
    if not biz_type:
        return ""
    if "nextgen" in biz_type:
        return "premium"
    if "devtier" in biz_type:
        return "essential"
    if biz_type in ("tidbcloud", "staging-tidbcloud", "dev-tidbcloud"):
        return "dedicated"
    return ""


def resolve_biz_type(deploy_type, env="prod"):
    base_map = {
        "essential": "devtier",
        "starter": "devtier",
        "dedicated": "tidbcloud",
        "premium": "tidbcloud-nextgen",
        "resource_pool": "tidbcloud-nextgen",
        "byoc": "tidbcloud",
    }
    base = base_map.get((deploy_type or "").lower(), "")
    if not base:
        return ""
    if not env:
        return base
    if env == "prod" and base == "tidbcloud":
        return "tidbcloud"
    return f"{env}-{base}"


def parse_time_arg(value, default=None):
    if value in (None, ""):
        return default if default is not None else int(time.time())
    if isinstance(value, (int, float)):
        return int(value)
    raw = str(value).strip()
    if raw.isdigit():
        return int(raw)
    try:
        dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ValueError(f"invalid timestamp: {value}") from exc
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.timestamp())


# ---------------------------------------------------------------------------
# OSS / S3 storage helpers
# ---------------------------------------------------------------------------

_OSS_ENDPOINT_FMT = "https://oss-{region}.aliyuncs.com"


def create_storage_client(vendor, region):
    """Create a boto3 S3 client appropriate for the vendor.

    - aws: standard boto3 S3 client with environment credentials
    - alicloud: S3-compatible client pointing at the OSS endpoint
    Returns (client, bucket_name).
    """
    import boto3
    from botocore.config import Config

    bucket = f"o11y-prod-shared-{region}"

    if vendor == "alicloud":
        endpoint = _OSS_ENDPOINT_FMT.format(region=region)
        client = boto3.client(
            "s3",
            aws_access_key_id=os.environ.get("O11Y_ALI_ACCESS_KEY_ID", ""),
            aws_secret_access_key=os.environ.get("O11Y_ALI_SECRET_ACCESS_KEY", ""),
            aws_session_token=os.environ.get("O11Y_ALI_SECURITY_TOKEN", ""),
            endpoint_url=endpoint,
            region_name=region,
            config=Config(s3={"addressing_style": "path"}),
        )
        return client, bucket

    # aws (default)
    client = boto3.client("s3", region_name=region)
    return client, bucket


def read_deltalake_parquet(s3_client, bucket, prefix, date_key, start_ts, end_ts,
                           query_sql, max_files=20):
    """List parquet files under *prefix/date_key=…/*, download to /tmp,
    and run *query_sql* against them with DuckDB.

    *query_sql* must use the placeholder table name ``__tbl__``.
    Returns the DuckDB connection and list of downloaded tmp paths.
    """
    import duckdb

    start_dt = datetime.fromtimestamp(start_ts, tz=timezone.utc)
    end_dt = datetime.fromtimestamp(end_ts, tz=timezone.utc)
    dates_to_check = set()
    d = start_dt.date()
    while d <= end_dt.date():
        dates_to_check.add(d)
        d += timedelta(days=1)

    all_keys = []
    for dt in sorted(dates_to_check):
        date_val = dt.strftime(date_key)
        full_prefix = f"{prefix}date={date_val}/"
        try:
            resp = s3_client.list_objects_v2(Bucket=bucket, Prefix=full_prefix, MaxKeys=max_files)
            for obj in resp.get("Contents", []):
                if obj["Key"].endswith(".parquet"):
                    all_keys.append(obj["Key"])
        except Exception:
            continue

    if not all_keys:
        return None, []

    tmp_dir = tempfile.mkdtemp(prefix="o11y_dl_")
    local_files = []
    for key in all_keys[:max_files]:
        fname = os.path.basename(key)
        local_path = os.path.join(tmp_dir, fname)
        try:
            s3_client.download_file(bucket, key, local_path)
            local_files.append(local_path)
        except Exception:
            continue

    if not local_files:
        return None, []

    conn = duckdb.connect(database=":memory:", read_only=False)
    glob_pattern = os.path.join(tmp_dir, "*.parquet")
    conn.execute(f"CREATE VIEW __tbl__ AS SELECT * FROM read_parquet('{glob_pattern}')")
    return conn, local_files


def _cleanup_tmp_files(paths):
    import shutil
    if not paths:
        return
    parent = os.path.dirname(paths[0])
    try:
        shutil.rmtree(parent, ignore_errors=True)
    except Exception:
        for p in paths:
            try:
                os.unlink(p)
            except OSError:
                pass


def metrics_api_base(region, vendor):
    return f"https://www.ds.{region}.{vendor}.observability.tidbcloud.com/internal/metrics"


def metrics_url_from_internal_uri(internal_read_uri):
    """Extract the query base URL from internalReadUri, preserving /v2/ if present."""
    if not internal_read_uri:
        return None
    # internalReadUri looks like: https://.../.../metrics/{uuid}/api/v1/
    # Strip trailing slash and return up to /api/v1
    base = internal_read_uri.rstrip("/")
    if base.endswith("/api/v1"):
        return base
    return None


def regex_union(values):
    values = sorted({value for value in values if value})
    if not values:
        return ""
    return "|".join(re.sub(r'([\\.^$*+?()[\]{}])', r'\\\1', value) for value in values)


def load_env():
    env_file = os.path.join(REPO_ROOT, ".env")
    if os.path.exists(env_file):
        with open(env_file) as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    k, v = line.split("=", 1)
                    os.environ.setdefault(k, v)


SUPPORTED_PROVIDERS = ("aws", "gcp", "alicloud")


def authenticate(provider):
    if provider == "azure":
        print("ERROR: Azure provider is not yet supported for credential retrieval.", file=sys.stderr)
        sys.exit(1)
    if provider not in SUPPORTED_PROVIDERS:
        print(f"ERROR: Unknown provider '{provider}'. Must be one of: {', '.join(SUPPORTED_PROVIDERS)}", file=sys.stderr)
        sys.exit(1)

    load_env()
    pat = os.environ.get("GITHUB_PAT", "")
    if not pat:
        github_file = os.path.expanduser("~/.observability/.github")
        if os.path.isfile(github_file):
            with open(github_file) as f:
                pat = f.read().strip()
    if not pat:
        print("ERROR: GITHUB_PAT not set. Provide it in .env, ~/.observability/.github, or environment.", file=sys.stderr)
        sys.exit(1)

    data = None
    last_error = None
    delay = 1
    for attempt in range(1, 4):
        try:
            resp = requests.post(
                DS_AUTH_URL,
                json={"github_token": pat, "provider": provider},
                timeout=30,
            )
            if resp.status_code == 200:
                data = resp.json()
                break
            if resp.status_code in {403, 429} or resp.status_code >= 500:
                last_error = f"HTTP {resp.status_code}: {resp.text[:200]}"
                if attempt < 3:
                    print(
                        f"WARN: /client/auth attempt {attempt} failed with HTTP {resp.status_code}; retrying...",
                        file=sys.stderr,
                    )
                    time.sleep(delay)
                    delay *= 2
                    continue
            resp.raise_for_status()
        except requests.RequestException as exc:
            last_error = str(exc)
            if attempt < 3:
                print(
                    f"WARN: /client/auth attempt {attempt} failed: {exc}; retrying...",
                    file=sys.stderr,
                )
                time.sleep(delay)
                delay *= 2
                continue
            break

    if data is None:
        print(
            f"WARN: requests authentication failed ({last_error or 'unknown error'}); trying curl fallback...",
            file=sys.stderr,
        )
        return authenticate_with_curl(pat, provider)

    auth0 = data.get("auth0_token")
    if not auth0:
        print(f"ERROR: Authentication failed: {data}", file=sys.stderr)
        sys.exit(1)

    if provider == "aws":
        os.environ["AWS_ACCESS_KEY_ID"] = data.get("access_key_id", "")
        os.environ["AWS_SECRET_ACCESS_KEY"] = data.get("secret_access_key", "")
        os.environ["AWS_SESSION_TOKEN"] = data.get("session_token", "")
    elif provider == "gcp":
        os.environ["O11Y_GCP_ACCESS_TOKEN"] = data.get("access_token", "")
    elif provider == "alicloud":
        os.environ["O11Y_ALI_ACCESS_KEY_ID"] = data.get("access_key_id", "")
        os.environ["O11Y_ALI_SECRET_ACCESS_KEY"] = data.get("secret_access_key", "")
        os.environ["O11Y_ALI_SECURITY_TOKEN"] = data.get("security_token", "")

    return auth0


def authenticate_with_curl(pat, provider):
    """Fallback auth path for environments where requests is rejected but curl succeeds."""
    try:
        proc = subprocess.run(
            [
                "curl",
                "-s",
                "-X",
                "POST",
                DS_AUTH_URL,
                "-H",
                "Content-Type: application/json",
                "-d",
                json.dumps({"github_token": pat, "provider": provider}),
            ],
            capture_output=True,
            text=True,
            check=True,
        )
        data = json.loads(proc.stdout)
    except Exception as exc:
        print(f"ERROR: curl authentication fallback failed: {exc}", file=sys.stderr)
        sys.exit(1)

    auth0 = data.get("auth0_token")
    if not auth0:
        print(f"ERROR: curl authentication fallback returned invalid payload: {data}", file=sys.stderr)
        sys.exit(1)

    if provider == "aws":
        os.environ["AWS_ACCESS_KEY_ID"] = data.get("access_key_id", "")
        os.environ["AWS_SECRET_ACCESS_KEY"] = data.get("secret_access_key", "")
        os.environ["AWS_SESSION_TOKEN"] = data.get("session_token", "")
    elif provider == "gcp":
        os.environ["O11Y_GCP_ACCESS_TOKEN"] = data.get("access_token", "")
    elif provider == "alicloud":
        os.environ["O11Y_ALI_ACCESS_KEY_ID"] = data.get("access_key_id", "")
        os.environ["O11Y_ALI_SECRET_ACCESS_KEY"] = data.get("secret_access_key", "")
        os.environ["O11Y_ALI_SECURITY_TOKEN"] = data.get("security_token", "")

    return auth0


def fetch_json_with_curl(url, headers=None, params=None):
    """HTTP JSON fallback using curl for endpoints that reject requests."""
    cmd = ["curl", "-s"]
    if headers:
        for key, value in headers.items():
            cmd.extend(["-H", f"{key}: {value}"])
    if params:
        cmd.append("-G")
        for key, value in params.items():
            cmd.extend(["--data-urlencode", f"{key}={value}"])
    cmd.append(url)
    proc = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return json.loads(proc.stdout)


def probe_metrics_target(auth0_token, o11y_id, region, vendor="aws", metrics_base=None):
    """Check whether a metrics UUID is queryable."""
    if not o11y_id:
        return False

    if metrics_base:
        url = f"{metrics_base}/query"
    else:
        url = f"{metrics_api_base(region, vendor)}/{o11y_id}/api/v1/query"

    try:
        r = requests.get(
            url,
            headers={"Authorization": f"Bearer {auth0_token}"},
            params={"query": "vector(1)", "time": int(time.time())},
            timeout=15,
        )
        return r.status_code == 200
    except requests.RequestException:
        return False


def resolve_metrics_o11y_id(auth0_token, cluster_info, region, vendor="aws"):
    """Resolve metrics UUID and base URL from internalReadUri. Returns (o11y_id, metrics_base_url)."""
    internal_uri = cluster_info.get("internalReadUri", "")
    uri_base = metrics_url_from_internal_uri(internal_uri)
    match = re.search(r"/metrics/([^/]+)/api/v1", internal_uri)
    uri_id = match.group(1) if match else ""

    raw_id = cluster_info.get("id", "")

    if uri_id and uri_base:
        if probe_metrics_target(auth0_token, uri_id, region, vendor, metrics_base=uri_base):
            return uri_id, uri_base

    for candidate in [uri_id, raw_id]:
        if not candidate:
            continue
        fallback_base = f"{metrics_api_base(region, vendor)}/{candidate}/api/v1"
        if probe_metrics_target(auth0_token, candidate, region, vendor, metrics_base=fallback_base):
            return candidate, fallback_base

    best_id = uri_id or raw_id
    best_base = uri_base or (f"{metrics_api_base(region, vendor)}/{best_id}/api/v1" if best_id else "")
    return best_id, best_base


def instant_query_metrics(auth0_token, o11y_id, region, promql, ts=None, vendor="aws", metrics_base=None):
    """Execute an instant Prometheus query."""
    ts = int(time.time()) if ts is None else ts
    if metrics_base:
        url = f"{metrics_base}/query"
    else:
        url = f"{metrics_api_base(region, vendor)}/{o11y_id}/api/v1/query"
    headers = {"Authorization": f"Bearer {auth0_token}"}
    params = {"query": promql, "time": ts}
    try:
        r = requests.get(url, headers=headers, params=params, timeout=30)
        r.raise_for_status()
        return r.json()
    except Exception as e:
        try:
            return fetch_json_with_curl(url, headers=headers, params=params)
        except Exception:
            return {"error": str(e)}


def instant_query_result(auth0_token, o11y_id, region, promql, ts=None, vendor="aws", metrics_base=None):
    resp = instant_query_metrics(
        auth0_token, o11y_id, region, promql, ts, vendor=vendor, metrics_base=metrics_base
    )
    return resp.get("data", {}).get("result", [])


def instant_query_scalar(auth0_token, o11y_id, region, promql, ts=None, vendor="aws", metrics_base=None):
    for series in instant_query_result(
        auth0_token, o11y_id, region, promql, ts, vendor=vendor, metrics_base=metrics_base
    ):
        try:
            return float(series.get("value", [0, 0])[1])
        except (TypeError, ValueError):
            continue
    return None


def target_has_serverless_pods(auth0_token, o11y_id, region, pod_names, ts=None, vendor="aws", metrics_base=None):
    pod_regex = regex_union(pod_names)
    if not pod_regex:
        return False
    value = instant_query_scalar(
        auth0_token,
        o11y_id,
        region,
        f'count(kube_pod_info{{namespace="tidb-serverless", pod=~"{pod_regex}"}})',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    )
    return bool(value and value > 0)


def resolve_serverless_keyspace(auth0_token, o11y_id, region, cluster_id, vendor="aws", metrics_base=None):
    """Map a serverless cluster_id to its stable keyspace_id."""
    queries = [
        f'sum by (keyspace_id) (0 * gateway_connection_flow_bytes_total{{cluster_id="{cluster_id}"}})',
        f'sum by (keyspace_id) (0 * tidb_server_connections{{cluster_id="{cluster_id}"}})',
    ]

    for promql in queries:
        resp = instant_query_metrics(
            auth0_token, o11y_id, region, promql, vendor=vendor, metrics_base=metrics_base
        )
        for series in resp.get("data", {}).get("result", []):
            keyspace_id = series.get("metric", {}).get("keyspace_id", "")
            if keyspace_id:
                return keyspace_id

    return ""


def resolve_serverless_tikv_instances(auth0_token, o11y_id, region, keyspace_id, ts=None, vendor="aws",
                                      metrics_base=None):
    """List the TiKV pods currently hosting data for a serverless keyspace."""
    ts = int(time.time()) if ts is None else ts
    resp = instant_query_metrics(
        auth0_token,
        o11y_id,
        region,
        f'sum(tikv_store_size_bytes{{keyspace_id="{keyspace_id}", type="used"}}) by (instance)',
        ts,
        vendor,
        metrics_base=metrics_base,
    )

    tikv_instances = []
    for series in resp.get("data", {}).get("result", []):
        instance = series.get("metric", {}).get("instance", "")
        try:
            value = float(series.get("value", [0, 0])[1])
        except (TypeError, ValueError):
            value = 0
        if instance and value > 0:
            tikv_instances.append(instance)

    return sorted(set(tikv_instances))


def resolve_serverless_tikv_nodes(auth0_token, o11y_id, region, keyspace_id, ts=None, vendor="aws",
                                  metrics_base=None, kube_o11y_id=None, kube_metrics_base=None,
                                  tikv_instances=None):
    """Map a serverless keyspace to the TiKV pods and nodes currently hosting it."""
    ts = int(time.time()) if ts is None else ts
    if tikv_instances is None:
        tikv_instances = resolve_serverless_tikv_instances(
            auth0_token, o11y_id, region, keyspace_id, ts, vendor, metrics_base=metrics_base
        )
    if not tikv_instances:
        return [], {}

    pod_regex = regex_union(tikv_instances)
    kube_target_id = kube_o11y_id or o11y_id
    kube_target_base = kube_metrics_base or metrics_base

    tikv_nodes = {}
    for series in instant_query_result(
        auth0_token,
        kube_target_id,
        region,
        f'kube_pod_info{{namespace="tidb-serverless", pod=~"{pod_regex}"}}',
        ts,
        vendor,
        metrics_base=kube_target_base,
    ):
        metric = series.get("metric", {})
        pod = metric.get("pod", "")
        node = metric.get("node", "")
        if pod and node:
            tikv_nodes[pod] = node

    return tikv_instances, tikv_nodes


def clinic_fetch_json(url, headers, params=None, timeout=5):
    try:
        r = requests.get(url, headers=headers, params=params, timeout=timeout)
        if r.status_code == 200 and r.headers.get("content-type", "").startswith("application/json"):
            return r.json()
    except Exception:
        pass
    try:
        return fetch_json_with_curl(url, headers=headers, params=params)
    except Exception:
        return None


def fetch_clinic_cluster_snapshot(cluster_id):
    """Fetch a merged Clinic cluster snapshot from dashboard + detail APIs."""
    load_env()
    clinic_key = os.environ.get("CLINIC_API_KEY", "")
    if not clinic_key:
        return None
    clinic_headers = {"Authorization": f"Bearer {clinic_key}"}
    payload = clinic_fetch_json(
        "https://clinic.pingcap.com/clinic/api/v1/dashboard/clusters",
        headers=clinic_headers,
        params={
            "cluster_id": cluster_id,
            "show_deleted": "true",
            "limit": 10,
            "page": 1,
        },
        timeout=5,
    )
    item = next(
        (entry for entry in (payload or {}).get("items", []) if entry.get("clusterID") == cluster_id),
        None,
    )
    if item:
        snapshot = {
            "id": cluster_id,
            "clusterID": cluster_id,
            "name": item.get("name", ""),
            "region": item.get("clusterRegionName", ""),
            "vendor": item.get("clusterProviderName", ""),
            "provider": item.get("clusterProviderName", ""),
            "deployType": (item.get("clusterDeployType", "") or "").lower(),
            "parentID": item.get("parentID", ""),
            "dataPlaneName": item.get("clusterDataPlaneName", ""),
            "orgID": item.get("orgID", ""),
            "tenantID": item.get("tenantID", ""),
            "projectID": item.get("projectID", ""),
        }
        org_id = snapshot.get("orgID", "")
        if org_id:
            detail = clinic_fetch_json(
                f"https://clinic.pingcap.com/clinic/api/v1/orgs/{org_id}/clusters/{cluster_id}",
                headers=clinic_headers,
                timeout=5,
            )
            if detail:
                snapshot["name"] = detail.get("name", snapshot["name"])
                snapshot["dataPlaneName"] = detail.get("dataPlaneName", snapshot["dataPlaneName"])
                snapshot["parentID"] = detail.get("parentID", snapshot["parentID"])
                snapshot["region"] = detail.get("region", snapshot["region"])
                snapshot["vendor"] = detail.get("provider", snapshot["vendor"])
                snapshot["provider"] = snapshot["vendor"]
                snapshot["deployType"] = (detail.get("deployType") or snapshot["deployType"] or "").lower()
        return snapshot

    return clinic_fetch_json(
        f"https://clinic.pingcap.com/api/v1/clusters/{cluster_id}",
        headers=clinic_headers,
        timeout=5,
    )


def try_clinic_api(cluster_id):
    """Tier 1: Try Clinic Metadata API for instant cluster lookup."""
    return fetch_clinic_cluster_snapshot(cluster_id)


def fetch_clinic_topology(cluster_snapshot):
    load_env()
    clinic_key = os.environ.get("CLINIC_API_KEY", "")
    if not clinic_key or not cluster_snapshot:
        return []

    cluster_id = str(cluster_snapshot.get("clusterID") or cluster_snapshot.get("id") or "")
    org_id = str(cluster_snapshot.get("orgID") or "")
    deploy_type = (cluster_snapshot.get("deployType") or cluster_snapshot.get("deploy_type") or "").lower()
    parent_id = str(cluster_snapshot.get("parentID") or "")
    headers = {"Authorization": f"Bearer {clinic_key}"}

    if not cluster_id or not org_id:
        return []

    if deploy_type == "dedicated":
        detail = clinic_fetch_json(
            f"https://clinic.pingcap.com/clinic/api/v1/orgs/{org_id}/clusters/{cluster_id}",
            headers=headers,
            timeout=5,
        )
        return (detail or {}).get("components", []) or []
    if deploy_type == "resource_pool":
        payload = clinic_fetch_json(
            f"https://clinic.pingcap.com/clinic/api/v1/orgs/{org_id}/clusters/{cluster_id}/resource-pool/components",
            headers=headers,
            timeout=5,
        )
        return payload or []
    if deploy_type == "premium" and parent_id:
        payload = clinic_fetch_json(
            f"https://clinic.pingcap.com/clinic/api/v1/orgs/{org_id}/clusters/{parent_id}/resource-pool/components",
            headers=headers,
            timeout=5,
        )
        return payload or []
    return []


def build_clinic_cluster_context(cluster_id):
    snapshot = fetch_clinic_cluster_snapshot(cluster_id)
    if not snapshot:
        return None
    deploy_type = (snapshot.get("deployType") or snapshot.get("deploy_type") or "").lower()
    return {
        "vendor": snapshot.get("vendor", "") or snapshot.get("provider", ""),
        "region": snapshot.get("region", ""),
        "deploy_type": deploy_type,
        "tenant_id": snapshot.get("tenantID", ""),
        "org_id": snapshot.get("orgID", ""),
        "parent_id": snapshot.get("parentID", ""),
        "data_plane_name": snapshot.get("dataPlaneName", ""),
        "gs_biz_type": resolve_biz_type(deploy_type),
    }


def build_shell_metrics_targets(
    cluster_id,
    analysis_ts,
    provider="aws",
    hint_region=None,
    hint_vendor=None,
):
    clinic_snapshot = fetch_clinic_cluster_snapshot(cluster_id)
    actual_vendor = hint_vendor or provider
    actual_region = hint_region
    if clinic_snapshot:
        actual_vendor = clinic_snapshot.get("vendor") or clinic_snapshot.get("provider") or actual_vendor
        actual_region = actual_region or clinic_snapshot.get("region")

    auth0_token = authenticate(actual_vendor or provider)
    cluster = find_cluster(auth0_token, cluster_id, actual_region, actual_vendor, analysis_ts)
    return {
        "cluster_id": cluster_id,
        "namespace": cluster.get("namespace", ""),
        "vendor": cluster.get("vendor", ""),
        "region": cluster.get("region", ""),
        "biz_type": cluster.get("biz_type", ""),
        "deploy_type": cluster.get("deploy_type", ""),
        "tenant_id": cluster.get("tenant_id", ""),
        "project_id": cluster.get("project_id", ""),
        "app_uuid": cluster.get("o11y_id", ""),
        "app_application_id": cluster.get("application_id", ""),
        "app_metrics_base": cluster.get("metrics_base", ""),
        "infra_uuid": cluster.get("infra_o11y_id", ""),
        "infra_application_id": cluster.get("data_plane_name", ""),
        "infra_metrics_base": cluster.get("infra_metrics_base", ""),
        "keyspace_id": cluster.get("keyspace_id", ""),
        "tikv_instances": cluster.get("tikv_instances", []),
        "tikv_nodes": cluster.get("tikv_nodes", {}),
    }


def _gs_search_one(
    auth0_token,
    cluster_id,
    vendor,
    region,
    biz_type,
    org_id_header=None,
    biz_type_header=None,
    region_id_header=None,
):
    """Single GS API lookup. Returns cluster dict or None."""
    clusters = gs_query_clusters(
        auth0_token,
        vendor,
        region,
        org_id_header=org_id_header,
        biz_type_header=biz_type_header or biz_type,
        region_id_header=region_id_header,
        bizType=biz_type,
        applicationID=cluster_id,
    )
    exclude_ids = {"k8s-infra", "base", "seed"}
    clusters = [
        c for c in clusters
        if c.get("applicationID") not in exclude_ids
        and c.get("projectID") not in {"seed", "base"}
    ]
    return clusters[0] if clusters else None


def gs_query_clusters(
    auth0_token,
    vendor,
    region,
    org_id_header=None,
    biz_type_header=None,
    region_id_header=None,
    **params,
):
    request_params = {"vendor": vendor, "region": region}
    for key, value in params.items():
        if value not in (None, ""):
            request_params[key] = value

    headers = {"Authorization": f"Bearer {auth0_token}"}
    header_biz_type = biz_type_header or request_params.get("bizType")
    header_region_id = region_id_header or f"{vendor}-{region}"
    if org_id_header:
        headers["x-org-id"] = org_id_header
    if header_biz_type:
        headers["x-biz-type"] = header_biz_type
    if header_region_id:
        headers["x-region-id"] = header_region_id
    try:
        r = requests.get(GS_API_BASE, headers=headers, params=request_params, timeout=10)
        r.raise_for_status()
        return r.json().get("data", {}).get("clusters", [])
    except Exception:
        try:
            data = fetch_json_with_curl(GS_API_BASE, headers=headers, params=request_params)
        except Exception:
            return []
        return data.get("data", {}).get("clusters", [])


def target_has_namespace(auth0_token, o11y_id, region, namespace, ts=None, vendor="aws", metrics_base=None):
    if not namespace:
        return False
    value = instant_query_scalar(
        auth0_token,
        o11y_id,
        region,
        f'count(kube_pod_info{{namespace="{namespace}"}})',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    )
    return bool(value and value > 0)


def resolve_non_serverless_infra_target(auth0_token, cluster_info, region, vendor, namespace):
    biz_type = cluster_info.get("bizType", "")
    project_id = cluster_info.get("projectID", "")
    tenant_id = cluster_info.get("tenantID", "")
    org_id = cluster_info.get("orgID", "")
    app_id = cluster_info.get("applicationID", "")
    data_plane_name = cluster_info.get("dataPlaneName", "")
    deploy_type = (cluster_info.get("deployType", "") or cluster_info.get("deploy_type", "")).lower()
    is_nextgen = "nextgen" in biz_type or deploy_type in {"premium", "resource_pool"}
    org_id_header = (tenant_id or org_id) if is_nextgen else None
    region_id_header = f"{vendor}-{region}"

    if biz_type == "tidbcloud" and project_id:
        for candidate in gs_query_clusters(
            auth0_token,
            vendor,
            region,
            org_id_header=org_id_header,
            biz_type_header=biz_type,
            region_id_header=region_id_header,
            bizType=biz_type,
            applicationID="k8s-infra",
            projectID=project_id,
        ):
            if candidate.get("applicationID") != "k8s-infra":
                continue
            if candidate.get("projectID") != project_id:
                continue
            return resolve_metrics_o11y_id(auth0_token, candidate, region, vendor)

    if data_plane_name:
        for discovery_biz_type in DISCOVERY_BIZ_TYPES:
            for candidate in gs_query_clusters(
                auth0_token,
                vendor,
                region,
                org_id_header=org_id_header,
                biz_type_header=discovery_biz_type,
                region_id_header=region_id_header,
                bizType=discovery_biz_type,
                applicationID=data_plane_name,
            ):
                if candidate.get("applicationID") != data_plane_name:
                    continue
                if candidate.get("projectID") in {"seed", "base"}:
                    continue
                return resolve_metrics_o11y_id(auth0_token, candidate, region, vendor)

    if not tenant_id:
        return "", ""

    candidates = [
        candidate
        for candidate in gs_query_clusters(
            auth0_token,
            vendor,
            region,
            org_id_header=org_id_header,
            biz_type_header=biz_type or None,
            region_id_header=region_id_header,
            tenantID=tenant_id,
            bizType=biz_type or None,
        )
        if (candidate.get("applicationID", "")).startswith("dp-")
        and candidate.get("applicationID") != app_id
    ]
    if not candidates:
        return "", ""

    if len(candidates) == 1:
        return resolve_metrics_o11y_id(auth0_token, candidates[0], region, vendor)

    for candidate in sorted(candidates, key=lambda item: item.get("applicationID", "")):
        infra_o11y_id, infra_metrics_base = resolve_metrics_o11y_id(
            auth0_token, candidate, region, vendor
        )
        if not infra_o11y_id:
            continue
        if target_has_namespace(
            auth0_token,
            infra_o11y_id,
            region,
            namespace,
            vendor=vendor,
            metrics_base=infra_metrics_base,
        ):
            return infra_o11y_id, infra_metrics_base

    return "", ""


def resolve_serverless_infra_target(auth0_token, cluster_info, region, vendor, pod_names):
    """Resolve the infra metrics target for a serverless cluster by matching TiKV pods."""
    tenant_id = cluster_info.get("tenantID", "")
    biz_type = cluster_info.get("bizType", "")
    if not tenant_id or not pod_names:
        return "", ""

    current_o11y_id, current_metrics_base = resolve_metrics_o11y_id(
        auth0_token, cluster_info, region, vendor
    )
    if target_has_serverless_pods(
        auth0_token, current_o11y_id, region, pod_names, vendor=vendor, metrics_base=current_metrics_base
    ):
        return current_o11y_id, current_metrics_base

    params = {"vendor": vendor, "region": region, "tenantID": tenant_id}
    if biz_type:
        params["bizType"] = biz_type

    try:
        r = requests.get(
            GS_API_BASE,
            headers={"Authorization": f"Bearer {auth0_token}"},
            params=params,
            timeout=5,
        )
        if r.status_code != 200:
            return "", ""
        clusters = r.json().get("data", {}).get("clusters", [])
    except Exception:
        return "", ""

    candidates = [
        c for c in clusters
        if c.get("applicationID") != cluster_info.get("applicationID")
    ]
    if not candidates:
        return "", ""

    for candidate in sorted(
        candidates,
        key=lambda c: (not (c.get("applicationID", "")).startswith("dp-"), c.get("applicationID", "")),
    ):
        infra_o11y_id, infra_metrics_base = resolve_metrics_o11y_id(
            auth0_token, candidate, region, vendor
        )
        if not infra_o11y_id:
            continue
        if target_has_serverless_pods(
            auth0_token,
            infra_o11y_id,
            region,
            pod_names,
            vendor=vendor,
            metrics_base=infra_metrics_base,
        ):
            return infra_o11y_id, infra_metrics_base

    return "", ""


def find_cluster(auth0_token, cluster_id, hint_region=None, hint_vendor=None, analysis_ts=None):
    """Three-tier cluster lookup: Clinic → quick GS scan → error with guidance."""
    clinic_deploy_type = ""
    clinic_parent_id = ""
    clinic_data_plane_name = ""
    clinic_tenant_id = ""
    clinic_org_id = ""

    # Tier 1: Clinic API
    print("  Tier 1: Trying Clinic API...", file=sys.stderr)
    clinic_data = try_clinic_api(cluster_id)
    if clinic_data and clinic_data.get("region"):
        print(f"  Clinic API found: {clinic_data.get('vendor')}/{clinic_data.get('region')}", file=sys.stderr)
        hint_vendor = clinic_data.get("vendor", hint_vendor)
        hint_region = clinic_data.get("region", hint_region)
        clinic_deploy_type = (clinic_data.get("deployType") or clinic_data.get("deploy_type") or "").lower()
        clinic_parent_id = str(clinic_data.get("parentID") or "")
        clinic_data_plane_name = clinic_data.get("dataPlaneName", "")
        clinic_tenant_id = str(clinic_data.get("tenantID") or "")
        clinic_org_id = str(clinic_data.get("orgID") or "")

    # Tier 2: Quick GS API scan across all vendors
    print("  Tier 2: Scanning GS API across vendors/regions...", file=sys.stderr)

    if hint_vendor and hint_region:
        vendors_to_scan = [(hint_vendor, [hint_region])]
        for v, regions in VENDOR_REGIONS.items():
            if v != hint_vendor:
                vendors_to_scan.append((v, regions))
            elif hint_region not in regions:
                vendors_to_scan.append((v, [r for r in regions if r != hint_region]))
    else:
        vendors_to_scan = [(v, regions) for v, regions in VENDOR_REGIONS.items()]

    for vendor, regions in vendors_to_scan:
        for region in regions:
            for biz_type in SCAN_BIZ_TYPES:
                nextgen_header_org_id = None
                nextgen_header_region_id = None
                if clinic_deploy_type in ("premium", "resource_pool"):
                    nextgen_header_org_id = clinic_tenant_id or clinic_org_id or None
                    nextgen_header_region_id = f"{vendor}-{region}"
                info = _gs_search_one(
                    auth0_token,
                    cluster_id,
                    vendor,
                    region,
                    biz_type,
                    org_id_header=nextgen_header_org_id,
                    biz_type_header=biz_type,
                    region_id_header=nextgen_header_region_id,
                )
                if info:
                    resolved_region = info["region"]
                    resolved_vendor = info.get("vendor", vendor)
                    biz_type_value = info.get("bizType", biz_type)
                    deploy_type = (info.get("deployType", "") or clinic_deploy_type or infer_deploy_type(biz_type_value)).lower()
                    shared_parent_id = str(info.get("parentID") or clinic_parent_id or "")
                    if "devtier" in biz_type_value or deploy_type in ("starter", "essential"):
                        namespace = "tidb-serverless"
                    elif deploy_type in ("premium", "resource_pool") and shared_parent_id:
                        namespace = f"tidb{shared_parent_id}"
                    else:
                        namespace = f"tidb{cluster_id}"

                    metrics_o11y_id, metrics_base = resolve_metrics_o11y_id(
                        auth0_token, info, resolved_region, resolved_vendor
                    )
                    cluster = {
                        "cluster_id": cluster_id,
                        "o11y_id": metrics_o11y_id,
                        "metrics_base": metrics_base,
                        "infra_o11y_id": "",
                        "infra_metrics_base": "",
                        "gs_id": info["id"],
                        "tenant_id": info["tenantID"],
                        "org_id": info.get("orgID", "") or clinic_org_id,
                        "project_id": info.get("projectID", ""),
                        "application_id": info.get("applicationID", ""),
                        "region": resolved_region,
                        "vendor": resolved_vendor,
                        "biz_type": biz_type_value,
                        "version": info.get("version", ""),
                        "deploy_type": deploy_type,
                        "namespace": namespace,
                        "data_plane_name": info.get("dataPlaneName", "") or clinic_data_plane_name,
                        "parent_id": shared_parent_id,
                    }
                    if "devtier" in biz_type_value or deploy_type in ("starter", "essential"):
                        keyspace_id = resolve_serverless_keyspace(
                            auth0_token,
                            metrics_o11y_id,
                            resolved_region,
                            cluster_id,
                            resolved_vendor,
                            metrics_base=metrics_base,
                        )
                        tikv_instances, tikv_nodes = ([], {})
                        if keyspace_id:
                            infra_o11y_id = ""
                            infra_metrics_base = ""
                            tikv_instances = resolve_serverless_tikv_instances(
                                auth0_token,
                                metrics_o11y_id,
                                resolved_region,
                                keyspace_id,
                                analysis_ts,
                                resolved_vendor,
                                metrics_base=metrics_base,
                            )
                            if tikv_instances:
                                infra_o11y_id, infra_metrics_base = resolve_serverless_infra_target(
                                    auth0_token, info, resolved_region, resolved_vendor, tikv_instances
                                )
                            cluster["infra_o11y_id"] = infra_o11y_id
                            cluster["infra_metrics_base"] = infra_metrics_base
                            tikv_instances, tikv_nodes = resolve_serverless_tikv_nodes(
                                auth0_token,
                                metrics_o11y_id,
                                resolved_region,
                                keyspace_id,
                                analysis_ts,
                                resolved_vendor,
                                metrics_base=metrics_base,
                                kube_o11y_id=infra_o11y_id or None,
                                kube_metrics_base=infra_metrics_base or None,
                                tikv_instances=tikv_instances,
                            )
                        cluster["keyspace_id"] = keyspace_id
                        cluster["tikv_instances"] = tikv_instances
                        cluster["tikv_nodes"] = tikv_nodes
                    else:
                        infra_o11y_id, infra_metrics_base = resolve_non_serverless_infra_target(
                            auth0_token,
                            info,
                            resolved_region,
                            resolved_vendor,
                            namespace,
                        )
                        cluster["infra_o11y_id"] = infra_o11y_id
                        cluster["infra_metrics_base"] = infra_metrics_base
                    print(f"  Found: {resolved_vendor}/{resolved_region} bizType={biz_type_value}", file=sys.stderr)
                    return cluster

    # Tier 3: Not found — give user clear guidance
    print(
        f"\nERROR: Cluster {cluster_id} not found in common regions.\n"
        f"Please provide --vendor and --region explicitly.\n"
        f"  Example: python o11y-inspect.py {cluster_id} --provider aws --vendor aws --region us-east-1\n"
        f"  Supported vendors: aws, gcp, alicloud\n",
        file=sys.stderr,
    )
    sys.exit(1)


def query_metrics(auth0_token, o11y_id, region, promql, start_ts, end_ts, step=60, vendor="aws", metrics_base=None):
    """Execute a Prometheus query_range and return parsed result."""
    if metrics_base:
        base_url = f"{metrics_base}/query_range"
    else:
        base_url = f"{metrics_api_base(region, vendor)}/{o11y_id}/api/v1/query_range"
    headers = {"Authorization": f"Bearer {auth0_token}"}
    params = {"query": promql, "start": start_ts, "end": end_ts, "step": step}
    try:
        r = requests.get(base_url, headers=headers, params=params, timeout=30)
        r.raise_for_status()
        return r.json()
    except Exception as e:
        try:
            return fetch_json_with_curl(base_url, headers=headers, params=params)
        except Exception:
            return {"error": str(e)}


def latest_value(metrics_resp):
    """Extract the latest numeric value from a query_range response."""
    try:
        results = metrics_resp.get("data", {}).get("result", [])
        if not results:
            return None
        values = results[0].get("values", [])
        if not values:
            return None
        return float(values[-1][1])
    except (IndexError, ValueError, TypeError):
        return None


def sum_latest_values(metrics_resp):
    """Sum latest values across all series."""
    try:
        results = metrics_resp.get("data", {}).get("result", [])
        total = 0
        for series in results:
            vals = series.get("values", [])
            if vals:
                total += float(vals[-1][1])
        return total
    except (ValueError, TypeError):
        return None


def collect_metrics(auth0_token, o11y_id, region, start_ts, end_ts, vendor="aws", metrics_base=None):
    """Collect standard set of cluster metrics."""
    metrics = {}

    qps_resp = query_metrics(auth0_token, o11y_id, region,
                             "sum(rate(tidb_server_query_total[5m])) by (type)",
                             start_ts, end_ts, step=300, vendor=vendor, metrics_base=metrics_base)
    qps_by_type = {}
    qps_total = 0
    for series in qps_resp.get("data", {}).get("result", []):
        t = series.get("metric", {}).get("type", "unknown")
        vals = series.get("values", [])
        v = float(vals[-1][1]) if vals else 0
        qps_by_type[t] = round(v, 2)
        qps_total += v
    metrics["qps"] = {"total": round(qps_total, 2), "by_type": qps_by_type}

    p99_resp = query_metrics(auth0_token, o11y_id, region,
                             "histogram_quantile(0.99, sum(rate(tidb_server_handle_query_duration_seconds_bucket[5m])) by (le))",
                             start_ts, end_ts, step=300, vendor=vendor, metrics_base=metrics_base)
    metrics["latency_p99_seconds"] = round(latest_value(p99_resp) or 0, 4)

    conn_resp = query_metrics(auth0_token, o11y_id, region,
                              "sum(tidb_server_connections)", start_ts, end_ts, step=300,
                              vendor=vendor, metrics_base=metrics_base)
    metrics["connections"] = int(latest_value(conn_resp) or 0)

    cpu_resp = query_metrics(auth0_token, o11y_id, region,
                             "sum(rate(tikv_thread_cpu_seconds_total[1m]))",
                             start_ts, end_ts, step=300, vendor=vendor, metrics_base=metrics_base)
    metrics["tikv_cpu_total"] = round(latest_value(cpu_resp) or 0, 2)

    region_resp = query_metrics(auth0_token, o11y_id, region,
                                "sum(tikv_raftstore_region_count) by (type)",
                                start_ts, end_ts, step=300, vendor=vendor, metrics_base=metrics_base)
    tikv_regions = {}
    for series in region_resp.get("data", {}).get("result", []):
        t = series.get("metric", {}).get("type", "unknown")
        vals = series.get("values", [])
        tikv_regions[t] = int(float(vals[-1][1])) if vals else 0
    metrics["tikv_regions"] = tikv_regions

    return metrics


def empty_stats():
    return {"current": None, "avg_1h": None, "peak_1h": None}


def capacity_usage(capacity, available):
    if capacity is None or available is None:
        return None, None
    used = max(capacity - available, 0)
    if capacity <= 0:
        return used, None
    return used, (used / capacity * 100)


def mapping_disk_name(mapping):
    return (
        mapping.get("disk_name", "")
        or mapping.get("pvc", "")
        or mapping.get("persistentvolume", "")
        or mapping.get("volume_id", "")
        or mapping.get("device", "")
    )


def normalize_disk_inventory(mappings):
    disks = []
    for mapping in mappings or []:
        disks.append({
            "name": mapping_disk_name(mapping),
            "pod": mapping.get("pod", ""),
            "node": mapping.get("node", ""),
            "pvc": mapping.get("pvc", ""),
            "persistentvolume": mapping.get("persistentvolume", ""),
            "storageclass": mapping.get("storageclass", ""),
            "device": mapping.get("device", ""),
            "linux_device_path": mapping.get("linux_device_path", ""),
            "mountpoint": mapping.get("mountpoint", ""),
            "capacity_bytes": mapping.get("capacity_bytes"),
            "used_bytes": mapping.get("used_bytes"),
            "usage_pct": mapping.get("usage_pct"),
            "read_iops": mapping.get("read_iops", empty_stats()),
            "write_iops": mapping.get("write_iops", empty_stats()),
            "read_latency_p95_ms": mapping.get("read_latency_ms", empty_stats()),
            "write_latency_p95_ms": mapping.get("write_latency_ms", empty_stats()),
            "read_bandwidth_mib_s": mapping.get("read_bandwidth_mib_s", empty_stats()),
            "write_bandwidth_mib_s": mapping.get("write_bandwidth_mib_s", empty_stats()),
            "util_pct": mapping.get("util_pct", empty_stats()),
            "metric_sources": mapping.get("metric_sources", {}),
            "latency_note": mapping.get("latency_note"),
        })
    return disks


def build_disk_sum_rate_expr(metric_name, disk_targets):
    parts = [
        f'rate({metric_name}{{instance="{target["instance"]}",device="{target["device"]}"}}[5m])'
        for target in disk_targets
        if target.get("instance") and target.get("device")
    ]
    if not parts:
        return "vector(0)"
    if len(parts) == 1:
        return f"sum({parts[0]})"
    return f"sum(({' ) or ( '.join(parts)}))"


def disk_target_key(instance, device):
    return (instance or "", device or "")


def build_disk_target_rate_expr(metric_name, disk_targets):
    parts = [
        f'rate({metric_name}{{instance="{target["instance"]}",device="{target["device"]}"}}[5m])'
        for target in disk_targets
        if target.get("instance") and target.get("device")
    ]
    if not parts:
        return "vector(0)"
    if len(parts) == 1:
        return parts[0]
    return f"({' ) or ( '.join(parts)})"


def build_disk_util_expr(disk_targets):
    if not disk_targets:
        return "vector(0)"
    io_expr = build_disk_sum_rate_expr("node_disk_io_time_seconds_total", disk_targets)
    if len(disk_targets) == 1:
        return io_expr
    return f"(({io_expr}) / {len(disk_targets)})"


def build_disk_target_util_expr(disk_targets):
    if not disk_targets:
        return "vector(0)"
    return build_disk_target_rate_expr("node_disk_io_time_seconds_total", disk_targets)


def collect_expr_stats(
    auth0_token,
    o11y_id,
    region,
    expr,
    ts=None,
    vendor="aws",
    metrics_base=None,
    window="1h",
    subquery_step_seconds=60,
):
    current = instant_query_scalar(
        auth0_token, o11y_id, region, expr, ts, vendor=vendor, metrics_base=metrics_base
    )
    avg_1h = instant_query_scalar(
        auth0_token,
        o11y_id,
        region,
        f"avg_over_time((({expr}))[{window}:{subquery_step_seconds}s])",
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    )
    peak_1h = instant_query_scalar(
        auth0_token,
        o11y_id,
        region,
        f"max_over_time((({expr}))[{window}:{subquery_step_seconds}s])",
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    )
    return {"current": current, "avg_1h": avg_1h, "peak_1h": peak_1h}


def collect_labeled_expr_stats(
    auth0_token,
    o11y_id,
    region,
    expr,
    labels,
    ts=None,
    vendor="aws",
    metrics_base=None,
    window="1h",
    subquery_step_seconds=60,
):
    stats_by_key = {}

    def collect(field_name, query_expr):
        for series in instant_query_result(
            auth0_token,
            o11y_id,
            region,
            query_expr,
            ts,
            vendor=vendor,
            metrics_base=metrics_base,
        ):
            metric = series.get("metric", {})
            key = tuple(metric.get(label, "") for label in labels)
            if not any(key):
                continue
            try:
                value = float(series.get("value", [0, 0])[1])
            except (TypeError, ValueError):
                continue
            stats_by_key.setdefault(key, empty_stats())[field_name] = value

    collect("current", expr)
    collect("avg_1h", f"avg_over_time((({expr}))[{window}:{subquery_step_seconds}s])")
    collect("peak_1h", f"max_over_time((({expr}))[{window}:{subquery_step_seconds}s])")
    return stats_by_key


def parse_storageclass_spec(storageclass):
    spec = {"storageclass": storageclass or ""}
    match = re.fullmatch(r"gp3-(\d+)mb-(\d+)io", storageclass or "")
    if match:
        spec["provisioned_bandwidth_mib_s"] = int(match.group(1))
        spec["provisioned_iops"] = int(match.group(2))
        return spec
    match = re.fullmatch(r"io2-(\d+)", storageclass or "")
    if match:
        spec["provisioned_iops"] = int(match.group(1))
    return spec


def build_tikv_component_spec(tikv_component):
    if not tikv_component:
        return {}
    node_count = tikv_component.get("node_count")
    storages = []
    for storage in tikv_component.get("storages", []) or []:
        item = dict(storage)
        size_gi = item.get("size_gi")
        iops = item.get("iops")
        throughput_mbs = item.get("throughput_mbs")
        item["total_size_gi"] = size_gi * node_count if size_gi is not None and node_count is not None else None
        item["total_iops"] = iops * node_count if iops is not None and node_count is not None else None
        item["total_throughput_mbs"] = (
            throughput_mbs * node_count if throughput_mbs is not None and node_count is not None else None
        )
        storages.append(item)
    return {
        "component_type": tikv_component.get("component_type", ""),
        "component_id": tikv_component.get("id", ""),
        "cpu": tikv_component.get("cpu"),
        "memory_gi": tikv_component.get("memory_gi"),
        "node_count": node_count,
        "version": tikv_component.get("version", ""),
        "storages": storages,
    }


def normalize_volume_id(value, vendor="aws"):
    if not value:
        return ""

    vendor_name = (vendor or "").lower()
    if vendor_name == "aws":
        match = re.search(r"(vol-?[0-9A-Za-z]+)", value)
        if match:
            token = match.group(1)
            if token.startswith("vol-"):
                return token
            if token.startswith("vol"):
                return f"vol-{token[3:]}"
        if value.startswith("vol-"):
            return value
        if value.startswith("vol"):
            return f"vol-{value[3:]}"
        return value

    if vendor_name == "alicloud":
        match = re.search(r"(d-[0-9A-Za-z]+)", value)
        if match:
            return match.group(1)
        return value

    return value


def normalize_aws_volume_id(value):
    return normalize_volume_id(value, vendor="aws")


def volume_identity_keys(value, vendor="aws"):
    normalized = normalize_volume_id(value, vendor=vendor)
    keys = set()
    for candidate in (value, normalized):
        if not candidate:
            continue
        keys.add(candidate)
        keys.add(candidate.replace("-", ""))

        if (vendor or "").lower() == "alicloud" and candidate.startswith("d-"):
            keys.add(candidate[2:])
            keys.add(candidate[2:].replace("-", ""))

    return sorted(key for key in keys if key)


def build_node_disk_map(auth0_token, o11y_id, region, volume_ids, ts=None, vendor="aws", metrics_base=None):
    serial_regex = regex_union(
        key
        for volume_id in volume_ids
        for key in volume_identity_keys(volume_id, vendor=vendor)
    )
    disk_map = {}
    if not serial_regex:
        return disk_map

    for series in instant_query_result(
        auth0_token,
        o11y_id,
        region,
        f'node_disk_info{{serial=~"{serial_regex}"}}',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    ):
        metric = series.get("metric", {})
        serial = metric.get("serial", "")
        if not serial:
            continue
        info = {
            "serial": serial,
            "instance": metric.get("instance", ""),
            "device": metric.get("device", ""),
            "model": metric.get("model", ""),
        }
        for key in volume_identity_keys(serial, vendor=vendor):
            disk_map[key] = info

    return disk_map


def lookup_disk_info(disk_map, volume_id, vendor="aws"):
    for key in volume_identity_keys(volume_id, vendor=vendor):
        info = disk_map.get(key)
        if info:
            return info
    return {}


def metric_node_name(metric):
    instance = metric.get("instance", "")
    if ":" in instance:
        instance = instance.split(":", 1)[0]
    return (
        metric.get("node", "")
        or metric.get("kubernetes_io_hostname", "")
        or metric.get("label_kubernetes_io_hostname", "")
        or instance
    )


def metric_instance_type(metric):
    return (
        metric.get("label_node_kubernetes_io_instance_type", "")
        or metric.get("label_beta_kubernetes_io_instance_type", "")
    )


def device_name_from_path(device_path):
    if not device_path:
        return ""
    return device_path.rsplit("/", 1)[-1]


def build_node_local_path_filesystem_map(
    auth0_token,
    o11y_id,
    region,
    nodes,
    ts=None,
    vendor="aws",
    metrics_base=None,
):
    node_set = {node for node in nodes if node}
    if not node_set:
        return {}

    local_path_expr = 'node_filesystem_size_bytes{mountpoint=~"/nvme/local-path-provisioner|/pv-disks/.+"}'
    fs_map = {}
    for series in instant_query_result(
        auth0_token,
        o11y_id,
        region,
        local_path_expr,
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    ):
        metric = series.get("metric", {})
        node = metric_node_name(metric)
        mountpoint = metric.get("mountpoint", "")
        if node not in node_set or not mountpoint:
            continue
        current = fs_map.get(node)
        prefer_canonical_mount = mountpoint == "/nvme/local-path-provisioner"
        if current is not None and not prefer_canonical_mount:
            continue
        try:
            capacity_bytes = float(series.get("value", [0, 0])[1])
        except (TypeError, ValueError):
            capacity_bytes = None
        fs_map[node] = {
            "mountpoint": mountpoint,
            "device_path": metric.get("device", ""),
            "fstype": metric.get("fstype", ""),
            "capacity_bytes": capacity_bytes,
            "available_bytes": None,
        }

    for series in instant_query_result(
        auth0_token,
        o11y_id,
        region,
        'node_filesystem_avail_bytes{mountpoint=~"/nvme/local-path-provisioner|/pv-disks/.+"}',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    ):
        metric = series.get("metric", {})
        node = metric_node_name(metric)
        mountpoint = metric.get("mountpoint", "")
        entry = fs_map.get(node)
        if not entry or mountpoint != entry.get("mountpoint"):
            continue
        try:
            entry["available_bytes"] = float(series.get("value", [0, 0])[1])
        except (TypeError, ValueError):
            continue

    return fs_map


def build_node_instance_store_map(
    auth0_token,
    o11y_id,
    region,
    nodes,
    ts=None,
    vendor="aws",
    metrics_base=None,
):
    node_set = {node for node in nodes if node}
    if not node_set:
        return {}

    disk_map = {}
    for series in instant_query_result(
        auth0_token,
        o11y_id,
        region,
        'node_disk_info{model="Amazon EC2 NVMe Instance Storage"}',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    ):
        metric = series.get("metric", {})
        node = metric_node_name(metric)
        device = metric.get("device", "")
        if node not in node_set or not device:
            continue
        disk_map.setdefault(node, {})[device] = {
            "instance": metric.get("instance", ""),
            "device": device,
            "serial": metric.get("serial", ""),
            "model": metric.get("model", ""),
        }

    return disk_map


def build_metric_fallback_expr(metric_name, builder):
    metric_names = [metric_name]
    if metric_name.startswith("aws_ebs_csi_"):
        metric_names.append(metric_name.replace("aws_ebs_csi_", "aws_ebs_"))

    expressions = []
    for candidate_name in metric_names:
        expr = builder(candidate_name)
        if expr not in expressions:
            expressions.append(expr)

    if not expressions:
        return "vector(0)"
    if len(expressions) == 1:
        return expressions[0]
    return f"({' ) or ( '.join(expressions)})"


def build_aws_ebs_rate_expr(metric_name, volume_regex):
    return build_metric_fallback_expr(
        metric_name,
        lambda candidate_name: f'sum(rate({candidate_name}{{volume_id=~"{volume_regex}"}}[5m]))',
    )


def build_aws_ebs_latency_expr(metric_name, volume_regex, quantile=0.95):
    return (
        "1000 * "
        + build_metric_fallback_expr(
            metric_name,
            lambda candidate_name: (
                f'histogram_quantile({quantile}, '
                f'sum by (le) (rate({candidate_name}{{volume_id=~"{volume_regex}"}}[5m])))'
            ),
        )
    )


def disk_report_target_kind(cluster_info):
    if cluster_info.get("infra_o11y_id"):
        if cluster_info.get("infra_o11y_id") == cluster_info.get("o11y_id"):
            return "shared"
        return "infra"
    return "app"


def collect_serverless_disk_report(
    auth0_token,
    cluster_info,
    end_ts,
    window="1h",
    subquery_step_seconds=60,
):
    tikv_instances = cluster_info.get("tikv_instances", [])
    tikv_nodes = cluster_info.get("tikv_nodes", {})
    if not tikv_instances or not tikv_nodes:
        return {"info": "No TiKV pod/node mapping available for serverless disk analysis", "disks": []}

    region = cluster_info["region"]
    vendor = cluster_info["vendor"]
    target_o11y_id = cluster_info.get("infra_o11y_id") or cluster_info["o11y_id"]
    target_metrics_base = cluster_info.get("infra_metrics_base") or cluster_info.get("metrics_base")
    pod_regex = regex_union(tikv_instances)

    claims = {}
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kube_pod_spec_volumes_persistentvolumeclaims_info{{namespace="tidb-serverless", pod=~"{pod_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        pod = metric.get("pod", "")
        if pvc and pod:
            claims[pvc] = {
                "role": metric.get("volume", ""),
                "pod": pod,
                "pvc": pvc,
                "node": tikv_nodes.get(pod, ""),
            }

    if not claims:
        return {"info": "No TiKV PVCs found for serverless disk analysis", "disks": []}

    pvc_regex = regex_union(claims.keys())
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kube_persistentvolumeclaim_info{{namespace="tidb-serverless", persistentvolumeclaim=~"{pvc_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        if pvc in claims:
            claims[pvc]["storageclass"] = metric.get("storageclass", "")
            claims[pvc]["persistentvolume"] = metric.get("volumename", "")

    pv_regex = regex_union(
        entry.get("persistentvolume", "") for entry in claims.values() if entry.get("persistentvolume")
    )
    pv_map = {}
    if pv_regex:
        for series in instant_query_result(
            auth0_token,
            target_o11y_id,
            region,
            f'kube_persistentvolume_info{{persistentvolume=~"{pv_regex}"}}',
            end_ts,
            vendor=vendor,
            metrics_base=target_metrics_base,
        ):
            metric = series.get("metric", {})
            pv = metric.get("persistentvolume", "")
            if pv:
                pv_map[pv] = {
                    "storageclass": metric.get("storageclass", ""),
                    "volume_id": metric.get("csi_volume_handle", ""),
                    "host_path": metric.get("host_path", ""),
                    "host_path_type": metric.get("host_path_type", ""),
                }

    attachment_map = build_volume_attachment_map(
        auth0_token,
        target_o11y_id,
        region,
        pv_map.keys(),
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    )

    capacity_map = {}
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kubelet_volume_stats_capacity_bytes{{namespace="tidb-serverless", persistentvolumeclaim=~"{pvc_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        try:
            capacity_map[pvc] = {
                "node": metric_node_name(metric),
                "value": float(series.get("value", [0, 0])[1]),
            }
        except (TypeError, ValueError):
            continue

    available_map = {}
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kubelet_volume_stats_available_bytes{{namespace="tidb-serverless", persistentvolumeclaim=~"{pvc_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        try:
            available_map[pvc] = {
                "node": metric_node_name(metric),
                "value": float(series.get("value", [0, 0])[1]),
            }
        except (TypeError, ValueError):
            continue

    local_path_nodes = {
        claim.get("node", "")
        for claim in claims.values()
        if claim.get("node")
        and (
            claim.get("storageclass") == "local-path"
            or pv_map.get(claim.get("persistentvolume", ""), {}).get("storageclass") == "local-path"
            or pv_map.get(claim.get("persistentvolume", ""), {}).get("host_path", "")
        )
    }
    local_path_fs_map = build_node_local_path_filesystem_map(
        auth0_token,
        target_o11y_id,
        region,
        local_path_nodes,
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    )
    local_path_disk_map = build_node_instance_store_map(
        auth0_token,
        target_o11y_id,
        region,
        local_path_nodes,
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    )

    disk_map = build_node_disk_map(
        auth0_token,
        target_o11y_id,
        region,
        [entry.get("volume_id", "") for entry in pv_map.values()],
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    )

    tikv_node_set = set(tikv_nodes.values())
    node_types = {}
    if tikv_node_set:
        for series in instant_query_result(
            auth0_token,
            target_o11y_id,
            region,
            'kube_node_labels{label_node_kubernetes_io_instance_type!=""}',
            end_ts,
            vendor=vendor,
            metrics_base=target_metrics_base,
        ):
            metric = series.get("metric", {})
            node = metric_node_name(metric)
            instance_type = metric_instance_type(metric)
            if node in tikv_node_set and instance_type:
                node_types[node] = instance_type

    disks_by_role = {}
    for pvc, claim in claims.items():
        pv = claim.get("persistentvolume", "")
        pv_info = pv_map.get(pv, {})
        attachment_info = attachment_map.get(pv, {})
        node = claim.get("node") or capacity_map.get(pvc, {}).get("node") or available_map.get(pvc, {}).get("node")
        storageclass = claim.get("storageclass") or pv_info.get("storageclass", "")
        role = claim.get("role") or "unknown"
        is_local_path = storageclass == "local-path" or bool(pv_info.get("host_path", ""))
        volume_id = normalize_volume_id(pv_info.get("volume_id", ""), vendor=vendor)
        disk_info = lookup_disk_info(disk_map, volume_id, vendor=vendor)
        capacity = capacity_map.get(pvc, {}).get("value")
        available = available_map.get(pvc, {}).get("value")
        mount_info = {}

        if is_local_path:
            mount_info = local_path_fs_map.get(node, {})
            local_device = device_name_from_path(mount_info.get("device_path", ""))
            local_disk_candidates = local_path_disk_map.get(node, {})
            disk_info = (
                local_disk_candidates.get(local_device)
                or next(iter(local_disk_candidates.values()), {})
            )
            volume_id = ""
            attachment_info = {}
            if mount_info:
                capacity = mount_info.get("capacity_bytes")
                available = mount_info.get("available_bytes")
        used, usage_pct = capacity_usage(capacity, available)

        group = disks_by_role.setdefault(
            role,
            {
                "role": role,
                "mappings": [],
                "capacity_bytes": 0.0,
                "used_bytes": 0.0,
                "storageclasses": set(),
                "node_instance_types": set(),
            },
        )
        group["mappings"].append({
            "disk_name": pvc,
            "pod": claim.get("pod", ""),
            "node": node,
            "pvc": pvc,
            "persistentvolume": pv,
            "storageclass": storageclass,
            "volume_id": volume_id,
            "volumeattachment": attachment_info.get("volumeattachment", ""),
            "attach_node": attachment_info.get("attach_node", ""),
            "linux_device_path": (
                mount_info.get("device_path", "")
                or attachment_info.get("linux_device_path", "")
            ),
            "host_path": pv_info.get("host_path", ""),
            "mountpoint": mount_info.get("mountpoint", ""),
            "instance": metric_node_name({"instance": disk_info.get("instance", "")}) or node,
            "device": disk_info.get("device", "") or device_name_from_path(mount_info.get("device_path", "")),
            "serial": disk_info.get("serial", ""),
            "model": disk_info.get("model", ""),
            "capacity_bytes": capacity,
            "available_bytes": available,
            "used_bytes": used,
            "usage_pct": usage_pct,
        })
        if storageclass:
            group["storageclasses"].add(storageclass)
        if node and node in node_types:
            group["node_instance_types"].add(node_types[node])
        if capacity is not None:
            group["capacity_bytes"] += capacity
        if used is not None:
            group["used_bytes"] += used

    role_order = {"tikv": 0, "tikv-raft-db": 1, "tikv-raft-wal": 2}
    disks = []
    for role, group in sorted(disks_by_role.items(), key=lambda item: (role_order.get(item[0], 99), item[0])):
        unique_targets = sorted({
            (mapping.get("instance", ""), mapping.get("device", ""))
            for mapping in group["mappings"]
            if mapping.get("instance") and mapping.get("device")
        })
        disk_targets = [{"instance": instance, "device": device} for instance, device in unique_targets]
        per_disk_stats = collect_node_disk_stats_by_target(
            auth0_token,
            target_o11y_id,
            region,
            disk_targets,
            end_ts,
            vendor=vendor,
            metrics_base=target_metrics_base,
            window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        attach_mapping_disk_stats(group["mappings"], per_disk_stats)

        storageclass = sorted(group["storageclasses"])[0] if group["storageclasses"] else ""
        spec = parse_storageclass_spec(storageclass)
        if group["node_instance_types"]:
            spec["node_instance_types"] = sorted(group["node_instance_types"])
        spec["disk_count"] = len(disk_targets)
        if group["capacity_bytes"] and disk_targets:
            spec["avg_capacity_bytes_per_disk"] = group["capacity_bytes"] / len(disk_targets)

        disk = {
            "role": role,
            "spec": spec,
            "mappings": group["mappings"],
            "capacity_bytes": group["capacity_bytes"] or None,
            "used_bytes": group["used_bytes"] or 0.0,
            "usage_pct": (
                (group["used_bytes"] / group["capacity_bytes"] * 100)
                if group["capacity_bytes"] > 0 else None
            ),
        }

        if not disk_targets:
            disk["read_iops"] = empty_stats()
            disk["write_iops"] = empty_stats()
            disk["read_latency_ms"] = empty_stats()
            disk["write_latency_ms"] = empty_stats()
            disk["read_bandwidth_mib_s"] = empty_stats()
            disk["write_bandwidth_mib_s"] = empty_stats()
            disk["util_pct"] = empty_stats()
            disks.append(disk)
            continue

        read_iops_expr = build_disk_sum_rate_expr("node_disk_reads_completed_total", disk_targets)
        write_iops_expr = build_disk_sum_rate_expr("node_disk_writes_completed_total", disk_targets)
        read_bw_expr = f"({build_disk_sum_rate_expr('node_disk_read_bytes_total', disk_targets)}) / 1048576"
        write_bw_expr = f"({build_disk_sum_rate_expr('node_disk_written_bytes_total', disk_targets)}) / 1048576"
        read_latency_expr = (
            "1000 * "
            f"({build_disk_sum_rate_expr('node_disk_read_time_seconds_total', disk_targets)}) / "
            f"clamp_min(({read_iops_expr}), 1e-9)"
        )
        write_latency_expr = (
            "1000 * "
            f"({build_disk_sum_rate_expr('node_disk_write_time_seconds_total', disk_targets)}) / "
            f"clamp_min(({write_iops_expr}), 1e-9)"
        )
        util_expr = f"100 * ({build_disk_util_expr(disk_targets)})"

        disk["read_iops"] = collect_expr_stats(
            auth0_token, target_o11y_id, region, read_iops_expr, end_ts, vendor=vendor,
            metrics_base=target_metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        disk["write_iops"] = collect_expr_stats(
            auth0_token, target_o11y_id, region, write_iops_expr, end_ts, vendor=vendor,
            metrics_base=target_metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        disk["read_latency_ms"] = collect_expr_stats(
            auth0_token, target_o11y_id, region, read_latency_expr, end_ts, vendor=vendor,
            metrics_base=target_metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        disk["write_latency_ms"] = collect_expr_stats(
            auth0_token, target_o11y_id, region, write_latency_expr, end_ts, vendor=vendor,
            metrics_base=target_metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        disk["read_bandwidth_mib_s"] = collect_expr_stats(
            auth0_token, target_o11y_id, region, read_bw_expr, end_ts, vendor=vendor,
            metrics_base=target_metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        disk["write_bandwidth_mib_s"] = collect_expr_stats(
            auth0_token, target_o11y_id, region, write_bw_expr, end_ts, vendor=vendor,
            metrics_base=target_metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        disk["util_pct"] = collect_expr_stats(
            auth0_token, target_o11y_id, region, util_expr, end_ts, vendor=vendor,
            metrics_base=target_metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        disk["metric_sources"] = {
            "read_iops": "node_disk",
            "write_iops": "node_disk",
            "read_latency": "node_disk",
            "write_latency": "node_disk",
            "read_bandwidth": "node_disk",
            "write_bandwidth": "node_disk",
            "util": "node_disk",
        }
        disk["latency_note"] = "node_disk service-time latency is an average, not histogram p95 latency"
        disks.append(disk)

    return {
        "target_kind": disk_report_target_kind(cluster_info),
        "target_o11y_id": target_o11y_id,
        "disks": disks,
    }


def stats_have_values(stats):
    if not isinstance(stats, dict):
        return False
    return any(stats.get(key) is not None for key in ("current", "avg_1h", "peak_1h"))


def build_volume_attachment_map(auth0_token, o11y_id, region, persistentvolumes, ts=None, vendor="aws", metrics_base=None):
    pv_regex = regex_union(persistentvolumes)
    if not pv_regex:
        return {}

    attachment_by_pv = {}
    attachment_ids = set()
    for series in instant_query_result(
        auth0_token,
        o11y_id,
        region,
        f'kube_volumeattachment_spec_source_persistentvolume{{volumename=~"{pv_regex}"}}',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    ):
        metric = series.get("metric", {})
        pv = metric.get("volumename", "")
        attachment = metric.get("volumeattachment", "")
        if not pv or not attachment:
            continue
        attachment_by_pv[pv] = {
            "volumeattachment": attachment,
        }
        attachment_ids.add(attachment)

    if not attachment_ids:
        return {}

    attachment_regex = regex_union(attachment_ids)
    attachment_info = {}
    for series in instant_query_result(
        auth0_token,
        o11y_id,
        region,
        f'kube_volumeattachment_info{{volumeattachment=~"{attachment_regex}"}}',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    ):
        metric = series.get("metric", {})
        attachment = metric.get("volumeattachment", "")
        if attachment:
            attachment_info.setdefault(attachment, {})["attach_node"] = metric.get("node", "")

    for series in instant_query_result(
        auth0_token,
        o11y_id,
        region,
        f'kube_volumeattachment_status_attachment_metadata{{volumeattachment=~"{attachment_regex}"}}',
        ts,
        vendor=vendor,
        metrics_base=metrics_base,
    ):
        metric = series.get("metric", {})
        attachment = metric.get("volumeattachment", "")
        if attachment:
            attachment_info.setdefault(attachment, {})["linux_device_path"] = metric.get("metadata_device_path", "")

    result = {}
    for pv, info in attachment_by_pv.items():
        attachment = info.get("volumeattachment", "")
        result[pv] = {
            **info,
            **attachment_info.get(attachment, {}),
        }
    return result


def collect_node_disk_stats(
    auth0_token,
    o11y_id,
    region,
    disk_targets,
    ts=None,
    vendor="aws",
    metrics_base=None,
    window="1h",
    subquery_step_seconds=60,
):
    if not disk_targets:
        return {
            "read_iops": empty_stats(),
            "write_iops": empty_stats(),
            "read_latency_ms": empty_stats(),
            "write_latency_ms": empty_stats(),
            "read_bandwidth_mib_s": empty_stats(),
            "write_bandwidth_mib_s": empty_stats(),
            "util_pct": empty_stats(),
        }

    read_iops_expr = build_disk_sum_rate_expr("node_disk_reads_completed_total", disk_targets)
    write_iops_expr = build_disk_sum_rate_expr("node_disk_writes_completed_total", disk_targets)
    read_bw_expr = f"({build_disk_sum_rate_expr('node_disk_read_bytes_total', disk_targets)}) / 1048576"
    write_bw_expr = f"({build_disk_sum_rate_expr('node_disk_written_bytes_total', disk_targets)}) / 1048576"
    read_latency_expr = (
        "1000 * "
        f"({build_disk_sum_rate_expr('node_disk_read_time_seconds_total', disk_targets)}) / "
        f"clamp_min(({read_iops_expr}), 1e-9)"
    )
    write_latency_expr = (
        "1000 * "
        f"({build_disk_sum_rate_expr('node_disk_write_time_seconds_total', disk_targets)}) / "
        f"clamp_min(({write_iops_expr}), 1e-9)"
    )
    util_expr = f"100 * ({build_disk_util_expr(disk_targets)})"

    return {
        "read_iops": collect_expr_stats(
            auth0_token, o11y_id, region, read_iops_expr, ts,
            vendor=vendor, metrics_base=metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        ),
        "write_iops": collect_expr_stats(
            auth0_token, o11y_id, region, write_iops_expr, ts,
            vendor=vendor, metrics_base=metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        ),
        "read_latency_ms": collect_expr_stats(
            auth0_token, o11y_id, region, read_latency_expr, ts,
            vendor=vendor, metrics_base=metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        ),
        "write_latency_ms": collect_expr_stats(
            auth0_token, o11y_id, region, write_latency_expr, ts,
            vendor=vendor, metrics_base=metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        ),
        "read_bandwidth_mib_s": collect_expr_stats(
            auth0_token, o11y_id, region, read_bw_expr, ts,
            vendor=vendor, metrics_base=metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        ),
        "write_bandwidth_mib_s": collect_expr_stats(
            auth0_token, o11y_id, region, write_bw_expr, ts,
            vendor=vendor, metrics_base=metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        ),
        "util_pct": collect_expr_stats(
            auth0_token, o11y_id, region, util_expr, ts,
            vendor=vendor, metrics_base=metrics_base, window=window,
            subquery_step_seconds=subquery_step_seconds,
        ),
    }


def collect_node_disk_stats_by_target(
    auth0_token,
    o11y_id,
    region,
    disk_targets,
    ts=None,
    vendor="aws",
    metrics_base=None,
    window="1h",
    subquery_step_seconds=60,
):
    targets = [
        target for target in disk_targets
        if target.get("instance") and target.get("device")
    ]
    if not targets:
        return {}

    read_iops_expr = build_disk_target_rate_expr("node_disk_reads_completed_total", targets)
    write_iops_expr = build_disk_target_rate_expr("node_disk_writes_completed_total", targets)
    read_bw_expr = f"({build_disk_target_rate_expr('node_disk_read_bytes_total', targets)}) / 1048576"
    write_bw_expr = f"({build_disk_target_rate_expr('node_disk_written_bytes_total', targets)}) / 1048576"
    read_latency_expr = (
        "1000 * "
        f"({build_disk_target_rate_expr('node_disk_read_time_seconds_total', targets)}) / "
        f"clamp_min(({read_iops_expr}), 1e-9)"
    )
    write_latency_expr = (
        "1000 * "
        f"({build_disk_target_rate_expr('node_disk_write_time_seconds_total', targets)}) / "
        f"clamp_min(({write_iops_expr}), 1e-9)"
    )
    util_expr = f"100 * ({build_disk_target_util_expr(targets)})"

    read_iops_by_target = collect_labeled_expr_stats(
        auth0_token, o11y_id, region, read_iops_expr, ("instance", "device"), ts,
        vendor=vendor, metrics_base=metrics_base, window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    write_iops_by_target = collect_labeled_expr_stats(
        auth0_token, o11y_id, region, write_iops_expr, ("instance", "device"), ts,
        vendor=vendor, metrics_base=metrics_base, window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    read_latency_by_target = collect_labeled_expr_stats(
        auth0_token, o11y_id, region, read_latency_expr, ("instance", "device"), ts,
        vendor=vendor, metrics_base=metrics_base, window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    write_latency_by_target = collect_labeled_expr_stats(
        auth0_token, o11y_id, region, write_latency_expr, ("instance", "device"), ts,
        vendor=vendor, metrics_base=metrics_base, window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    read_bw_by_target = collect_labeled_expr_stats(
        auth0_token, o11y_id, region, read_bw_expr, ("instance", "device"), ts,
        vendor=vendor, metrics_base=metrics_base, window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    write_bw_by_target = collect_labeled_expr_stats(
        auth0_token, o11y_id, region, write_bw_expr, ("instance", "device"), ts,
        vendor=vendor, metrics_base=metrics_base, window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    util_by_target = collect_labeled_expr_stats(
        auth0_token, o11y_id, region, util_expr, ("instance", "device"), ts,
        vendor=vendor, metrics_base=metrics_base, window=window,
        subquery_step_seconds=subquery_step_seconds,
    )

    stats_by_target = {}
    for target in targets:
        key = disk_target_key(target.get("instance", ""), target.get("device", ""))
        stats_by_target[key] = {
            "read_iops": read_iops_by_target.get(key, empty_stats()),
            "write_iops": write_iops_by_target.get(key, empty_stats()),
            "read_latency_ms": read_latency_by_target.get(key, empty_stats()),
            "write_latency_ms": write_latency_by_target.get(key, empty_stats()),
            "read_bandwidth_mib_s": read_bw_by_target.get(key, empty_stats()),
            "write_bandwidth_mib_s": write_bw_by_target.get(key, empty_stats()),
            "util_pct": util_by_target.get(key, empty_stats()),
            "metric_sources": {
                "read_iops": "node_disk",
                "write_iops": "node_disk",
                "read_latency": "node_disk",
                "write_latency": "node_disk",
                "read_bandwidth": "node_disk",
                "write_bandwidth": "node_disk",
                "util": "node_disk",
            },
            "latency_note": "node_disk service-time latency is an average, not histogram p95 latency",
        }
    return stats_by_target


def attach_mapping_disk_stats(mappings, stats_by_target):
    for mapping in mappings or []:
        stats = stats_by_target.get(disk_target_key(mapping.get("instance", ""), mapping.get("device", "")))
        if not stats:
            mapping["read_iops"] = empty_stats()
            mapping["write_iops"] = empty_stats()
            mapping["read_latency_ms"] = empty_stats()
            mapping["write_latency_ms"] = empty_stats()
            mapping["read_bandwidth_mib_s"] = empty_stats()
            mapping["write_bandwidth_mib_s"] = empty_stats()
            mapping["util_pct"] = empty_stats()
            mapping["metric_sources"] = {}
            mapping["latency_note"] = None
            continue
        mapping["read_iops"] = stats["read_iops"]
        mapping["write_iops"] = stats["write_iops"]
        mapping["read_latency_ms"] = stats["read_latency_ms"]
        mapping["write_latency_ms"] = stats["write_latency_ms"]
        mapping["read_bandwidth_mib_s"] = stats["read_bandwidth_mib_s"]
        mapping["write_bandwidth_mib_s"] = stats["write_bandwidth_mib_s"]
        mapping["util_pct"] = stats["util_pct"]
        mapping["metric_sources"] = stats["metric_sources"]
        mapping["latency_note"] = stats["latency_note"]


def collect_tikv_pvc_disk_report(
    auth0_token,
    cluster_info,
    end_ts,
    window="1h",
    subquery_step_seconds=60,
):
    target_o11y_id = cluster_info.get("infra_o11y_id") or cluster_info.get("o11y_id")
    target_metrics_base = cluster_info.get("infra_metrics_base") or cluster_info.get("metrics_base")
    region = cluster_info["region"]
    vendor = cluster_info["vendor"]
    namespace = cluster_info.get("namespace") or f'tidb{cluster_info.get("cluster_id", "")}'
    pvc_pattern = "((data|raftlog)-.*-tikv-.*|tikv-.*-tikv-.*)"

    claims = {}
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kube_persistentvolumeclaim_info{{namespace="{namespace}", persistentvolumeclaim=~"{pvc_pattern}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        if not pvc:
            continue
        claims[pvc] = {
            "role": "RAFTLOG" if pvc.startswith("raftlog-") else "DATA",
            "pvc": pvc,
            "storageclass": metric.get("storageclass", ""),
            "persistentvolume": metric.get("volumename", ""),
        }

    if not claims:
        return {
            "info": f"No TiKV PVCs found in namespace {namespace} for AWS disk analysis",
            "target_kind": disk_report_target_kind(cluster_info),
            "target_o11y_id": target_o11y_id,
            "disks": [],
        }

    pvc_regex = regex_union(claims.keys())
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kube_pod_spec_volumes_persistentvolumeclaims_info{{namespace="{namespace}", persistentvolumeclaim=~"{pvc_regex}", pod=~".*-tikv-.*"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        pod = metric.get("pod", "")
        if pvc in claims and pod:
            claims[pvc]["pod"] = pod

    pod_regex = regex_union(claim.get("pod", "") for claim in claims.values())
    pod_nodes = {}
    if pod_regex:
        for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kube_pod_info{{namespace="{namespace}", pod=~"{pod_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
            metric = series.get("metric", {})
            pod = metric.get("pod", "")
            node = metric.get("node", "")
            if pod and node:
                pod_nodes[pod] = node

    pv_map = {}
    pv_regex = regex_union(
        claim.get("persistentvolume", "")
        for claim in claims.values()
        if claim.get("persistentvolume")
    )
    if pv_regex:
        for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kube_persistentvolume_info{{persistentvolume=~"{pv_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
            metric = series.get("metric", {})
            pv = metric.get("persistentvolume", "")
            if not pv:
                continue
            pv_map[pv] = {
                "storageclass": metric.get("storageclass", ""),
                "volume_id": normalize_volume_id(metric.get("csi_volume_handle", ""), vendor=vendor),
            }

    attachment_map = build_volume_attachment_map(
        auth0_token,
        target_o11y_id,
        region,
        pv_map.keys(),
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    )

    capacity_map = {}
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kubelet_volume_stats_capacity_bytes{{namespace="{namespace}", persistentvolumeclaim=~"{pvc_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        try:
            capacity_map[pvc] = {
                "node": metric_node_name(metric),
                "value": float(series.get("value", [0, 0])[1]),
            }
        except (TypeError, ValueError):
            continue

    available_map = {}
    for series in instant_query_result(
        auth0_token,
        target_o11y_id,
        region,
        f'kubelet_volume_stats_available_bytes{{namespace="{namespace}", persistentvolumeclaim=~"{pvc_regex}"}}',
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    ):
        metric = series.get("metric", {})
        pvc = metric.get("persistentvolumeclaim", "")
        try:
            available_map[pvc] = {
                "node": metric_node_name(metric),
                "value": float(series.get("value", [0, 0])[1]),
            }
        except (TypeError, ValueError):
            continue

    disk_map = build_node_disk_map(
        auth0_token,
        target_o11y_id,
        region,
        [entry.get("volume_id", "") for entry in pv_map.values()],
        end_ts,
        vendor=vendor,
        metrics_base=target_metrics_base,
    )

    node_names = {
        pod_nodes.get(claim.get("pod", "")) or capacity_map.get(pvc, {}).get("node", "") or available_map.get(pvc, {}).get("node", "")
        for pvc, claim in claims.items()
    }
    node_names.discard("")

    node_types = {}
    if node_names:
        for series in instant_query_result(
            auth0_token,
            target_o11y_id,
            region,
            'kube_node_labels{label_node_kubernetes_io_instance_type!=""}',
            end_ts,
            vendor=vendor,
            metrics_base=target_metrics_base,
        ):
            metric = series.get("metric", {})
            node = metric_node_name(metric)
            instance_type = metric_instance_type(metric)
            if node in node_names and instance_type:
                node_types[node] = instance_type

    disks_by_role = {}
    for pvc, claim in claims.items():
        pv = claim.get("persistentvolume", "")
        pv_info = pv_map.get(pv, {})
        attachment_info = attachment_map.get(pv, {})
        volume_id = normalize_volume_id(pv_info.get("volume_id", ""), vendor=vendor)
        disk_info = lookup_disk_info(disk_map, volume_id, vendor=vendor)
        capacity = capacity_map.get(pvc, {}).get("value")
        available = available_map.get(pvc, {}).get("value")
        node = (
            pod_nodes.get(claim.get("pod", ""))
            or capacity_map.get(pvc, {}).get("node", "")
            or available_map.get(pvc, {}).get("node", "")
        )
        storageclass = claim.get("storageclass") or pv_info.get("storageclass", "")
        role = claim.get("role") or "DATA"
        used, usage_pct = capacity_usage(capacity, available)

        group = disks_by_role.setdefault(
            role,
            {
                "role": role,
                "mappings": [],
                "capacity_bytes": 0.0,
                "used_bytes": 0.0,
                "storageclasses": set(),
                "node_instance_types": set(),
                "volume_ids": set(),
            },
        )
        group["mappings"].append({
            "disk_name": pvc,
            "pod": claim.get("pod", ""),
            "node": node,
            "pvc": pvc,
            "persistentvolume": pv,
            "storageclass": storageclass,
            "volume_id": volume_id,
            "volumeattachment": attachment_info.get("volumeattachment", ""),
            "attach_node": attachment_info.get("attach_node", ""),
            "linux_device_path": attachment_info.get("linux_device_path", ""),
            "instance": disk_info.get("instance") or node,
            "device": disk_info.get("device", ""),
            "serial": disk_info.get("serial", ""),
            "model": disk_info.get("model", ""),
            "capacity_bytes": capacity,
            "available_bytes": available,
            "used_bytes": used,
            "usage_pct": usage_pct,
        })
        if storageclass:
            group["storageclasses"].add(storageclass)
        if volume_id:
            group["volume_ids"].add(volume_id)
        if node and node in node_types:
            group["node_instance_types"].add(node_types[node])
        if capacity is not None:
            group["capacity_bytes"] += capacity
        if used is not None:
            group["used_bytes"] += used

    role_order = {"DATA": 0, "RAFTLOG": 1}
    disks = []
    for role, group in sorted(disks_by_role.items(), key=lambda item: (role_order.get(item[0], 99), item[0])):
        unique_targets = sorted({
            (mapping.get("instance", ""), mapping.get("device", ""))
            for mapping in group["mappings"]
            if mapping.get("instance") and mapping.get("device")
        })
        disk_targets = [{"instance": instance, "device": device} for instance, device in unique_targets]
        per_disk_stats = collect_node_disk_stats_by_target(
            auth0_token,
            target_o11y_id,
            region,
            disk_targets,
            end_ts,
            vendor=vendor,
            metrics_base=target_metrics_base,
            window=window,
            subquery_step_seconds=subquery_step_seconds,
        )
        attach_mapping_disk_stats(group["mappings"], per_disk_stats)
        volume_regex = regex_union(group["volume_ids"])
        node_stats = collect_node_disk_stats(
            auth0_token,
            target_o11y_id,
            region,
            disk_targets,
            end_ts,
            vendor=vendor,
            metrics_base=target_metrics_base,
            window=window,
            subquery_step_seconds=subquery_step_seconds,
        )

        storageclass = sorted(group["storageclasses"])[0] if group["storageclasses"] else ""
        spec = parse_storageclass_spec(storageclass)
        if group["node_instance_types"]:
            spec["node_instance_types"] = sorted(group["node_instance_types"])
        spec["disk_count"] = max(len(disk_targets), len(group["volume_ids"]))
        if group["capacity_bytes"] and spec["disk_count"]:
            spec["avg_capacity_bytes_per_disk"] = group["capacity_bytes"] / spec["disk_count"]

        disk = {
            "role": role,
            "spec": spec,
            "mappings": group["mappings"],
            "capacity_bytes": group["capacity_bytes"] or None,
            "used_bytes": group["used_bytes"] or 0.0,
            "usage_pct": (
                (group["used_bytes"] / group["capacity_bytes"] * 100)
                if group["capacity_bytes"] > 0 else None
            ),
        }

        aws_read_iops = empty_stats()
        aws_write_iops = empty_stats()
        aws_read_latency = empty_stats()
        aws_write_latency = empty_stats()
        aws_read_bandwidth = empty_stats()
        aws_write_bandwidth = empty_stats()
        if vendor == "aws" and volume_regex:
            read_iops_expr = build_aws_ebs_rate_expr("aws_ebs_csi_read_ops_total", volume_regex)
            write_iops_expr = build_aws_ebs_rate_expr("aws_ebs_csi_write_ops_total", volume_regex)
            read_bw_expr = f"({build_aws_ebs_rate_expr('aws_ebs_csi_read_bytes_total', volume_regex)}) / 1048576"
            write_bw_expr = f"({build_aws_ebs_rate_expr('aws_ebs_csi_write_bytes_total', volume_regex)}) / 1048576"
            read_latency_expr = build_aws_ebs_latency_expr("aws_ebs_csi_read_io_latency_seconds_bucket", volume_regex)
            write_latency_expr = build_aws_ebs_latency_expr("aws_ebs_csi_write_io_latency_seconds_bucket", volume_regex)

            aws_read_iops = collect_expr_stats(
                auth0_token, target_o11y_id, region, read_iops_expr, end_ts, vendor=vendor,
                metrics_base=target_metrics_base, window=window,
                subquery_step_seconds=subquery_step_seconds,
            )
            aws_write_iops = collect_expr_stats(
                auth0_token, target_o11y_id, region, write_iops_expr, end_ts, vendor=vendor,
                metrics_base=target_metrics_base, window=window,
                subquery_step_seconds=subquery_step_seconds,
            )
            aws_read_latency = collect_expr_stats(
                auth0_token, target_o11y_id, region, read_latency_expr, end_ts, vendor=vendor,
                metrics_base=target_metrics_base, window=window,
                subquery_step_seconds=subquery_step_seconds,
            )
            aws_write_latency = collect_expr_stats(
                auth0_token, target_o11y_id, region, write_latency_expr, end_ts, vendor=vendor,
                metrics_base=target_metrics_base, window=window,
                subquery_step_seconds=subquery_step_seconds,
            )
            aws_read_bandwidth = collect_expr_stats(
                auth0_token, target_o11y_id, region, read_bw_expr, end_ts, vendor=vendor,
                metrics_base=target_metrics_base, window=window,
                subquery_step_seconds=subquery_step_seconds,
            )
            aws_write_bandwidth = collect_expr_stats(
                auth0_token, target_o11y_id, region, write_bw_expr, end_ts, vendor=vendor,
                metrics_base=target_metrics_base, window=window,
                subquery_step_seconds=subquery_step_seconds,
            )

        disk["read_iops"] = aws_read_iops if stats_have_values(aws_read_iops) else node_stats["read_iops"]
        disk["write_iops"] = aws_write_iops if stats_have_values(aws_write_iops) else node_stats["write_iops"]
        disk["read_latency_ms"] = aws_read_latency if stats_have_values(aws_read_latency) else node_stats["read_latency_ms"]
        disk["write_latency_ms"] = aws_write_latency if stats_have_values(aws_write_latency) else node_stats["write_latency_ms"]
        disk["read_bandwidth_mib_s"] = aws_read_bandwidth if stats_have_values(aws_read_bandwidth) else node_stats["read_bandwidth_mib_s"]
        disk["write_bandwidth_mib_s"] = aws_write_bandwidth if stats_have_values(aws_write_bandwidth) else node_stats["write_bandwidth_mib_s"]
        disk["util_pct"] = node_stats["util_pct"]
        disk["metric_sources"] = {
            "read_iops": "aws_ebs" if stats_have_values(aws_read_iops) else "node_disk",
            "write_iops": "aws_ebs" if stats_have_values(aws_write_iops) else "node_disk",
            "read_latency": "aws_ebs" if stats_have_values(aws_read_latency) else "node_disk",
            "write_latency": "aws_ebs" if stats_have_values(aws_write_latency) else "node_disk",
            "read_bandwidth": "aws_ebs" if stats_have_values(aws_read_bandwidth) else "node_disk",
            "write_bandwidth": "aws_ebs" if stats_have_values(aws_write_bandwidth) else "node_disk",
            "util": "node_disk",
        }
        if disk["metric_sources"]["read_latency"] == "node_disk" or disk["metric_sources"]["write_latency"] == "node_disk":
            disk["latency_note"] = "node_disk fallback uses average service time, not aws_ebs histogram p95 latency"

        disks.append(disk)

    return {
        "target_kind": disk_report_target_kind(cluster_info),
        "target_o11y_id": target_o11y_id,
        "disks": disks,
    }


def collect_aws_tikv_disk_report(
    auth0_token,
    cluster_info,
    end_ts,
    window="1h",
    subquery_step_seconds=60,
):
    if cluster_info.get("vendor") != "aws":
        return {"info": "AWS EBS metrics are only available for AWS clusters", "disks": []}
    return collect_tikv_pvc_disk_report(
        auth0_token,
        cluster_info,
        end_ts,
        window=window,
        subquery_step_seconds=subquery_step_seconds,
    )


def normalize_premium_disk_section(disk, disk_type, storage_spec=None):
    if not disk:
        return {
            "disk_type": disk_type,
            "spec": storage_spec or {},
            "mappings": [],
            "disks": [],
            "capacity_bytes": None,
            "used_bytes": None,
            "usage_pct": None,
            "read_iops": empty_stats(),
            "write_iops": empty_stats(),
            "read_latency_p95_ms": empty_stats(),
            "write_latency_p95_ms": empty_stats(),
            "read_bandwidth_mib_s": empty_stats(),
            "write_bandwidth_mib_s": empty_stats(),
            "util_pct": empty_stats(),
        }
    return {
        "disk_type": disk_type,
        "spec": storage_spec or disk.get("spec", {}),
        "mappings": disk.get("mappings", []),
        "disks": normalize_disk_inventory(disk.get("mappings", [])),
        "capacity_bytes": disk.get("capacity_bytes"),
        "used_bytes": disk.get("used_bytes"),
        "usage_pct": disk.get("usage_pct"),
        "read_iops": disk.get("read_iops", empty_stats()),
        "write_iops": disk.get("write_iops", empty_stats()),
        "read_latency_p95_ms": disk.get("read_latency_ms", empty_stats()),
        "write_latency_p95_ms": disk.get("write_latency_ms", empty_stats()),
        "read_bandwidth_mib_s": disk.get("read_bandwidth_mib_s", empty_stats()),
        "write_bandwidth_mib_s": disk.get("write_bandwidth_mib_s", empty_stats()),
        "util_pct": disk.get("util_pct", empty_stats()),
        "metric_sources": disk.get("metric_sources", {}),
        "latency_note": disk.get("latency_note"),
    }


def build_serverless_tikv_disk_report(
    auth0_token,
    cluster_id,
    end_ts,
    window="1h",
    subquery_step_seconds=60,
    hint_region=None,
    hint_vendor=None,
):
    cluster_snapshot = fetch_clinic_cluster_snapshot(cluster_id) or {}
    snapshot_vendor = cluster_snapshot.get("vendor") or cluster_snapshot.get("provider") or ""
    vendor_hint = hint_vendor or snapshot_vendor or "aws"
    if snapshot_vendor and snapshot_vendor != vendor_hint:
        auth0_token = authenticate(snapshot_vendor)
        vendor_hint = snapshot_vendor

    region_hint = hint_region or cluster_snapshot.get("region")
    cluster = find_cluster(auth0_token, cluster_id, region_hint, vendor_hint, end_ts)
    deploy_type = (cluster.get("deploy_type", "") or cluster_snapshot.get("deployType") or "").lower()
    biz_type = cluster.get("biz_type", "")
    if "devtier" not in biz_type and deploy_type not in ("starter", "essential", "shared"):
        raise RuntimeError(
            f"build_serverless_tikv_disk_report only supports serverless/essential/starter clusters. "
            f"Got deployType='{deploy_type}', bizType='{biz_type}'."
        )

    disk_report = collect_serverless_disk_report(
        auth0_token,
        cluster,
        end_ts,
        window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    disks_by_role = {disk.get("role"): disk for disk in disk_report.get("disks", [])}

    data_disk = normalize_premium_disk_section(disks_by_role.get("tikv"), "DATA")
    raft_db_disk = normalize_premium_disk_section(disks_by_role.get("tikv-raft-db"), "RAFT_DB")
    raft_wal_disk = normalize_premium_disk_section(disks_by_role.get("tikv-raft-wal"), "RAFT_WAL")

    return {
        "generated_at": datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "cluster": {
            "cluster_id": cluster_id,
            "cluster_name": cluster_snapshot.get("name", ""),
            "deploy_type": deploy_type or "shared",
            "biz_type": biz_type,
            "vendor": cluster.get("vendor", ""),
            "region": cluster.get("region", ""),
            "org_id": cluster.get("org_id", "") or cluster_snapshot.get("orgID", ""),
            "tenant_id": cluster.get("tenant_id", "") or cluster_snapshot.get("tenantID", ""),
            "project_id": cluster.get("project_id", "") or cluster_snapshot.get("projectID", ""),
            "namespace": cluster.get("namespace", ""),
            "keyspace_id": cluster.get("keyspace_id", ""),
        },
        "targets": {
            "infra_uuid": disk_report.get("target_o11y_id", "") or cluster.get("infra_o11y_id", ""),
            "infra_metrics_base": cluster.get("infra_metrics_base", "") or cluster.get("metrics_base", ""),
        },
        "window": {
            "lookback": window,
            "time_ts": end_ts,
            "subquery_step_seconds": subquery_step_seconds,
        },
        "disks": {
            "data": data_disk,
            "raft_db": raft_db_disk,
            "raft_wal": raft_wal_disk,
        },
    }


def build_premium_tikv_disk_report(
    auth0_token,
    cluster_id,
    end_ts,
    window="1h",
    subquery_step_seconds=60,
    hint_region=None,
    hint_vendor=None,
):
    cluster_snapshot = fetch_clinic_cluster_snapshot(cluster_id)
    if not cluster_snapshot:
        raise RuntimeError(f"Cluster not found in Clinic: {cluster_id}")

    deploy_type = (cluster_snapshot.get("deployType") or cluster_snapshot.get("deploy_type") or "").lower()
    if deploy_type != "premium":
        raise RuntimeError(
            f"build_premium_tikv_disk_report only supports premium clusters. Got deployType='{deploy_type}'."
        )

    snapshot_vendor = cluster_snapshot.get("vendor") or cluster_snapshot.get("provider") or ""
    vendor_hint = hint_vendor or snapshot_vendor or "aws"
    if snapshot_vendor and snapshot_vendor != vendor_hint:
        auth0_token = authenticate(snapshot_vendor)
        vendor_hint = snapshot_vendor
    region_hint = hint_region or cluster_snapshot.get("region")
    cluster = find_cluster(auth0_token, cluster_id, region_hint, vendor_hint, end_ts)

    actual_vendor = cluster.get("vendor", "")

    topology = fetch_clinic_topology(cluster_snapshot)
    tikv_component = next((entry for entry in topology if entry.get("component_type") == "TIKV"), None)
    if not tikv_component:
        raise RuntimeError(f"Failed to locate TIKV component in Clinic topology for cluster {cluster_id}")

    tikv_spec = build_tikv_component_spec(tikv_component)
    storage_specs = {
        storage.get("name", ""): storage
        for storage in tikv_spec.get("storages", [])
        if storage.get("name")
    }

    disk_report = collect_tikv_pvc_disk_report(
        auth0_token,
        cluster,
        end_ts,
        window=window,
        subquery_step_seconds=subquery_step_seconds,
    )
    disks_by_role = {disk.get("role"): disk for disk in disk_report.get("disks", [])}

    data_disk = normalize_premium_disk_section(
        disks_by_role.get("DATA"), "DATA", storage_spec=storage_specs.get("DATA")
    )
    raftlog_disk = normalize_premium_disk_section(
        disks_by_role.get("RAFTLOG"), "RAFTLOG", storage_spec=storage_specs.get("RAFTLOG")
    )

    return {
        "generated_at": datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "cluster": {
            "cluster_id": cluster_id,
            "cluster_name": cluster_snapshot.get("name", ""),
            "deploy_type": "premium",
            "vendor": cluster.get("vendor", ""),
            "region": cluster.get("region", ""),
            "org_id": cluster.get("org_id", "") or cluster_snapshot.get("orgID", ""),
            "tenant_id": cluster.get("tenant_id", "") or cluster_snapshot.get("tenantID", ""),
            "parent_id": cluster.get("parent_id", "") or cluster_snapshot.get("parentID", ""),
            "data_plane_name": cluster.get("data_plane_name", "") or cluster_snapshot.get("dataPlaneName", ""),
            "namespace": cluster.get("namespace", ""),
        },
        "targets": {
            "infra_uuid": disk_report.get("target_o11y_id", "") or cluster.get("infra_o11y_id", ""),
            "infra_metrics_base": cluster.get("infra_metrics_base", "") or cluster.get("metrics_base", ""),
        },
        "window": {
            "lookback": window,
            "time_ts": end_ts,
            "subquery_step_seconds": subquery_step_seconds,
        },
        "tikv_spec": tikv_spec,
        "disks": {
            "data": data_disk,
            "raftlog": raftlog_disk,
        },
    }


def collect_log_stats(cluster_id, region, start_ts, end_ts, vendor="aws"):
    """Download log-stats SQLite files and analyze.
    If no data is found for the requested range, automatically falls back
    to the previous hour (log-stats are generated hourly with ~30min delay).
    Only available for AWS (S3). GCP/alicloud clusters should use Loki.
    """
    if vendor != "aws":
        return {"info": f"Log-stats SQLite not available via {vendor} storage; use Loki logs instead"}

    try:
        import boto3
    except ImportError:
        return {"info": "boto3 not installed — skipping S3 log-stats (pip install boto3)"}

    s3 = boto3.client("s3", region_name=region)
    bucket = f"o11y-prod-shared-{region}"

    start_dt = datetime.fromtimestamp(start_ts, tz=timezone.utc)
    end_dt = datetime.fromtimestamp(end_ts, tz=timezone.utc)

    hours = []
    current = start_dt.replace(minute=0, second=0, microsecond=0)
    while current <= end_dt:
        hours.append(current.strftime("%Y%m%d%H"))
        current += timedelta(hours=1)

    all_stats = []
    downloaded_hours = []

    for hour in hours:
        key = f"diagnosis/data/{cluster_id}/log-stats/1h/{hour}.db"
        tmp_path = os.path.join(tempfile.gettempdir(), f"logstats_{cluster_id}_{hour}.db")

        try:
            s3.download_file(bucket, key, tmp_path)
            downloaded_hours.append(hour)
        except Exception:
            continue

        try:
            conn = sqlite3.connect(tmp_path)
            cur = conn.cursor()
            cur.execute("SELECT component, level, position, msg_digest, count, sample FROM log_stats")
            rows = cur.fetchall()
            for row in rows:
                all_stats.append({
                    "component": row[0],
                    "level": row[1],
                    "position": row[2],
                    "msg_digest": row[3],
                    "count": row[4],
                    "sample": (row[5] or "")[:200],
                    "hour": hour,
                })
            conn.close()
        except Exception as e:
            all_stats.append({"error": f"Failed to parse {hour}.db: {e}"})
        finally:
            try:
                os.unlink(tmp_path)
            except OSError:
                pass

    # Fallback: if no data found, try the hour before the requested range
    fallback_used = False
    if not downloaded_hours:
        fb_dt = start_dt.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1)
        fb_hour = fb_dt.strftime("%Y%m%d%H")
        fb_key = f"diagnosis/data/{cluster_id}/log-stats/1h/{fb_hour}.db"
        fb_path = os.path.join(tempfile.gettempdir(), f"logstats_{cluster_id}_{fb_hour}.db")
        try:
            s3.download_file(bucket, fb_key, fb_path)
            downloaded_hours.append(fb_hour)
            hours.append(fb_hour)
            fallback_used = True
            conn = sqlite3.connect(fb_path)
            cur = conn.cursor()
            cur.execute("SELECT component, level, position, msg_digest, count, sample FROM log_stats")
            for row in cur.fetchall():
                all_stats.append({
                    "component": row[0], "level": row[1], "position": row[2],
                    "msg_digest": row[3], "count": row[4],
                    "sample": (row[5] or "")[:200], "hour": fb_hour,
                })
            conn.close()
        except Exception:
            pass
        finally:
            try:
                os.unlink(fb_path)
            except OSError:
                pass

    if not all_stats:
        return {
            "hours_checked": hours,
            "hours_found": downloaded_hours,
            "summary_by_component_level": {},
            "top_errors": [],
            "info": "No log-stats data found for the time range (including fallback)"
        }

    summary = {}
    for s in all_stats:
        if "error" in s:
            continue
        key = f"{s['component']}_{s['level']}"
        summary[key] = summary.get(key, 0) + s["count"]

    valid_stats = [s for s in all_stats if "error" not in s]
    errors_warns = [s for s in valid_stats if s["level"] in ("ERROR", "WARN")]
    errors_warns.sort(key=lambda x: x["count"], reverse=True)

    top_errors = []
    for e in errors_warns[:20]:
        top_errors.append({
            "component": e["component"],
            "level": e["level"],
            "count": e["count"],
            "msg": (e["msg_digest"] or "")[:120],
            "position": (e["position"] or "")[:60],
            "sample": e["sample"][:200],
        })

    result = {
        "hours_checked": hours,
        "hours_found": downloaded_hours,
        "summary_by_component_level": summary,
        "top_errors": top_errors,
    }
    if fallback_used:
        result["fallback_info"] = f"No data for requested range; used previous hour {downloaded_hours[0]}"
    return result


def collect_slowlogs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts):
    """Fallback: collect slowlogs from Loki when S3/DeltaLake is unavailable."""
    logql = f'{{cluster_id="{cluster_id}", container="slowlog"}}'
    resp = query_loki(auth0_token, region, vendor, logql, start_ts, end_ts, limit=200)

    streams = resp.get("data", {}).get("result", [])
    total_entries = sum(len(s.get("values", [])) for s in streams)

    if total_entries == 0:
        return {
            "source": "loki",
            "total_count": 0,
            "info": "No slowlogs found in Loki for this cluster",
            "top_slow_queries": [],
        }

    sample_entries = []
    for stream in streams:
        for ts_val, line in stream.get("values", [])[:50]:
            sample_entries.append(line[:300])

    return {
        "source": "loki",
        "total_count": total_entries,
        "info": f"Showing {len(sample_entries)} sample entries from Loki (Delta Lake not available for {vendor})",
        "sample_entries": sample_entries[:10],
        "top_slow_queries": [],
    }


def _analyse_slowlog_db(db, start_ts, end_ts, source_label="deltalake"):
    """Run standard slowlog analytics on a DuckDB connection that already
    has a view called ``__tbl__`` containing the parquet rows."""
    try:
        count_row = db.execute(
            "SELECT COUNT(*) FROM __tbl__ WHERE time >= ? AND time < ?",
            [start_ts, end_ts],
        ).fetchone()
        total_count = count_row[0] if count_row else 0

        if total_count == 0:
            db.close()
            return {"source": source_label, "total_count": 0,
                    "info": "No slowlogs in time range", "top_slow_queries": []}

        top_slow = db.execute("""
            SELECT
                digest,
                first(query) as sample_query,
                avg(query_time) as avg_time,
                max(query_time) as max_time,
                count(*) as cnt,
                avg(mem_max) as avg_mem
            FROM __tbl__
            WHERE time >= ? AND time < ?
            GROUP BY digest
            ORDER BY avg_time DESC
            LIMIT 10
        """, [start_ts, end_ts]).fetchall()

        top_queries = []
        for row in top_slow:
            top_queries.append({
                "digest": row[0] or "",
                "sample_query": (row[1] or "")[:200],
                "avg_time_seconds": round(row[2] or 0, 4),
                "max_time_seconds": round(row[3] or 0, 4),
                "count": row[4],
                "avg_mem_bytes": int(row[5] or 0),
            })

        p99_row = db.execute("""
            SELECT approx_quantile(query_time, 0.99)
            FROM __tbl__ WHERE time >= ? AND time < ?
        """, [start_ts, end_ts]).fetchone()
        p99 = round(p99_row[0], 4) if p99_row and p99_row[0] else 0

        db.close()
        return {"source": source_label, "total_count": total_count,
                "p99_seconds": p99, "top_slow_queries": top_queries}
    except Exception as e:
        return {"source": source_label, "error": str(e), "top_slow_queries": []}


def collect_slowlogs(o11y_id, tenant_id, region, vendor, start_ts, end_ts,
                     auth0_token=None, cluster_id=None):
    """Query Delta Lake slowlogs via DuckDB.

    - AWS / GCP: use delta_scan directly against S3 / GCS.
    - AliCloud: download parquet files from OSS via boto3, read locally.
    - Others: fall back to Loki.
    """
    try:
        import duckdb
    except ImportError:
        if auth0_token and cluster_id:
            return collect_slowlogs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts)
        return {"error": "duckdb not installed (pip install duckdb)", "top_slow_queries": []}

    # --- AliCloud: download parquet from OSS → local DuckDB ---
    if vendor == "alicloud":
        try:
            s3_client, bucket = create_storage_client("alicloud", region)
            prefix = f"deltalake/{tenant_id}/{o11y_id}/slowlogs/"
            conn, tmp_files = read_deltalake_parquet(
                s3_client, bucket, prefix, "%Y%m%d", start_ts, end_ts, "", max_files=20,
            )
            if conn is None:
                _cleanup_tmp_files(tmp_files)
                if auth0_token and cluster_id:
                    return collect_slowlogs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts)
                return {"source": "oss-deltalake", "total_count": 0,
                        "info": "No slowlog parquet files found in OSS", "top_slow_queries": []}
            result = _analyse_slowlog_db(conn, start_ts, end_ts, source_label="oss-deltalake")
            _cleanup_tmp_files(tmp_files)
            return result
        except Exception as e:
            if auth0_token and cluster_id:
                return collect_slowlogs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts)
            return {"source": "oss-deltalake", "error": str(e), "top_slow_queries": []}

    if vendor not in ("aws", "gcp"):
        if auth0_token and cluster_id:
            return collect_slowlogs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts)
        return {"info": f"Delta Lake slowlogs not available for {vendor}; use Loki instead",
                "top_slow_queries": []}

    # --- AWS / GCP: use delta_scan ---
    try:
        db = duckdb.connect(database=":memory:", read_only=False)
        db.execute("INSTALL httpfs; LOAD httpfs;")
        db.execute("INSTALL delta; LOAD delta;")

        if vendor == "aws":
            db.execute("INSTALL aws; LOAD aws;")
            db.execute(
                f"CREATE OR REPLACE SECRET secret "
                f"(TYPE s3, PROVIDER credential_chain, ENDPOINT 's3.{region}.amazonaws.com');"
            )
            delta_url = f"s3://o11y-prod-shared-{region}/deltalake/{tenant_id}/{o11y_id}/slowlogs"
        elif vendor == "gcp":
            db.execute("INSTALL gcs; LOAD gcs;")
            delta_url = f"gs://o11y-prod-shared-{region}/deltalake/{tenant_id}/{o11y_id}/slowlogs"
        else:
            return {"error": f"Unsupported vendor for slowlogs: {vendor}", "top_slow_queries": []}

        db.execute(f"CREATE VIEW __tbl__ AS SELECT * FROM delta_scan('{delta_url}')")
        return _analyse_slowlog_db(db, start_ts, end_ts, source_label="deltalake")

    except Exception as e:
        return {"error": str(e), "top_slow_queries": []}


def collect_statement_logs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts):
    """Fallback: collect statement logs from Loki when S3 is unavailable."""
    logql = f'{{cluster_id="{cluster_id}", container="statementlog"}}'
    resp = query_loki(auth0_token, region, vendor, logql, start_ts, end_ts, limit=100)

    streams = resp.get("data", {}).get("result", [])
    total_entries = sum(len(s.get("values", [])) for s in streams)

    if total_entries == 0:
        return {
            "source": "loki",
            "info": "No statement logs found in Loki",
            "top_statements": [],
        }

    sample_entries = []
    for stream in streams:
        for _, line in stream.get("values", [])[:20]:
            sample_entries.append(line[:300])

    return {
        "source": "loki",
        "total_entries": total_entries,
        "info": f"Showing samples from Loki (S3 not available for {vendor})",
        "sample_entries": sample_entries[:10],
        "top_statements": [],
    }


def collect_statement_logs_s3(cluster_id, tenant_id, project_id, region, start_ts, end_ts):
    """Collect SQL statement statistics from S3 statementlog files.
    Path: 0/{tenantId}/{projectId}/{clusterId}/logs/statementlog/
    Files are vector.dev format: {unix_timestamp}-{uuid}.log.gz
    Each line: timestamp;host;cluster;instance;statementlog {JSON}
    """
    try:
        import boto3
        import gzip
    except ImportError:
        return {"error": "boto3 not installed — skipping S3 statements", "source": "s3", "top_statements": []}

    s3 = boto3.client("s3", region_name=region)
    bucket = f"o11y-prod-shared-{region}"
    prefix = f"0/{tenant_id}/{project_id}/{cluster_id}/logs/statementlog/"

    # Use StartAfter to skip to files within our time range
    start_key = f"{prefix}{start_ts}"

    try:
        paginator = s3.get_paginator("list_objects_v2")
        file_keys = []
        for page in paginator.paginate(Bucket=bucket, Prefix=prefix, StartAfter=start_key):
            for obj in page.get("Contents", []):
                fname = obj["Key"].split("/")[-1]
                file_ts = fname.split("-")[0] if "-" in fname else ""
                if file_ts.isdigit() and int(file_ts) <= end_ts:
                    file_keys.append(obj["Key"])
                elif file_ts.isdigit() and int(file_ts) > end_ts:
                    break
            if file_keys and file_ts.isdigit() and int(file_ts) > end_ts:
                break
    except Exception as e:
        return {"error": f"Failed to list statementlog files: {e}", "source": "s3"}

    if not file_keys:
        return {"source": "s3", "path": prefix, "files_found": 0,
                "info": "No statementlog files in time range", "top_statements": []}

    # Parse up to 10 files to avoid excessive download
    max_files = min(len(file_keys), 10)
    stmts_by_digest = {}

    for key in file_keys[:max_files]:
        try:
            resp = s3.get_object(Bucket=bucket, Key=key)
            data = gzip.decompress(resp["Body"].read())
            for line in data.decode("utf-8", errors="replace").split("\n"):
                if not line or "statementlog {" not in line:
                    continue
                json_start = line.find("statementlog {") + len("statementlog ")
                try:
                    obj = json.loads(line[json_start:])
                except (json.JSONDecodeError, ValueError):
                    continue

                digest = obj.get("digest", "")
                if not digest:
                    continue
                if digest not in stmts_by_digest:
                    stmts_by_digest[digest] = {
                        "digest": digest,
                        "stmt_type": obj.get("stmt_type", ""),
                        "schema_name": obj.get("schema_name", ""),
                        "normalized_sql": (obj.get("normalized_sql", "") or "")[:200],
                        "total_exec_count": 0,
                        "total_sum_latency_ns": 0,
                        "max_latency_ns": 0,
                        "max_mem": 0,
                    }
                entry = stmts_by_digest[digest]
                entry["total_exec_count"] += obj.get("exec_count", 0)
                entry["total_sum_latency_ns"] += obj.get("sum_latency", 0)
                entry["max_latency_ns"] = max(entry["max_latency_ns"], obj.get("max_latency", 0))
                entry["max_mem"] = max(entry["max_mem"], obj.get("max_mem", 0))
        except Exception:
            continue

    # Sort by total latency
    all_stmts = sorted(stmts_by_digest.values(), key=lambda x: -x["total_sum_latency_ns"])
    top = []
    for s in all_stmts[:15]:
        avg_lat_ms = (s["total_sum_latency_ns"] / max(s["total_exec_count"], 1)) / 1e6
        top.append({
            "digest": s["digest"],
            "stmt_type": s["stmt_type"],
            "schema_name": s["schema_name"],
            "normalized_sql": s["normalized_sql"],
            "exec_count": s["total_exec_count"],
            "avg_latency_ms": round(avg_lat_ms, 2),
            "max_latency_ms": round(s["max_latency_ns"] / 1e6, 2),
            "max_mem_bytes": s["max_mem"],
        })

    return {
        "source": "s3",
        "path": prefix,
        "files_found": len(file_keys),
        "files_parsed": max_files,
        "unique_digests": len(stmts_by_digest),
        "top_statements": top,
    }


def _analyse_statement_db(db, start_ts, end_ts, source_label="deltalake"):
    """Run standard statement analytics on a DuckDB connection that already
    has a view called ``__tbl__`` with the parquet rows."""
    try:
        col_rows = db.execute("SELECT column_name, column_type FROM (DESCRIBE SELECT * FROM __tbl__)").fetchall()
        col_map = {r[0].upper(): (r[0], r[1]) for r in col_rows}

        def pick(upper_name, fallback):
            if upper_name in col_map:
                return col_map[upper_name][0]
            return fallback

        digest_col = pick("DIGEST", "digest")
        text_col = pick("DIGEST_TEXT", "digest_text")
        type_col = pick("STMT_TYPE", "stmt_type")
        schema_col = pick("SCHEMA_NAME", "schema_name")
        exec_col = pick("EXEC_COUNT", "exec_count")
        avg_lat_col = pick("AVG_LATENCY", "avg_latency")
        max_lat_col = pick("MAX_LATENCY", "max_latency")
        avg_mem_col = pick("AVG_MEM", "avg_mem")

        # Prefer SUMMARY_BEGIN_TIME (proper TIMESTAMP) over _vector_timestamp (string)
        if "SUMMARY_BEGIN_TIME" in col_map:
            ts_col = pick("SUMMARY_BEGIN_TIME", "SUMMARY_BEGIN_TIME")
            start_iso = datetime.fromtimestamp(start_ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00")
            end_iso = datetime.fromtimestamp(end_ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00")
            time_filter = f"{ts_col} >= TIMESTAMPTZ '{start_iso}' AND {ts_col} < TIMESTAMPTZ '{end_iso}'"
        else:
            ts_col = pick("_VECTOR_TIMESTAMP", "_vector_timestamp")
            start_iso = datetime.fromtimestamp(start_ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
            end_iso = datetime.fromtimestamp(end_ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
            time_filter = f"{ts_col} >= '{start_iso}' AND {ts_col} < '{end_iso}'"

        top_rows = db.execute(f"""
            SELECT
                {digest_col},
                first({text_col}) as sql_text,
                first({type_col}) as stmt_type,
                first({schema_col}) as schema_name,
                SUM({exec_col}) as total_exec,
                AVG({avg_lat_col}) as avg_latency_ns,
                MAX({max_lat_col}) as max_latency_ns,
                AVG({avg_mem_col}) as avg_mem
            FROM __tbl__
            WHERE {time_filter}
            GROUP BY {digest_col}
            ORDER BY avg_latency_ns DESC
            LIMIT 15
        """).fetchall()

        top = []
        for row in top_rows:
            mem_val = int(row[7] or 0)
            top.append({
                "digest": row[0] or "",
                "normalized_sql": (row[1] or "")[:200],
                "stmt_type": row[2] or "",
                "schema_name": row[3] or "",
                "exec_count": int(row[4] or 0),
                "avg_latency_ms": round((row[5] or 0) / 1e6, 2),
                "max_latency_ms": round((row[6] or 0) / 1e6, 2),
                "avg_mem_bytes": mem_val,
                "max_mem_bytes": mem_val,
            })

        db.close()
        return {"source": source_label, "top_statements": top}
    except Exception as e:
        return {"source": source_label, "error": str(e), "top_statements": []}


def collect_statement_logs_deltalake(o11y_id, tenant_id, region, vendor, start_ts, end_ts):
    """Query Delta Lake persisted_statements_summary for Premium clusters.
    Path: deltalake/{tenantId}/{o11yId}/persisted_statements_summary/

    - AWS: use delta_scan against S3
    - AliCloud: download parquet from OSS, read locally
    """
    try:
        import duckdb
    except ImportError:
        return {"error": "duckdb not installed", "source": "deltalake", "top_statements": []}

    prefix = f"deltalake/{tenant_id}/{o11y_id}/persisted_statements_summary/"

    # --- AliCloud: download parquet from OSS → local DuckDB ---
    if vendor == "alicloud":
        try:
            s3_client, bucket = create_storage_client("alicloud", region)
            conn, tmp_files = read_deltalake_parquet(
                s3_client, bucket, prefix, "%Y-%m-%d", start_ts, end_ts, "", max_files=20,
            )
            if conn is None:
                _cleanup_tmp_files(tmp_files)
                return {"source": "oss-deltalake", "info": "No statement parquet files in OSS",
                        "top_statements": []}
            result = _analyse_statement_db(conn, start_ts, end_ts, source_label="oss-deltalake")
            _cleanup_tmp_files(tmp_files)
            return result
        except Exception as e:
            return {"source": "oss-deltalake", "error": str(e), "top_statements": []}

    # --- AWS / GCP: use delta_scan ---
    if vendor not in ("aws", "gcp"):
        return {"source": "deltalake", "error": f"Unsupported vendor: {vendor}",
                "top_statements": []}

    bucket = f"o11y-prod-shared-{region}"
    delta_path = f"deltalake/{tenant_id}/{o11y_id}/persisted_statements_summary"

    if vendor == "aws":
        try:
            import boto3
            s3 = boto3.client("s3", region_name=region)
            resp = s3.list_objects_v2(Bucket=bucket, Prefix=f"{delta_path}/", MaxKeys=1)
            if resp.get("KeyCount", 0) == 0:
                return {"source": "deltalake", "info": "No persisted_statements_summary table found",
                        "top_statements": []}
        except ImportError:
            pass
        except Exception:
            pass

    try:
        db = duckdb.connect(database=":memory:", read_only=False)
        db.execute("INSTALL httpfs; LOAD httpfs;")
        db.execute("INSTALL delta; LOAD delta;")

        if vendor == "aws":
            db.execute("INSTALL aws; LOAD aws;")
            db.execute(
                f"CREATE OR REPLACE SECRET secret "
                f"(TYPE s3, PROVIDER credential_chain, ENDPOINT 's3.{region}.amazonaws.com');"
            )
            delta_url = f"s3://{bucket}/{delta_path}"
        elif vendor == "gcp":
            db.execute("INSTALL gcs; LOAD gcs;")
            delta_url = f"gs://{bucket}/{delta_path}"
        else:
            db.close()
            return {"source": "deltalake", "error": f"Unsupported vendor: {vendor}",
                    "top_statements": []}

        db.execute(f"CREATE VIEW __tbl__ AS SELECT * FROM delta_scan('{delta_url}')")
        return _analyse_statement_db(db, start_ts, end_ts, source_label="deltalake")

    except Exception as e:
        return {"source": "deltalake", "error": str(e), "top_statements": []}


def collect_statements(cluster_info, start_ts, end_ts, auth0_token=None):
    """Dispatch to the correct statement collection method based on cluster type."""
    cluster_id = cluster_info["application_id"]
    tenant_id = cluster_info["tenant_id"]
    project_id = cluster_info["project_id"]
    o11y_id = cluster_info["o11y_id"]
    region = cluster_info["region"]
    vendor = cluster_info["vendor"]
    biz_type = cluster_info.get("biz_type", "")
    deploy_type = cluster_info.get("deploy_type", "")

    is_premium = "nextgen" in biz_type or deploy_type in ("premium", "resource_pool")
    is_serverless = "devtier" in biz_type or deploy_type in ("starter", "essential")

    # Premium clusters: always try Delta Lake first (works for AWS, GCP, and AliCloud)
    if is_premium:
        result = collect_statement_logs_deltalake(o11y_id, tenant_id, region, vendor, start_ts, end_ts)
        if result.get("top_statements"):
            return result
        # For AWS, also try S3 raw files as fallback
        if vendor == "aws":
            return collect_statement_logs_s3(cluster_id, tenant_id, project_id, region, start_ts, end_ts)
        if auth0_token:
            return collect_statement_logs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts)
        return result

    # Non-AWS/GCP non-premium: try Loki
    if vendor not in ("aws", "gcp"):
        if auth0_token:
            return collect_statement_logs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts)
        return {"info": f"Statement logs S3/DeltaLake not available for {vendor}",
                "top_statements": []}

    if is_serverless:
        for env in ("prod", "staging", "dev"):
            prefix = f"0/{env}-devtier/serverless/{tenant_id}/{project_id}/{cluster_id}/logs/"
            try:
                import boto3
                s3 = boto3.client("s3", region_name=region)
                resp = s3.list_objects_v2(
                    Bucket=f"o11y-prod-shared-{region}", Prefix=prefix, Delimiter="/", MaxKeys=10)
                dirs = [p["Prefix"].rstrip("/").split("/")[-1] for p in resp.get("CommonPrefixes", [])]
                if "statementlog" in dirs or "statement" in dirs:
                    stmt_dir = "statementlog" if "statementlog" in dirs else "statement"
                    return _collect_serverless_statements(
                        tenant_id, project_id, cluster_id, region, env, stmt_dir, start_ts, end_ts)
            except ImportError:
                if auth0_token:
                    return collect_statement_logs_loki(auth0_token, cluster_id, region, vendor, start_ts, end_ts)
                return {"info": "boto3 not installed — skipping", "top_statements": []}
            except Exception:
                continue
        return {"source": "serverless", "info": "No statement logs found for serverless cluster",
                "top_statements": []}

    return collect_statement_logs_s3(cluster_id, tenant_id, project_id, region, start_ts, end_ts)


def _collect_serverless_statements(tenant_id, project_id, cluster_id, region, env, stmt_dir, start_ts, end_ts):
    """Collect statement logs from Essential/Starter S3 path with vector.dev file naming."""
    import boto3
    import gzip

    s3 = boto3.client("s3", region_name=region)
    bucket = f"o11y-prod-shared-{region}"
    prefix = f"0/{env}-devtier/serverless/{tenant_id}/{project_id}/{cluster_id}/logs/{stmt_dir}/"

    start_key = f"{prefix}{start_ts}"
    try:
        paginator = s3.get_paginator("list_objects_v2")
        file_keys = []
        for page in paginator.paginate(Bucket=bucket, Prefix=prefix, StartAfter=start_key):
            for obj in page.get("Contents", []):
                fname = obj["Key"].split("/")[-1]
                file_ts = fname.split("-")[0] if "-" in fname else ""
                if file_ts.isdigit() and int(file_ts) <= end_ts:
                    file_keys.append(obj["Key"])
                elif file_ts.isdigit() and int(file_ts) > end_ts:
                    break
            if file_keys and file_ts.isdigit() and int(file_ts) > end_ts:
                break
    except Exception as e:
        return {"source": "serverless", "error": str(e), "top_statements": []}

    if not file_keys:
        return {"source": "serverless", "path": prefix, "files_found": 0,
                "info": "No statement files in time range", "top_statements": []}

    max_files = min(len(file_keys), 10)
    stmts_by_digest = {}

    for key in file_keys[:max_files]:
        try:
            resp = s3.get_object(Bucket=bucket, Key=key)
            raw = resp["Body"].read()
            data = gzip.decompress(raw) if key.endswith(".gz") else raw
            for line in data.decode("utf-8", errors="replace").split("\n"):
                if not line:
                    continue
                # Try to find JSON payload (may be prefixed with vector metadata or bare JSON)
                json_start = -1
                for marker in ("statementlog {", "statement {"):
                    pos = line.find(marker)
                    if pos >= 0:
                        json_start = pos + len(marker) - 1
                        break
                if json_start < 0 and line.lstrip().startswith("{"):
                    json_start = line.index("{")
                if json_start < 0:
                    continue
                try:
                    obj = json.loads(line[json_start:])
                except (json.JSONDecodeError, ValueError):
                    continue

                digest = obj.get("digest", obj.get("DIGEST", ""))
                if not digest:
                    continue
                if digest not in stmts_by_digest:
                    stmts_by_digest[digest] = {
                        "digest": digest,
                        "stmt_type": obj.get("stmt_type", obj.get("STMT_TYPE", "")),
                        "schema_name": obj.get("schema_name", obj.get("SCHEMA_NAME", "")),
                        "normalized_sql": (obj.get("normalized_sql", obj.get("DIGEST_TEXT", "")) or "")[:200],
                        "total_exec_count": 0,
                        "total_sum_latency_ns": 0,
                        "max_latency_ns": 0,
                        "max_mem": 0,
                    }
                entry = stmts_by_digest[digest]
                entry["total_exec_count"] += obj.get("exec_count", obj.get("EXEC_COUNT", 0)) or 0
                entry["total_sum_latency_ns"] += obj.get("sum_latency", obj.get("SUM_LATENCY", 0)) or 0
                ml = obj.get("max_latency", obj.get("MAX_LATENCY", 0)) or 0
                entry["max_latency_ns"] = max(entry["max_latency_ns"], ml)
                entry["max_mem"] = max(entry["max_mem"], obj.get("max_mem", obj.get("MAX_MEM", 0)) or 0)
        except Exception:
            continue

    all_stmts = sorted(stmts_by_digest.values(), key=lambda x: -x["total_sum_latency_ns"])
    top = []
    for s in all_stmts[:15]:
        avg_lat_ms = (s["total_sum_latency_ns"] / max(s["total_exec_count"], 1)) / 1e6
        top.append({
            "digest": s["digest"],
            "stmt_type": s["stmt_type"],
            "schema_name": s["schema_name"],
            "normalized_sql": s["normalized_sql"],
            "exec_count": s["total_exec_count"],
            "avg_latency_ms": round(avg_lat_ms, 2),
            "max_latency_ms": round(s["max_latency_ns"] / 1e6, 2),
            "max_mem_bytes": s["max_mem"],
        })

    return {
        "source": "serverless",
        "path": prefix,
        "files_found": len(file_keys),
        "files_parsed": max_files,
        "unique_digests": len(stmts_by_digest),
        "top_statements": top,
    }


def loki_api_base(region, vendor):
    return f"https://www.ds.{region}.{vendor}.observability.tidbcloud.com/loki/self-monitoring/loki/api/v1"


def query_loki(auth0_token, region, vendor, logql, start_ts, end_ts, limit=100):
    """Execute a Loki query_range and return parsed result."""
    url = f"{loki_api_base(region, vendor)}/query_range"
    headers = {"Authorization": f"Bearer {auth0_token}"}
    params = {
        "query": logql,
        "start": f"{start_ts}.0",
        "end": f"{end_ts}.0",
        "limit": limit,
    }
    try:
        r = requests.get(url, headers=headers, params=params, timeout=30)
        r.raise_for_status()
        return r.json()
    except Exception as e:
        try:
            return fetch_json_with_curl(url, headers=headers, params=params)
        except Exception:
            return {"error": str(e)}


def collect_loki_logs(auth0_token, cluster_id, region, vendor, start_ts, end_ts, deploy_type=""):
    """Collect recent log summary from Loki for a cluster.
    For premium clusters, shared components (tikv/pd/tiflash) use namespace-based queries.
    """
    result = {"source": "loki", "containers": {}}

    is_premium = deploy_type == "premium"
    physical_ns = ""

    tidb_logql = f'{{cluster_id="{cluster_id}", container="tidb"}}'
    tidb_resp = query_loki(auth0_token, region, vendor, tidb_logql, start_ts, end_ts, limit=10)
    tidb_streams = tidb_resp.get("data", {}).get("result", [])
    tidb_entries = sum(len(s.get("values", [])) for s in tidb_streams)
    tidb_info = {"stream_count": len(tidb_streams), "entry_count": tidb_entries}

    if tidb_streams:
        physical_ns = tidb_streams[0].get("stream", {}).get("namespace", "")

    error_logql = f'{{cluster_id="{cluster_id}", container="tidb"}} |= "ERROR"'
    err_resp = query_loki(auth0_token, region, vendor, error_logql, start_ts, end_ts, limit=10)
    err_streams = err_resp.get("data", {}).get("result", [])
    tidb_info["error_entries_sample"] = sum(len(s.get("values", [])) for s in err_streams)
    result["containers"]["tidb"] = tidb_info

    shared_containers = ("tikv", "pd", "tiflash")
    for container in shared_containers:
        if is_premium and physical_ns:
            logql = f'{{namespace="{physical_ns}", container="{container}"}}'
        else:
            logql = f'{{cluster_id="{cluster_id}", container="{container}"}}'

        resp = query_loki(auth0_token, region, vendor, logql, start_ts, end_ts, limit=10)
        streams = resp.get("data", {}).get("result", [])
        entry_count = sum(len(s.get("values", [])) for s in streams)
        result["containers"][container] = {"stream_count": len(streams), "entry_count": entry_count}

    for container in ("slowlog",):
        logql = f'{{cluster_id="{cluster_id}", container="{container}"}}'
        resp = query_loki(auth0_token, region, vendor, logql, start_ts, end_ts, limit=10)
        streams = resp.get("data", {}).get("result", [])
        entry_count = sum(len(s.get("values", [])) for s in streams)
        result["containers"][container] = {"stream_count": len(streams), "entry_count": entry_count}

    if physical_ns:
        result["physical_namespace"] = physical_ns

    return result


def detect_anomalies(metrics, log_stats, slowlogs, disk_report=None):
    """Analyze collected data and flag anomalies."""
    anomalies = []

    conn = metrics.get("connections", 0)
    if conn > 10000:
        anomalies.append({"severity": "WARN", "type": "high_connections",
                          "msg": f"Connection count: {conn} (threshold: 10,000)"})

    p99 = metrics.get("latency_p99_seconds", 0)
    if p99 > 1.0:
        anomalies.append({"severity": "WARN", "type": "high_latency",
                          "msg": f"P99 latency: {p99:.3f}s (threshold: 1s)"})

    summary = log_stats.get("summary_by_component_level", {})
    for key, count in summary.items():
        if "_ERROR" in key and count > 100:
            comp = key.replace("_ERROR", "")
            anomalies.append({"severity": "ERROR", "type": "log_errors",
                              "msg": f"{comp} ERROR: {count} occurrences"})
        elif "_WARN" in key and count > 1000:
            comp = key.replace("_WARN", "")
            anomalies.append({"severity": "WARN", "type": "log_warnings",
                              "msg": f"{comp} WARN: {count} occurrences"})

    sl_total = slowlogs.get("total_count", 0)
    if sl_total > 10000:
        anomalies.append({"severity": "WARN", "type": "high_slowlog_count",
                          "msg": f"Slowlog count: {sl_total} in time range"})

    sl_p99 = slowlogs.get("p99_seconds", 0)
    if sl_p99 > 5.0:
        anomalies.append({"severity": "WARN", "type": "high_slowlog_p99",
                          "msg": f"Slowlog P99: {sl_p99:.3f}s"})

    for disk in (disk_report or {}).get("disks", []):
        peak_util = disk.get("util_pct", {}).get("peak_1h")
        if peak_util and peak_util > 80:
            anomalies.append({
                "severity": "WARN",
                "type": "high_disk_util",
                "msg": f"{disk.get('role', 'disk')} peak disk util: {peak_util:.1f}%",
            })

    return anomalies


def fmt_float(value, digits=2):
    if value is None:
        return "N/A"
    return f"{value:.{digits}f}"


def fmt_triplet(stats, digits=2):
    if not stats:
        return "N/A"
    return (
        f"{fmt_float(stats.get('current'), digits)} / "
        f"{fmt_float(stats.get('avg_1h'), digits)} / "
        f"{fmt_float(stats.get('peak_1h'), digits)}"
    )


def fmt_gib(value):
    if value is None:
        return "N/A"
    return f"{value / (1024 ** 3):.2f} GiB"


def print_report(report):
    """Print human-readable report."""
    c = report["cluster"]
    print()
    print("=" * 72)
    print(f"  Cluster Inspection Report: {c['application_id']}")
    print("=" * 72)
    print(f"  O11Y ID    : {c['o11y_id']}")
    print(f"  Region     : {c['region']} ({c['vendor']})")
    print(f"  Deploy Type: {c.get('deploy_type', 'N/A')}")
    print(f"  Version    : {c.get('version', 'N/A')}")
    print(f"  Time Range : {report['time_range']}")
    print()

    m = report["metrics"]
    print("── Metrics ──────────────────────────────────────────────")
    print(f"  QPS Total      : {m['qps']['total']:,.0f}")
    top_types = sorted(m["qps"]["by_type"].items(), key=lambda x: -x[1])[:5]
    for t, v in top_types:
        if v > 0:
            print(f"    {t:20s}: {v:,.0f}")
    print(f"  P99 Latency    : {m['latency_p99_seconds']:.4f}s")
    print(f"  Connections    : {m['connections']:,}")
    print(f"  TiKV CPU Total : {m['tikv_cpu_total']:.1f} cores")
    regions = m.get("tikv_regions", {})
    if regions:
        print(f"  TiKV Regions   : {regions}")
    print()

    disk_report = report.get("disk_report", {})
    disks = disk_report.get("disks", [])
    if disks:
        print("── Disk ─────────────────────────────────────────────────")
        print(f"  Metrics Target : {disk_report.get('target_kind', 'unknown')} ({disk_report.get('target_o11y_id', 'N/A')})")
        print("  Values         : current / avg1h / peak1h")
        for disk in disks:
            spec = disk.get("spec", {})
            storageclass = spec.get("storageclass", "N/A")
            instance_types = ", ".join(spec.get("node_instance_types", [])) or "N/A"
            print(f"  {disk.get('role', 'unknown')}:")
            print(f"    Spec         : {storageclass} on {instance_types}")
            print(f"    Capacity     : {fmt_gib(disk.get('capacity_bytes'))}")
            print(f"    Used         : {fmt_gib(disk.get('used_bytes'))} ({fmt_float(disk.get('usage_pct'))}%)")
            print(f"    Read IOPS    : {fmt_triplet(disk.get('read_iops'), 2)}")
            print(f"    Write IOPS   : {fmt_triplet(disk.get('write_iops'), 2)}")
            print(f"    Read Lat ms  : {fmt_triplet(disk.get('read_latency_ms'), 3)}")
            print(f"    Write Lat ms : {fmt_triplet(disk.get('write_latency_ms'), 3)}")
            print(f"    Read BW MiB/s: {fmt_triplet(disk.get('read_bandwidth_mib_s'), 3)}")
            print(f"    Write BW MiB/s: {fmt_triplet(disk.get('write_bandwidth_mib_s'), 3)}")
            print(f"    Util %       : {fmt_triplet(disk.get('util_pct'), 2)}")
        print()

    ls = report["log_stats"]
    print("── Log Stats ────────────────────────────────────────────")
    print(f"  Hours checked : {len(ls.get('hours_checked', []))}")
    print(f"  Hours found   : {len(ls.get('hours_found', []))}")
    if ls.get("fallback_info"):
        print(f"  Note          : {ls['fallback_info']}")
    summary = ls.get("summary_by_component_level", {})
    if summary:
        print("  Summary:")
        for key in sorted(summary, key=lambda k: -summary[k]):
            print(f"    {key:20s}: {summary[key]:,}")
    top_errs = ls.get("top_errors", [])
    if top_errs:
        print(f"\n  Top Errors/Warnings (showing {min(len(top_errs), 10)}):")
        for i, e in enumerate(top_errs[:10], 1):
            print(f"    {i:2d}. [{e['level']:5s}] {e['component']:8s} count={e['count']:>6,}  {e['msg'][:70]}")
            if e.get("position"):
                print(f"        at {e['position']}")
    else:
        print(f"  {ls.get('info', 'No data')}")
    print()

    loki = report.get("loki_logs", {})
    print("── Loki Logs ────────────────────────────────────────────")
    if loki.get("error"):
        print(f"  Error: {loki['error']}")
    else:
        containers = loki.get("containers", {})
        for name, info in containers.items():
            streams = info.get("stream_count", 0)
            entries = info.get("entry_count", 0)
            extra = ""
            if "error_entries_sample" in info:
                extra = f"  (errors sample: {info['error_entries_sample']})"
            print(f"  {name:12s}: {streams} streams, {entries} entries{extra}")
    print()

    sl = report["slowlogs"]
    print("── Slowlogs ─────────────────────────────────────────────")
    if "error" in sl:
        print(f"  Error: {sl['error']}")
    elif sl.get("total_count", 0) == 0:
        print(f"  {sl.get('info', 'No slowlogs')}")
    else:
        print(f"  Total Count : {sl['total_count']:,}")
        print(f"  P99 Time    : {sl.get('p99_seconds', 0):.4f}s")
        top_sq = sl.get("top_slow_queries", [])
        if top_sq:
            print("\n  Top Slow Queries (by avg time):")
            for i, q in enumerate(top_sq[:5], 1):
                print(f"    {i}. avg={q['avg_time_seconds']:.3f}s  max={q['max_time_seconds']:.3f}s  count={q['count']}  mem={q['avg_mem_bytes']//1024//1024}MB")
                print(f"       {q['sample_query'][:100]}")
    print()

    st = report.get("statements", {})
    print("── SQL Statements ───────────────────────────────────────")
    if "error" in st:
        print(f"  Error: {st['error']}")
    elif not st.get("top_statements"):
        print(f"  {st.get('info', 'No statement data available')}")
    else:
        src = st.get("source", "unknown")
        print(f"  Source       : {src}")
        if st.get("files_found") is not None:
            print(f"  Files found  : {st['files_found']} (parsed {st.get('files_parsed', '?')})")
        if st.get("unique_digests") is not None:
            print(f"  SQL Digests  : {st['unique_digests']}")
        top_st = st["top_statements"]
        print(f"\n  Top SQL Statements (by latency, showing {min(len(top_st), 10)}):")
        for i, s in enumerate(top_st[:10], 1):
            print(f"    {i:2d}. [{s['stmt_type']:6s}] exec={s['exec_count']:>6,}  "
                  f"avg={s['avg_latency_ms']:.1f}ms  max={s['max_latency_ms']:.1f}ms  "
                  f"mem={s.get('max_mem_bytes', s.get('avg_mem_bytes', 0))//1024//1024}MB")
            print(f"        {s.get('schema_name', '')}: {s['normalized_sql'][:80]}")
    print()

    anomalies = report["anomalies"]
    print("── Anomalies ────────────────────────────────────────────")
    if not anomalies:
        print("  No anomalies detected ✓")
    else:
        print(f"  {len(anomalies)} anomaly(ies) detected:")
        for a in anomalies:
            print(f"    [{a['severity']:5s}] {a['msg']}")
    print()
    print("=" * 72)


def main():
    parser = argparse.ArgumentParser(description="O11Y Cluster Inspection")
    parser.add_argument("cluster_id", help="TiDB Cloud cluster ID")
    parser.add_argument("--provider", default="aws",
                        help="Cloud provider: aws, gcp, alicloud (default: aws)")
    parser.add_argument("--hours", type=float, default=1, help="Hours to look back (default: 1)")
    parser.add_argument("--region", default=None, help="Hint region (auto-detected if omitted)")
    parser.add_argument("--vendor", default=None,
                        help="Cloud vendor for API calls (defaults to provider value)")
    parser.add_argument("--output", "-o", default=None, help="Output JSON file path")
    parser.add_argument(
        "--premium-disk-report",
        action="store_true",
        help="Emit premium TiKV disk report JSON and exit",
    )
    parser.add_argument(
        "--serverless-disk-report",
        action="store_true",
        help="Emit serverless/essential TiKV disk report JSON and exit",
    )
    parser.add_argument(
        "--time-ts",
        default=None,
        help="Analysis timestamp as epoch seconds or ISO8601 (used by disk-report modes)",
    )
    parser.add_argument(
        "--window",
        default="1h",
        help="Lookback window for summary stats (used by disk-report modes, default: 1h)",
    )
    parser.add_argument(
        "--subquery-step-seconds",
        type=int,
        default=60,
        help="Subquery step seconds for summary stats (used by disk-report modes, default: 60)",
    )
    parser.add_argument(
        "--clinic-context",
        action="store_true",
        help="Emit normalized Clinic cluster context JSON and exit",
    )
    parser.add_argument(
        "--resolve-targets",
        action="store_true",
        help="Emit shell-compatible metrics target resolution JSON and exit",
    )
    args = parser.parse_args()

    if args.provider == "azure":
        print("ERROR: Azure provider is not yet supported.", file=sys.stderr)
        sys.exit(1)

    vendor = args.vendor or args.provider

    if args.clinic_context:
        context = build_clinic_cluster_context(args.cluster_id)
        if context is None:
            print(f"ERROR: Cluster not found in Clinic: {args.cluster_id}", file=sys.stderr)
            sys.exit(1)
        json.dump(context, sys.stdout, ensure_ascii=False, default=str)
        sys.stdout.write("\n")
        return

    if args.resolve_targets:
        targets = build_shell_metrics_targets(
            args.cluster_id,
            parse_time_arg(args.time_ts),
            provider=args.provider,
            hint_region=args.region,
            hint_vendor=vendor,
        )
        json.dump(targets, sys.stdout, ensure_ascii=False, default=str)
        sys.stdout.write("\n")
        return

    if args.premium_disk_report:
        end_ts = parse_time_arg(args.time_ts)
        print(f"Authenticating with provider={args.provider}...", file=sys.stderr)
        auth0_token = authenticate(args.provider)
        print(f"Building premium disk report for cluster {args.cluster_id}...", file=sys.stderr)
        report = build_premium_tikv_disk_report(
            auth0_token,
            args.cluster_id,
            end_ts,
            window=args.window,
            subquery_step_seconds=args.subquery_step_seconds,
            hint_region=args.region,
            hint_vendor=vendor,
        )
        json.dump(report, sys.stdout, ensure_ascii=False, default=str)
        sys.stdout.write("\n")
        return

    if args.serverless_disk_report:
        end_ts = parse_time_arg(args.time_ts)
        print(f"Authenticating with provider={args.provider}...", file=sys.stderr)
        auth0_token = authenticate(args.provider)
        print(f"Building serverless disk report for cluster {args.cluster_id}...", file=sys.stderr)
        report = build_serverless_tikv_disk_report(
            auth0_token,
            args.cluster_id,
            end_ts,
            window=args.window,
            subquery_step_seconds=args.subquery_step_seconds,
            hint_region=args.region,
            hint_vendor=vendor,
        )
        json.dump(report, sys.stdout, ensure_ascii=False, default=str)
        sys.stdout.write("\n")
        return

    now = int(time.time())
    start_ts = now - int(args.hours * 3600)
    end_ts = now

    time_range_str = (
        f"{datetime.fromtimestamp(start_ts, tz=timezone.utc).isoformat()} to "
        f"{datetime.fromtimestamp(end_ts, tz=timezone.utc).isoformat()}"
    )

    print(f"Authenticating with provider={args.provider}...", file=sys.stderr)
    auth0_token = authenticate(args.provider)

    print(f"Looking up cluster {args.cluster_id}...", file=sys.stderr)
    cluster = find_cluster(auth0_token, args.cluster_id, args.region, vendor, end_ts)
    actual_vendor = cluster["vendor"]
    actual_region = cluster["region"]

    if actual_vendor != args.provider:
        print(f"  Cluster is on {actual_vendor}, re-authenticating...", file=sys.stderr)
        auth0_token = authenticate(actual_vendor)

    print(f"Found: region={actual_region}, vendor={actual_vendor}, "
          f"o11y_id={cluster['o11y_id']}, deploy_type={cluster.get('deploy_type', 'N/A')}",
          file=sys.stderr)

    print("Collecting metrics...", file=sys.stderr)
    metrics = collect_metrics(
        auth0_token, cluster["o11y_id"], actual_region, start_ts, end_ts, actual_vendor,
        metrics_base=cluster.get("metrics_base"),
    )

    deploy_type = cluster.get("deploy_type", "")
    biz_type = cluster.get("biz_type", "")
    is_serverless = "devtier" in biz_type or deploy_type in ("starter", "essential")
    is_byoc = deploy_type == "byoc"

    log_stats = {}
    if not is_serverless and not is_byoc:
        print("Collecting log-stats...", file=sys.stderr)
        log_stats = collect_log_stats(args.cluster_id, actual_region, start_ts, end_ts, actual_vendor)
    else:
        log_stats = {"info": f"Log-stats not available for {deploy_type or biz_type}; use Loki instead"}

    print("Querying Loki logs...", file=sys.stderr)
    loki_logs = collect_loki_logs(
        auth0_token, args.cluster_id, actual_region, actual_vendor, start_ts, end_ts,
        deploy_type=deploy_type,
    )

    slowlogs = {}
    if not is_byoc:
        print("Collecting slowlogs...", file=sys.stderr)
        slowlogs = collect_slowlogs(
            cluster["o11y_id"], cluster["tenant_id"],
            actual_region, actual_vendor, start_ts, end_ts,
            auth0_token=auth0_token, cluster_id=args.cluster_id,
        )
    else:
        slowlogs = {"info": "Slowlogs not available for BYOC (data in user's bucket)"}

    statements = {}
    if not is_byoc:
        print("Collecting statement logs...", file=sys.stderr)
        statements = collect_statements(cluster, start_ts, end_ts, auth0_token=auth0_token)
    else:
        statements = {"info": "Statement logs not available for BYOC (data in user's bucket)"}

    disk_report = {}
    if is_serverless:
        print("Collecting serverless disk metrics...", file=sys.stderr)
        disk_report = collect_serverless_disk_report(auth0_token, cluster, end_ts)
    elif not is_byoc:
        print("Collecting disk metrics...", file=sys.stderr)
        disk_report = collect_tikv_pvc_disk_report(auth0_token, cluster, end_ts)

    anomalies = detect_anomalies(metrics, log_stats, slowlogs, disk_report=disk_report)

    report = {
        "cluster": cluster,
        "provider": args.provider,
        "time_range": time_range_str,
        "generated_at": datetime.now(tz=timezone.utc).isoformat(),
        "metrics": metrics,
        "disk_report": disk_report,
        "log_stats": log_stats,
        "loki_logs": loki_logs,
        "slowlogs": slowlogs,
        "statements": statements,
        "anomalies": anomalies,
        "summary": {
            "anomaly_count": len(anomalies),
            "status": "ANOMALIES" if anomalies else "HEALTHY",
        },
    }

    print_report(report)

    output_file = args.output or f"inspect_{args.cluster_id}_{now}.json"
    with open(output_file, "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False, default=str)
    print(f"Full report saved to: {output_file}", file=sys.stderr)


if __name__ == "__main__":
    main()