Grafana Billing

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "httpx",
#     "python-dotenv",
#     "rich",
# ]
# ///
"""
Grafana Billing Metrics CLI

Query key billing metrics from Prometheus and Loki through Grafana's
data source proxy API.

Usage:
    # Query both environments (default)
    uv run billing_metrics.py

    # Query specific environment
    uv run billing_metrics.py --env staging
    uv run billing_metrics.py --env prod

    # JSON output for automation
    uv run billing_metrics.py --json

    # Filter to specific service
    uv run billing_metrics.py --service prometheus
    uv run billing_metrics.py --service loki

Environment Variables:
    GRAFANA_STAGING_API_KEY - API key for staging Grafana
    GRAFANA_PROD_API_KEY - API key for prod Grafana
"""

import argparse
import sys
from pathlib import Path

# Add scripts directory to path for sibling imports
sys.path.insert(0, str(Path(__file__).parent))

from rich.console import Console

from grafana_client import GrafanaClient, GrafanaClientError, ENVIRONMENTS
from prometheus_metrics import query_prometheus_metrics
from loki_metrics import query_loki_metrics
from formatters import print_all_metrics, format_json


def parse_args() -> argparse.Namespace:
    """Parse command line arguments."""
    parser = argparse.ArgumentParser(
        description="Query Grafana billing metrics from Prometheus and Loki",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__
    )

    parser.add_argument(
        "--env",
        choices=list(ENVIRONMENTS.keys()),
        help="Environment to query (default: all)"
    )

    parser.add_argument(
        "--service",
        choices=["prometheus", "loki"],
        help="Service to query (default: all)"
    )

    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON"
    )

    parser.add_argument(
        "--limit",
        type=int,
        default=10,
        help="Number of top cardinality entries to show (default: 10)"
    )

    return parser.parse_args()


def query_environment(
    env: str,
    service_filter: str | None = None,
    cardinality_limit: int = 10
) -> dict:
    """
    Query all metrics for a single environment.

    Args:
        env: Environment name
        service_filter: Optional filter for 'prometheus' or 'loki'
        cardinality_limit: Number of top cardinality entries

    Returns:
        Dict with prometheus and/or loki metrics
    """
    results = {}

    try:
        client = GrafanaClient(env)
    except GrafanaClientError as e:
        return {"error": str(e)}

    # Query Prometheus
    if service_filter is None or service_filter == "prometheus":
        try:
            prom_metrics = query_prometheus_metrics(client, limit=cardinality_limit)
            results["prometheus"] = prom_metrics.to_dict()
        except Exception as e:
            results["prometheus"] = {"error": str(e)}

    # Query Loki
    if service_filter is None or service_filter == "loki":
        try:
            loki_metrics = query_loki_metrics(client)
            results["loki"] = loki_metrics.to_dict()
        except Exception as e:
            results["loki"] = {"error": str(e)}

    return results


def main() -> int:
    """Main entry point."""
    args = parse_args()
    console = Console(stderr=True)

    # Determine which environments to query
    if args.env:
        envs_to_query = [args.env]
    else:
        envs_to_query = list(ENVIRONMENTS.keys())

    # Query all environments
    all_results = {}
    errors = []

    for env in envs_to_query:
        if not args.json_output:
            console.print(f"[dim]Querying {env}...[/dim]", highlight=False)

        result = query_environment(
            env,
            service_filter=args.service,
            cardinality_limit=args.limit
        )

        if "error" in result:
            errors.append(f"{env}: {result['error']}")
        else:
            all_results[env] = result

    # Handle case where all environments failed
    if not all_results:
        console.print("[red]Error: Could not query any environments[/red]")
        for error in errors:
            console.print(f"  [yellow]{error}[/yellow]")
        return 1

    # Output results
    if args.json_output:
        print(format_json(all_results))
    else:
        output_console = Console()
        print_all_metrics(all_results, output_console)

        # Print any errors that occurred
        if errors:
            console.print()
            for error in errors:
                console.print(f"[yellow]Warning: {error}[/yellow]")

    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "rich",
# ]
# ///
"""
Output formatters for billing metrics.

Provides both human-readable table output and JSON serialization.
"""

import json
from typing import Any

from rich.console import Console
from rich.table import Table


def format_bytes(value: int | float | None) -> str:
    """Format bytes as human-readable string."""
    if value is None:
        return "N/A"

    units = ["B", "KB", "MB", "GB", "TB"]
    size = float(value)
    unit_index = 0

    while size >= 1024 and unit_index < len(units) - 1:
        size /= 1024
        unit_index += 1

    if unit_index == 0:
        return f"{int(size)} {units[unit_index]}"
    return f"{size:.2f} {units[unit_index]}"


def format_number(value: int | float | None) -> str:
    """Format large numbers with commas."""
    if value is None:
        return "N/A"
    if isinstance(value, float):
        return f"{value:,.2f}"
    return f"{value:,}"


def format_rate(value: float | None, unit: str = "/sec") -> str:
    """Format a rate value."""
    if value is None:
        return "N/A"
    return f"{format_number(value)}{unit}"


def print_prometheus_metrics(env: str, metrics: dict[str, Any], console: Console) -> None:
    """Print Prometheus metrics as a formatted table."""
    table = Table(title="Prometheus Metrics", show_header=True, header_style="bold cyan")
    table.add_column("Metric", style="dim")
    table.add_column("Value", justify="right")
    table.add_column("Notes", style="dim")

    # Active series
    active = metrics.get("active_series")
    table.add_row(
        "Active Time Series",
        format_number(active),
        "Primary billing metric"
    )

    # Samples per second / DPM
    sps = metrics.get("samples_per_second")
    dpm = metrics.get("data_points_per_minute")
    table.add_row(
        "Samples/sec",
        format_rate(sps),
        f"DPM: {format_number(dpm)}" if dpm else ""
    )

    # Head chunks
    chunks = metrics.get("head_chunks")
    table.add_row("Head Chunks", format_number(chunks), "Memory indicator")

    # Storage
    storage = metrics.get("storage_bytes")
    table.add_row("TSDB Storage", format_bytes(storage), "On-disk size")

    console.print(table)

    # Top cardinality
    top_metrics = metrics.get("top_metrics_by_series", [])
    if top_metrics:
        card_table = Table(title="Top Metrics by Series Count", show_header=True)
        card_table.add_column("#", style="dim", width=3)
        card_table.add_column("Metric Name")
        card_table.add_column("Series", justify="right")

        for i, entry in enumerate(top_metrics[:10], 1):
            card_table.add_row(
                str(i),
                entry["name"],
                format_number(entry["count"])
            )

        console.print(card_table)

    # Errors
    errors = metrics.get("errors")
    if errors:
        console.print(f"[yellow]Warnings: {', '.join(errors)}[/yellow]")


def print_loki_metrics(env: str, metrics: dict[str, Any], console: Console) -> None:
    """Print Loki metrics as a formatted table."""
    table = Table(title="Loki Metrics", show_header=True, header_style="bold green")
    table.add_column("Metric", style="dim")
    table.add_column("Value", justify="right")
    table.add_column("Notes", style="dim")

    # Ingestion rate
    gb_day = metrics.get("ingestion_rate_gb_per_day")
    bytes_sec = metrics.get("ingestion_rate_bytes_per_second")
    table.add_row(
        "Ingestion Rate",
        f"{gb_day:.2f} GB/day" if gb_day else "N/A",
        f"({format_bytes(bytes_sec)}/sec)" if bytes_sec else ""
    )

    # Total bytes
    total = metrics.get("bytes_received_total")
    table.add_row("Total Bytes Received", format_bytes(total), "Cumulative")

    # Active streams
    streams = metrics.get("active_streams")
    table.add_row("Active Streams", format_number(streams), "")

    # Memory chunks
    chunks = metrics.get("memory_chunks")
    table.add_row("Memory Chunks", format_number(chunks), "")

    # Rejected bytes
    rejected = metrics.get("rejected_bytes_total")
    if rejected and rejected > 0:
        table.add_row(
            "Rejected Bytes",
            format_bytes(rejected),
            "[red]Rate limiting[/red]"
        )

    console.print(table)

    # Errors
    errors = metrics.get("errors")
    if errors:
        console.print(f"[yellow]Warnings: {', '.join(errors)}[/yellow]")


def print_environment_header(env: str, console: Console) -> None:
    """Print an environment header."""
    console.print()
    console.rule(f"[bold]{env.upper()} Environment[/bold]", style="blue")
    console.print()


def print_all_metrics(
    results: dict[str, dict[str, Any]],
    console: Console | None = None
) -> None:
    """
    Print all metrics for all environments.

    Args:
        results: Dict mapping env -> {"prometheus": ..., "loki": ...}
        console: Rich console (created if not provided)
    """
    if console is None:
        console = Console()

    for env, data in results.items():
        print_environment_header(env, console)

        if "prometheus" in data:
            print_prometheus_metrics(env, data["prometheus"], console)
            console.print()

        if "loki" in data:
            print_loki_metrics(env, data["loki"], console)
            console.print()


def format_json(results: dict[str, dict[str, Any]], pretty: bool = True) -> str:
    """Format results as JSON string."""
    # Remove None values for cleaner output
    cleaned = {}
    for env, data in results.items():
        cleaned[env] = {}
        for service, metrics in data.items():
            if metrics:
                # Remove None and empty list values
                cleaned_metrics = {
                    k: v for k, v in metrics.items()
                    if v is not None and v != []
                }
                cleaned[env][service] = cleaned_metrics

    if pretty:
        return json.dumps(cleaned, indent=2)
    return json.dumps(cleaned)


if __name__ == "__main__":
    # Demo the formatters
    console = Console()

    sample_results = {
        "staging": {
            "prometheus": {
                "active_series": 1234567,
                "samples_per_second": 45678.9,
                "data_points_per_minute": 2740734,
                "head_chunks": 2500000,
                "storage_bytes": 13421772800,
                "top_metrics_by_series": [
                    {"name": "http_requests_total", "count": 125000},
                    {"name": "container_cpu_usage", "count": 98000},
                    {"name": "node_memory_bytes", "count": 45000},
                ],
                "errors": None,
            },
            "loki": {
                "bytes_received_total": 1234567890123,
                "ingestion_rate_bytes_per_second": 28571.43,
                "ingestion_rate_gb_per_day": 2.3,
                "active_streams": 5432,
                "memory_chunks": 12000,
                "errors": None,
            },
        }
    }

    print_all_metrics(sample_results, console)
    console.print("\n[dim]JSON output:[/dim]")
    console.print(format_json(sample_results))

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "httpx",
#     "python-dotenv",
# ]
# ///
"""
Grafana API Client for querying metrics through data source proxy.

Supports querying Prometheus and Loki data sources through Grafana's
authenticated proxy API, avoiding the need for direct access to the
underlying data stores.
"""

import os
import sys
import time as time_module
from dataclasses import dataclass
from typing import Any

import httpx
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Environment configurations
ENVIRONMENTS = {
    "staging": {
        "url": "https://g-36d2ad532d.grafana-workspace.us-east-1.amazonaws.com",
        "api_key_env": "GRAFANA_STAGING_API_KEY",
    },
    "prod": {
        "url": "https://g-d7b664d183.grafana-workspace.us-east-1.amazonaws.com",
        "api_key_env": "GRAFANA_PROD_API_KEY",
    },
}


@dataclass
class DataSource:
    """Represents a Grafana data source."""
    id: int
    uid: str
    name: str
    type: str
    url: str


class GrafanaClientError(Exception):
    """Base exception for Grafana client errors."""
    pass


class GrafanaClient:
    """Client for interacting with Grafana HTTP API."""

    def __init__(self, env: str, timeout: float = 30.0):
        """
        Initialize Grafana client for a specific environment.

        Args:
            env: Environment name ('staging' or 'prod')
            timeout: Request timeout in seconds
        """
        if env not in ENVIRONMENTS:
            raise ValueError(f"Unknown environment: {env}. Must be one of {list(ENVIRONMENTS.keys())}")

        config = ENVIRONMENTS[env]
        self.env = env
        self.base_url = config["url"]
        self.timeout = timeout

        api_key = os.getenv(config["api_key_env"])
        if not api_key:
            raise GrafanaClientError(
                f"Missing API key for {env}. Set {config['api_key_env']} environment variable."
            )

        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

        self._datasources_cache: dict[str, DataSource] | None = None

    def _request(self, method: str, path: str, **kwargs) -> dict[str, Any]:
        """Make an HTTP request to Grafana API."""
        url = f"{self.base_url}{path}"

        try:
            with httpx.Client(timeout=self.timeout) as client:
                response = client.request(
                    method,
                    url,
                    headers=self.headers,
                    **kwargs
                )
                response.raise_for_status()
                return response.json()
        except httpx.TimeoutException:
            raise GrafanaClientError(f"Request timed out: {url}")
        except httpx.HTTPStatusError as e:
            raise GrafanaClientError(f"HTTP {e.response.status_code}: {e.response.text}")
        except Exception as e:
            raise GrafanaClientError(f"Request failed: {e}")

    def get_datasources(self) -> list[DataSource]:
        """Get all configured data sources."""
        if self._datasources_cache is not None:
            return list(self._datasources_cache.values())

        data = self._request("GET", "/api/datasources")
        datasources = [
            DataSource(
                id=ds["id"],
                uid=ds["uid"],
                name=ds["name"],
                type=ds["type"],
                url=ds.get("url", ""),
            )
            for ds in data
        ]

        self._datasources_cache = {ds.type: ds for ds in datasources}
        return datasources

    def find_datasource(self, ds_type: str) -> DataSource | None:
        """Find a data source by type (e.g., 'prometheus', 'loki')."""
        if self._datasources_cache is None:
            self.get_datasources()

        return self._datasources_cache.get(ds_type)

    def proxy_query(self, datasource_id: int, query: str, time: float | None = None) -> dict[str, Any]:
        """
        Execute a PromQL query through Grafana's data source proxy.

        Args:
            datasource_id: The data source ID
            query: PromQL query string
            time: Evaluation time as Unix timestamp (default: current time)

        Returns:
            Query result from Prometheus API
        """
        if time is None:
            time = time_module.time()
        path = f"/api/datasources/proxy/{datasource_id}/api/v1/query"
        return self._request("GET", path, params={"query": query, "time": str(time)})

    def proxy_query_range(
        self,
        datasource_id: int,
        query: str,
        start: str,
        end: str,
        step: str = "60s"
    ) -> dict[str, Any]:
        """
        Execute a range query through Grafana's data source proxy.

        Args:
            datasource_id: The data source ID
            query: PromQL query string
            start: Start time (e.g., 'now-1h')
            end: End time (e.g., 'now')
            step: Query resolution step

        Returns:
            Query result from Prometheus API
        """
        path = f"/api/datasources/proxy/{datasource_id}/api/v1/query_range"
        return self._request(
            "GET",
            path,
            params={"query": query, "start": start, "end": end, "step": step}
        )

    def proxy_tsdb_status(self, datasource_id: int, limit: int = 10) -> dict[str, Any]:
        """
        Get TSDB status (cardinality info) through Grafana's data source proxy.

        Args:
            datasource_id: The data source ID
            limit: Number of top series to return

        Returns:
            TSDB status from Prometheus API
        """
        path = f"/api/datasources/proxy/{datasource_id}/api/v1/status/tsdb"
        return self._request("GET", path, params={"limit": limit})

    def proxy_labels(self, datasource_id: int) -> dict[str, Any]:
        """Get all label names from the data source."""
        path = f"/api/datasources/proxy/{datasource_id}/api/v1/labels"
        return self._request("GET", path)


def get_client(env: str) -> GrafanaClient:
    """
    Factory function to create a Grafana client for an environment.

    Args:
        env: Environment name ('staging' or 'prod')

    Returns:
        Configured GrafanaClient instance
    """
    return GrafanaClient(env)


def get_all_clients() -> dict[str, GrafanaClient]:
    """
    Create clients for all available environments.

    Returns:
        Dict mapping environment name to client
    """
    clients = {}
    errors = []

    for env in ENVIRONMENTS:
        try:
            clients[env] = GrafanaClient(env)
        except GrafanaClientError as e:
            errors.append(f"{env}: {e}")

    if errors and not clients:
        raise GrafanaClientError("No environments available:\n" + "\n".join(errors))

    return clients


if __name__ == "__main__":
    # Test the client
    for env in ENVIRONMENTS:
        print(f"\n=== Testing {env} ===")
        try:
            client = GrafanaClient(env)
            datasources = client.get_datasources()
            print(f"Found {len(datasources)} data sources:")
            for ds in datasources:
                print(f"  - {ds.name} ({ds.type})")
        except GrafanaClientError as e:
            print(f"Error: {e}", file=sys.stderr)

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "httpx",
#     "python-dotenv",
# ]
# ///
"""
Loki metrics queries for billing analysis.

Queries key metrics that affect billing:
- Bytes ingested (total and rate)
- Active streams
- Memory usage (chunks)
"""

import sys
from dataclasses import dataclass, field
from typing import Any

# Import from sibling module
from grafana_client import GrafanaClient, GrafanaClientError


@dataclass
class LokiMetrics:
    """Container for Loki billing metrics."""
    bytes_received_total: int | None = None
    ingestion_rate_bytes_per_second: float | None = None
    active_streams: int | None = None
    memory_chunks: int | None = None
    rejected_bytes_total: int | None = None
    errors: list[str] = field(default_factory=list)

    @property
    def ingestion_rate_gb_per_day(self) -> float | None:
        """Calculate GB/day ingestion rate."""
        if self.ingestion_rate_bytes_per_second is None:
            return None
        # bytes/sec * 60 * 60 * 24 / (1024^3)
        return self.ingestion_rate_bytes_per_second * 86400 / (1024 ** 3)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        return {
            "bytes_received_total": self.bytes_received_total,
            "ingestion_rate_bytes_per_second": self.ingestion_rate_bytes_per_second,
            "ingestion_rate_gb_per_day": (
                round(self.ingestion_rate_gb_per_day, 2)
                if self.ingestion_rate_gb_per_day else None
            ),
            "active_streams": self.active_streams,
            "memory_chunks": self.memory_chunks,
            "rejected_bytes_total": self.rejected_bytes_total,
            "errors": self.errors if self.errors else None,
        }


def _extract_scalar(result: dict[str, Any]) -> float | None:
    """Extract a scalar value from a Prometheus query result."""
    try:
        data = result.get("data", {})
        if data.get("resultType") == "vector" and data.get("result"):
            # Vector result - take first value
            value = data["result"][0]["value"][1]
            return float(value)
        elif data.get("resultType") == "scalar":
            return float(data["result"][1])
    except (IndexError, KeyError, ValueError, TypeError):
        pass
    return None


def _sum_all_values(result: dict[str, Any]) -> float | None:
    """Sum all values from a vector result."""
    try:
        data = result.get("data", {})
        if data.get("resultType") == "vector" and data.get("result"):
            total = 0.0
            for entry in data["result"]:
                total += float(entry["value"][1])
            return total
    except (IndexError, KeyError, ValueError, TypeError):
        pass
    return None


def query_loki_metrics(client: GrafanaClient) -> LokiMetrics:
    """
    Query all Loki billing metrics.

    Note: Loki metrics are typically exposed through a Prometheus data source
    that scrapes Loki's /metrics endpoint. We query these through the
    Prometheus proxy.

    Args:
        client: Grafana client instance

    Returns:
        LokiMetrics with all available data
    """
    metrics = LokiMetrics()

    # Find Prometheus data source (Loki metrics are scraped by Prometheus)
    prom_ds = client.find_datasource("prometheus")
    if not prom_ds:
        metrics.errors.append("No Prometheus data source found (needed for Loki metrics)")
        return metrics

    ds_id = prom_ds.id

    # Query total bytes received
    try:
        result = client.proxy_query(
            ds_id,
            "sum(loki_distributor_bytes_received_total)"
        )
        value = _sum_all_values(result) or _extract_scalar(result)
        metrics.bytes_received_total = int(value) if value else None
    except GrafanaClientError as e:
        metrics.errors.append(f"bytes_received_total: {e}")

    # Query ingestion rate (bytes/sec over 5m)
    try:
        result = client.proxy_query(
            ds_id,
            "sum(rate(loki_distributor_bytes_received_total[5m]))"
        )
        metrics.ingestion_rate_bytes_per_second = _extract_scalar(result)
    except GrafanaClientError as e:
        metrics.errors.append(f"ingestion_rate: {e}")

    # Query active streams
    try:
        result = client.proxy_query(
            ds_id,
            "sum(loki_ingester_memory_streams)"
        )
        value = _extract_scalar(result)
        metrics.active_streams = int(value) if value else None
    except GrafanaClientError as e:
        metrics.errors.append(f"active_streams: {e}")

    # Query memory chunks
    try:
        result = client.proxy_query(
            ds_id,
            "sum(loki_ingester_memory_chunks)"
        )
        value = _extract_scalar(result)
        metrics.memory_chunks = int(value) if value else None
    except GrafanaClientError as e:
        metrics.errors.append(f"memory_chunks: {e}")

    # Query rejected bytes (rate limiting hits)
    try:
        result = client.proxy_query(
            ds_id,
            'sum(loki_distributor_bytes_received_total{reason=~".+"})'
        )
        value = _sum_all_values(result) or _extract_scalar(result)
        metrics.rejected_bytes_total = int(value) if value else None
    except GrafanaClientError:
        # This is expected to fail if there are no rejected bytes
        pass

    return metrics


if __name__ == "__main__":
    # Test the module
    import json
    from grafana_client import get_all_clients

    clients = get_all_clients()

    for env, client in clients.items():
        print(f"\n=== {env.upper()} Loki Metrics ===")
        try:
            metrics = query_loki_metrics(client)
            print(json.dumps(metrics.to_dict(), indent=2))
        except Exception as e:
            print(f"Error: {e}", file=sys.stderr)

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "httpx",
#     "python-dotenv",
# ]
# ///
"""
Prometheus metrics queries for billing analysis.

Queries key metrics that affect billing:
- Active time series
- Samples per second (DPM)
- Storage size
- Cardinality breakdown
"""

import sys
from dataclasses import dataclass, field
from typing import Any

# Import from sibling module
from grafana_client import GrafanaClient, GrafanaClientError


@dataclass
class CardinalityEntry:
    """A single cardinality entry (metric or label with series count)."""
    name: str
    count: int


@dataclass
class PrometheusMetrics:
    """Container for Prometheus billing metrics."""
    active_series: int | None = None
    samples_per_second: float | None = None
    head_chunks: int | None = None
    storage_bytes: int | None = None
    top_metrics_by_series: list[CardinalityEntry] = field(default_factory=list)
    top_labels_by_cardinality: list[CardinalityEntry] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        return {
            "active_series": self.active_series,
            "samples_per_second": self.samples_per_second,
            "data_points_per_minute": (
                round(self.samples_per_second * 60) if self.samples_per_second else None
            ),
            "head_chunks": self.head_chunks,
            "storage_bytes": self.storage_bytes,
            "top_metrics_by_series": [
                {"name": e.name, "count": e.count} for e in self.top_metrics_by_series
            ],
            "top_labels_by_cardinality": [
                {"name": e.name, "count": e.count} for e in self.top_labels_by_cardinality
            ],
            "errors": self.errors if self.errors else None,
        }


def _extract_scalar(result: dict[str, Any]) -> float | None:
    """Extract a scalar value from a Prometheus query result."""
    try:
        data = result.get("data", {})
        if data.get("resultType") == "vector" and data.get("result"):
            # Vector result - take first value
            value = data["result"][0]["value"][1]
            return float(value)
        elif data.get("resultType") == "scalar":
            return float(data["result"][1])
    except (IndexError, KeyError, ValueError, TypeError):
        pass
    return None


def query_prometheus_metrics(client: GrafanaClient, limit: int = 10) -> PrometheusMetrics:
    """
    Query all Prometheus billing metrics.

    Args:
        client: Grafana client instance
        limit: Number of top cardinality entries to return

    Returns:
        PrometheusMetrics with all available data
    """
    metrics = PrometheusMetrics()

    # Find Prometheus data source
    prom_ds = client.find_datasource("prometheus")
    if not prom_ds:
        metrics.errors.append("No Prometheus data source found")
        return metrics

    ds_id = prom_ds.id

    # Query TSDB status for cardinality FIRST (most reliable)
    # AWS Managed Prometheus/Cortex/Mimir all support this endpoint
    try:
        status = client.proxy_tsdb_status(ds_id, limit=limit)
        data = status.get("data", {})

        # Top metrics by series count
        series_by_metric = data.get("seriesCountByMetricName", [])
        metrics.top_metrics_by_series = [
            CardinalityEntry(name=entry["name"], count=entry["value"])
            for entry in series_by_metric[:limit]
        ]

        # Calculate total active series from TSDB status
        # This works for AWS Managed Prometheus where prometheus_tsdb_head_series doesn't exist
        total_series = sum(entry["value"] for entry in series_by_metric)
        if total_series > 0:
            metrics.active_series = total_series

        # Top labels by cardinality
        labels_by_count = data.get("labelValueCountByLabelName", [])
        metrics.top_labels_by_cardinality = [
            CardinalityEntry(name=entry["name"], count=entry["value"])
            for entry in labels_by_count[:limit]
        ]

        # headStats contains series count in some Prometheus versions
        head_stats = data.get("headStats", {})
        if head_stats.get("numSeries") and not metrics.active_series:
            metrics.active_series = head_stats["numSeries"]
        if head_stats.get("numChunks"):
            metrics.head_chunks = head_stats["numChunks"]

    except GrafanaClientError as e:
        metrics.errors.append(f"tsdb_status: {e}")

    # Try Cortex/Mimir active series metric (AWS Managed Prometheus)
    if not metrics.active_series:
        try:
            result = client.proxy_query(ds_id, "sum(cortex_ingester_active_series)")
            value = _extract_scalar(result)
            if value and value > 0:
                metrics.active_series = int(value)
        except GrafanaClientError:
            pass  # Try next metric

    # Fallback to standard Prometheus metric
    if not metrics.active_series:
        try:
            result = client.proxy_query(ds_id, "prometheus_tsdb_head_series")
            value = _extract_scalar(result)
            if value and value > 0:
                metrics.active_series = int(value)
        except GrafanaClientError:
            pass

    # Query samples per second - try multiple metrics
    sample_rate_queries = [
        "sum(rate(cortex_ingester_ingested_samples_total[5m]))",  # Cortex/Mimir
        "sum(rate(prometheus_tsdb_head_samples_appended_total[5m]))",  # Standard Prometheus
    ]
    for query in sample_rate_queries:
        if metrics.samples_per_second:
            break
        try:
            result = client.proxy_query(ds_id, query)
            value = _extract_scalar(result)
            if value and value > 0:
                metrics.samples_per_second = value
        except GrafanaClientError:
            pass

    # Query storage size - try multiple metrics
    storage_queries = [
        "sum(cortex_ingester_tsdb_storage_blocks_bytes)",  # Cortex/Mimir
        "sum(prometheus_tsdb_storage_blocks_bytes)",  # Standard Prometheus
    ]
    for query in storage_queries:
        if metrics.storage_bytes:
            break
        try:
            result = client.proxy_query(ds_id, query)
            value = _extract_scalar(result)
            if value and value > 0:
                metrics.storage_bytes = int(value)
        except GrafanaClientError:
            pass

    return metrics


if __name__ == "__main__":
    # Test the module
    import json
    from grafana_client import get_all_clients

    clients = get_all_clients()

    for env, client in clients.items():
        print(f"\n=== {env.upper()} Prometheus Metrics ===")
        try:
            metrics = query_prometheus_metrics(client)
            print(json.dumps(metrics.to_dict(), indent=2))
        except Exception as e:
            print(f"Error: {e}", file=sys.stderr)