#!/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 = "http://www.gs.us-west-2.aws.observability.tidbcloud.com/api/v1/clusters/"

VENDOR_REGIONS = {
    "aws": ["us-east-1", "us-west-2", "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"]


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


# ---------------------------------------------------------------------------
# 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 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 resolve_serverless_keyspace(auth0_token, o11y_id, region, cluster_id, vendor="aws"):
    """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)
        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_nodes(auth0_token, o11y_id, region, keyspace_id, ts=None, vendor="aws"):
    """Map a serverless keyspace to the TiKV pods and nodes currently hosting it."""
    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,
    )

    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)

    tikv_nodes = {}
    for instance in sorted(set(tikv_instances)):
        pod_resp = instant_query_metrics(
            auth0_token,
            o11y_id,
            region,
            f'kube_pod_info{{namespace="tidb-serverless", pod="{instance}"}}',
            ts,
            vendor,
        )
        result = pod_resp.get("data", {}).get("result", [])
        if result:
            node = result[0].get("metric", {}).get("node", "")
            if node:
                tikv_nodes[instance] = node

    return sorted(set(tikv_instances)), tikv_nodes


def try_clinic_api(cluster_id):
    """Tier 1: Try Clinic Metadata API for instant cluster lookup."""
    load_env()
    clinic_key = os.environ.get("CLINIC_API_KEY", "")
    if not clinic_key:
        return None
    try:
        r = requests.get(
            f"https://clinic.pingcap.com/api/v1/clusters/{cluster_id}",
            headers={"Authorization": f"Bearer {clinic_key}"},
            timeout=5,
        )
        if r.status_code == 200 and r.headers.get("content-type", "").startswith("application/json"):
            return r.json()
    except Exception:
        pass
    return None


def _gs_search_one(auth0_token, cluster_id, vendor, region, biz_type):
    """Single GS API lookup. Returns cluster dict or None."""
    url = f"{GS_API_BASE}?bizType={biz_type}&vendor={vendor}&region={region}&applicationID={cluster_id}"
    try:
        r = requests.get(url, headers={"Authorization": f"Bearer {auth0_token}"}, timeout=5)
        if r.status_code != 200:
            return None
        clusters = r.json().get("data", {}).get("clusters", [])
        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
    except Exception:
        return None


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."""

    # 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)

    # 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:
                info = _gs_search_one(auth0_token, cluster_id, vendor, region, biz_type)
                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 infer_deploy_type(biz_type_value)

                    metrics_o11y_id, metrics_base = resolve_metrics_o11y_id(
                        auth0_token, info, resolved_region, resolved_vendor
                    )
                    cluster = {
                        "o11y_id": metrics_o11y_id,
                        "metrics_base": metrics_base,
                        "gs_id": info["id"],
                        "tenant_id": info["tenantID"],
                        "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,
                    }
                    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
                        )
                        tikv_instances, tikv_nodes = ([], {})
                        if keyspace_id:
                            tikv_instances, tikv_nodes = resolve_serverless_tikv_nodes(
                                auth0_token, metrics_o11y_id, resolved_region, keyspace_id, analysis_ts, resolved_vendor
                            )
                        cluster["namespace"] = "tidb-serverless"
                        cluster["keyspace_id"] = keyspace_id
                        cluster["tikv_instances"] = tikv_instances
                        cluster["tikv_nodes"] = tikv_nodes
                    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 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 (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 (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):
    """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"})

    return anomalies


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()

    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")
    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

    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)"}

    anomalies = detect_anomalies(metrics, log_stats, slowlogs)

    report = {
        "cluster": cluster,
        "provider": args.provider,
        "time_range": time_range_str,
        "generated_at": datetime.now(tz=timezone.utc).isoformat(),
        "metrics": metrics,
        "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()