Prometheus

#!/usr/bin/env python3
"""
Prometheus Health Check Tool

Check Prometheus server health, targets status, and basic statistics.

Usage:
    python prom_health.py <prometheus_url> [options]

Examples:
    python prom_health.py http://localhost:9090
    python prom_health.py http://localhost:9090 --targets
    python prom_health.py http://localhost:9090 --rules
    python prom_health.py http://localhost:9090 --all
"""

import argparse
import json
import sys
import urllib.request
from typing import Any


def fetch_endpoint(base_url: str, endpoint: str) -> dict[str, Any]:
    """Fetch data from Prometheus endpoint."""
    url = f"{base_url.rstrip('/')}{endpoint}"
    try:
        with urllib.request.urlopen(url, timeout=10) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        return {"status": "error", "error": f"HTTP {e.code}", "errorType": "http_error"}
    except urllib.error.URLError as e:
        return {"status": "error", "error": str(e.reason), "errorType": "connection_error"}
    except Exception as e:
        return {"status": "error", "error": str(e), "errorType": "unknown"}


def check_ready(base_url: str) -> bool:
    """Check if Prometheus is ready."""
    url = f"{base_url.rstrip('/')}/-/ready"
    try:
        with urllib.request.urlopen(url, timeout=5) as response:
            return response.status == 200
    except Exception:
        return False


def check_healthy(base_url: str) -> bool:
    """Check if Prometheus is healthy."""
    url = f"{base_url.rstrip('/')}/-/healthy"
    try:
        with urllib.request.urlopen(url, timeout=5) as response:
            return response.status == 200
    except Exception:
        return False


def print_section(title: str) -> None:
    """Print section header."""
    print(f"\n{'='*60}")
    print(f" {title}")
    print(f"{'='*60}")


def print_status(label: str, status: bool, details: str = "") -> None:
    """Print status line with icon."""
    icon = "✓" if status else "✗"
    status_text = "OK" if status else "FAIL"
    detail_str = f" ({details})" if details else ""
    print(f"  [{icon}] {label}: {status_text}{detail_str}")


def check_build_info(base_url: str) -> dict[str, Any] | None:
    """Get build info."""
    result = fetch_endpoint(base_url, "/api/v1/status/buildinfo")
    if result.get("status") == "success":
        return result.get("data", {})
    return None


def check_runtime_info(base_url: str) -> dict[str, Any] | None:
    """Get runtime info."""
    result = fetch_endpoint(base_url, "/api/v1/status/runtimeinfo")
    if result.get("status") == "success":
        return result.get("data", {})
    return None


def check_targets(base_url: str) -> None:
    """Check and display target status."""
    result = fetch_endpoint(base_url, "/api/v1/targets")

    if result.get("status") != "success":
        print(f"  Error fetching targets: {result.get('error', 'Unknown')}")
        return

    data = result.get("data", {})
    active = data.get("activeTargets", [])
    dropped = data.get("droppedTargets", [])

    # Count by health status
    health_counts = {"up": 0, "down": 0, "unknown": 0}
    for target in active:
        health = target.get("health", "unknown")
        health_counts[health] = health_counts.get(health, 0) + 1

    print(f"  Active targets: {len(active)}")
    print(f"    - Up: {health_counts.get('up', 0)}")
    print(f"    - Down: {health_counts.get('down', 0)}")
    print(f"    - Unknown: {health_counts.get('unknown', 0)}")
    print(f"  Dropped targets: {len(dropped)}")

    # List down targets
    down_targets = [t for t in active if t.get("health") == "down"]
    if down_targets:
        print("\n  Down targets:")
        for target in down_targets[:10]:  # Limit to 10
            labels = target.get("labels", {})
            job = labels.get("job", "unknown")
            instance = labels.get("instance", "unknown")
            error = target.get("lastError", "No error info")
            print(f"    - {job}/{instance}: {error[:60]}")
        if len(down_targets) > 10:
            print(f"    ... and {len(down_targets) - 10} more")


def check_rules(base_url: str) -> None:
    """Check and display rules status."""
    result = fetch_endpoint(base_url, "/api/v1/rules")

    if result.get("status") != "success":
        print(f"  Error fetching rules: {result.get('error', 'Unknown')}")
        return

    groups = result.get("data", {}).get("groups", [])

    total_rules = 0
    alerting_rules = 0
    recording_rules = 0
    firing_alerts = 0
    unhealthy_rules = 0

    for group in groups:
        for rule in group.get("rules", []):
            total_rules += 1
            rule_type = rule.get("type")

            if rule_type == "alerting":
                alerting_rules += 1
                alerts = rule.get("alerts", [])
                firing = [a for a in alerts if a.get("state") == "firing"]
                firing_alerts += len(firing)
            elif rule_type == "recording":
                recording_rules += 1

            if rule.get("health") != "ok":
                unhealthy_rules += 1

    print(f"  Rule groups: {len(groups)}")
    print(f"  Total rules: {total_rules}")
    print(f"    - Alerting: {alerting_rules}")
    print(f"    - Recording: {recording_rules}")
    print(f"  Firing alerts: {firing_alerts}")
    print(f"  Unhealthy rules: {unhealthy_rules}")


def check_tsdb(base_url: str) -> None:
    """Check TSDB statistics."""
    result = fetch_endpoint(base_url, "/api/v1/status/tsdb")

    if result.get("status") != "success":
        print(f"  Error fetching TSDB stats: {result.get('error', 'Unknown')}")
        return

    data = result.get("data", {})
    head_stats = data.get("headStats", {})

    print(f"  Series count: {head_stats.get('numSeries', 'N/A')}")
    print(f"  Chunk count: {head_stats.get('chunkCount', 'N/A')}")

    # Top metrics by series count
    series_by_metric = data.get("seriesCountByMetricName", [])
    if series_by_metric:
        print("\n  Top 5 metrics by series count:")
        for item in series_by_metric[:5]:
            print(f"    - {item.get('name', 'unknown')}: {item.get('value', 0)}")


def check_alerts(base_url: str) -> None:
    """Check active alerts."""
    result = fetch_endpoint(base_url, "/api/v1/alerts")

    if result.get("status") != "success":
        print(f"  Error fetching alerts: {result.get('error', 'Unknown')}")
        return

    alerts = result.get("data", {}).get("alerts", [])

    # Group by state
    by_state = {}
    for alert in alerts:
        state = alert.get("state", "unknown")
        by_state[state] = by_state.get(state, 0) + 1

    print(f"  Total active alerts: {len(alerts)}")
    for state, count in sorted(by_state.items()):
        print(f"    - {state}: {count}")

    # List firing alerts
    firing = [a for a in alerts if a.get("state") == "firing"]
    if firing:
        print("\n  Firing alerts:")
        for alert in firing[:10]:
            name = alert.get("labels", {}).get("alertname", "unknown")
            severity = alert.get("labels", {}).get("severity", "unknown")
            print(f"    - {name} (severity: {severity})")
        if len(firing) > 10:
            print(f"    ... and {len(firing) - 10} more")


def main():
    parser = argparse.ArgumentParser(
        description="Check Prometheus server health and status",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument("url", help="Prometheus server URL")
    parser.add_argument("--targets", "-t", action="store_true", help="Show detailed target info")
    parser.add_argument("--rules", "-r", action="store_true", help="Show detailed rules info")
    parser.add_argument("--tsdb", "-d", action="store_true", help="Show TSDB statistics")
    parser.add_argument("--alerts", "-a", action="store_true", help="Show active alerts")
    parser.add_argument("--all", action="store_true", help="Show all information")

    args = parser.parse_args()

    if args.all:
        args.targets = args.rules = args.tsdb = args.alerts = True

    base_url = args.url.rstrip("/")

    # Basic health checks
    print_section("Health Checks")

    ready = check_ready(base_url)
    healthy = check_healthy(base_url)

    print_status("Ready", ready)
    print_status("Healthy", healthy)

    if not (ready and healthy):
        print("\n  ⚠ Prometheus is not fully operational")
        sys.exit(1)

    # Build info
    build_info = check_build_info(base_url)
    if build_info:
        print_section("Build Information")
        print(f"  Version: {build_info.get('version', 'N/A')}")
        print(f"  Revision: {build_info.get('revision', 'N/A')[:12]}")
        print(f"  Go version: {build_info.get('goVersion', 'N/A')}")

    # Runtime info
    runtime_info = check_runtime_info(base_url)
    if runtime_info:
        print_section("Runtime Information")
        print(f"  Start time: {runtime_info.get('startTime', 'N/A')}")
        print(f"  Time series: {runtime_info.get('timeSeriesCount', 'N/A')}")
        print(f"  Goroutines: {runtime_info.get('goroutineCount', 'N/A')}")
        print(f"  Storage retention: {runtime_info.get('storageRetention', 'N/A')}")

        reload_success = runtime_info.get("reloadConfigSuccess")
        if reload_success is not None:
            print_status("Config reload", reload_success)

    # Optional detailed sections
    if args.targets:
        print_section("Targets")
        check_targets(base_url)

    if args.rules:
        print_section("Rules")
        check_rules(base_url)

    if args.alerts:
        print_section("Alerts")
        check_alerts(base_url)

    if args.tsdb:
        print_section("TSDB Statistics")
        check_tsdb(base_url)

    print()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Prometheus Metadata Tool

Query metadata from Prometheus: series, labels, label values, and metric info.

Usage:
    python prom_metadata.py <prometheus_url> <command> [options]

Commands:
    series      Find series matching label selectors
    labels      List all label names
    values      Get values for a specific label
    metadata    Get metric metadata
    targets     List scrape targets

Examples:
    python prom_metadata.py http://localhost:9090 series 'up' 'process_start_time_seconds{job="prometheus"}'
    python prom_metadata.py http://localhost:9090 labels
    python prom_metadata.py http://localhost:9090 values job
    python prom_metadata.py http://localhost:9090 metadata --metric http_requests_total
    python prom_metadata.py http://localhost:9090 targets --state active
"""

import argparse
import json
import sys
import urllib.parse
import urllib.request
from typing import Any


def fetch_api(base_url: str, endpoint: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
    """Fetch from Prometheus API."""
    url = f"{base_url.rstrip('/')}{endpoint}"
    if params:
        # Handle repeated parameters (like match[])
        param_parts = []
        for key, value in params.items():
            if isinstance(value, list):
                for v in value:
                    param_parts.append(f"{urllib.parse.quote(key)}={urllib.parse.quote(str(v))}")
            elif value is not None:
                param_parts.append(f"{urllib.parse.quote(key)}={urllib.parse.quote(str(value))}")
        if param_parts:
            url += "?" + "&".join(param_parts)

    try:
        with urllib.request.urlopen(url, timeout=30) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        body = e.read().decode()
        try:
            return json.loads(body)
        except json.JSONDecodeError:
            return {"status": "error", "error": body, "errorType": str(e.code)}
    except urllib.error.URLError as e:
        return {"status": "error", "error": str(e.reason), "errorType": "connection_error"}


def cmd_series(base_url: str, matchers: list[str], start: str | None, end: str | None, limit: int | None) -> int:
    """Find series by label matchers."""
    params = {"match[]": matchers}
    if start:
        params["start"] = start
    if end:
        params["end"] = end
    if limit:
        params["limit"] = limit

    result = fetch_api(base_url, "/api/v1/series", params)

    if result.get("status") != "success":
        print(f"Error: {result.get('error', 'Unknown error')}", file=sys.stderr)
        return 1

    series = result.get("data", [])
    print(f"Found {len(series)} series:\n")

    for s in series:
        name = s.get("__name__", "")
        labels = {k: v for k, v in s.items() if k != "__name__"}
        label_str = ", ".join(f'{k}="{v}"' for k, v in sorted(labels.items()))
        if name:
            print(f"{name}{{{label_str}}}" if labels else name)
        else:
            print(f"{{{label_str}}}")

    return 0


def cmd_labels(base_url: str, matchers: list[str] | None, start: str | None, end: str | None, limit: int | None) -> int:
    """List all label names."""
    params = {}
    if matchers:
        params["match[]"] = matchers
    if start:
        params["start"] = start
    if end:
        params["end"] = end
    if limit:
        params["limit"] = limit

    result = fetch_api(base_url, "/api/v1/labels", params if params else None)

    if result.get("status") != "success":
        print(f"Error: {result.get('error', 'Unknown error')}", file=sys.stderr)
        return 1

    labels = result.get("data", [])
    print(f"Found {len(labels)} labels:\n")
    for label in sorted(labels):
        print(f"  {label}")

    return 0


def cmd_values(base_url: str, label_name: str, matchers: list[str] | None, start: str | None, end: str | None, limit: int | None) -> int:
    """Get values for a label."""
    params = {}
    if matchers:
        params["match[]"] = matchers
    if start:
        params["start"] = start
    if end:
        params["end"] = end
    if limit:
        params["limit"] = limit

    result = fetch_api(base_url, f"/api/v1/label/{urllib.parse.quote(label_name)}/values", params if params else None)

    if result.get("status") != "success":
        print(f"Error: {result.get('error', 'Unknown error')}", file=sys.stderr)
        return 1

    values = result.get("data", [])
    print(f"Found {len(values)} values for label '{label_name}':\n")
    for value in sorted(values):
        print(f"  {value}")

    return 0


def cmd_metadata(base_url: str, metric: str | None, limit: int | None, limit_per_metric: int | None) -> int:
    """Get metric metadata."""
    params = {}
    if metric:
        params["metric"] = metric
    if limit:
        params["limit"] = limit
    if limit_per_metric:
        params["limit_per_metric"] = limit_per_metric

    result = fetch_api(base_url, "/api/v1/metadata", params if params else None)

    if result.get("status") != "success":
        print(f"Error: {result.get('error', 'Unknown error')}", file=sys.stderr)
        return 1

    metadata = result.get("data", {})
    print(f"Found metadata for {len(metadata)} metrics:\n")

    for metric_name, info_list in sorted(metadata.items()):
        print(f"{metric_name}:")
        for info in info_list:
            print(f"  Type: {info.get('type', 'unknown')}")
            print(f"  Help: {info.get('help', 'N/A')}")
            if info.get("unit"):
                print(f"  Unit: {info.get('unit')}")
        print()

    return 0


def cmd_targets(base_url: str, state: str | None, scrape_pool: str | None) -> int:
    """List scrape targets."""
    params = {}
    if state:
        params["state"] = state
    if scrape_pool:
        params["scrapePool"] = scrape_pool

    result = fetch_api(base_url, "/api/v1/targets", params if params else None)

    if result.get("status") != "success":
        print(f"Error: {result.get('error', 'Unknown error')}", file=sys.stderr)
        return 1

    data = result.get("data", {})
    active = data.get("activeTargets", [])
    dropped = data.get("droppedTargets", [])

    if active:
        print(f"Active targets ({len(active)}):\n")
        for target in active:
            labels = target.get("labels", {})
            job = labels.get("job", "unknown")
            instance = labels.get("instance", "unknown")
            health = target.get("health", "unknown")
            health_icon = "✓" if health == "up" else "✗" if health == "down" else "?"

            print(f"  [{health_icon}] {job}/{instance}")
            print(f"      URL: {target.get('scrapeUrl', 'N/A')}")
            print(f"      Pool: {target.get('scrapePool', 'N/A')}")
            print(f"      Last scrape: {target.get('lastScrape', 'N/A')}")
            if target.get("lastError"):
                print(f"      Error: {target.get('lastError')}")
            print()

    if dropped and (state is None or state in ("dropped", "any")):
        print(f"\nDropped targets ({len(dropped)}):\n")
        for target in dropped[:20]:  # Limit output
            discovered = target.get("discoveredLabels", {})
            job = discovered.get("job", "unknown")
            address = discovered.get("__address__", "unknown")
            print(f"  - {job}: {address}")
        if len(dropped) > 20:
            print(f"  ... and {len(dropped) - 20} more")

    return 0


def main():
    parser = argparse.ArgumentParser(
        description="Query Prometheus metadata",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )

    parser.add_argument("url", help="Prometheus server URL")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # Series command
    series_parser = subparsers.add_parser("series", help="Find series by label matchers")
    series_parser.add_argument("matchers", nargs="+", help="Series selectors (e.g., 'up', 'http_requests_total{job=\"api\"}')")
    series_parser.add_argument("--start", help="Start timestamp")
    series_parser.add_argument("--end", help="End timestamp")
    series_parser.add_argument("--limit", type=int, help="Max series to return")

    # Labels command
    labels_parser = subparsers.add_parser("labels", help="List all label names")
    labels_parser.add_argument("--match", action="append", dest="matchers", help="Filter by series selector")
    labels_parser.add_argument("--start", help="Start timestamp")
    labels_parser.add_argument("--end", help="End timestamp")
    labels_parser.add_argument("--limit", type=int, help="Max labels to return")

    # Values command
    values_parser = subparsers.add_parser("values", help="Get values for a label")
    values_parser.add_argument("label_name", help="Label name")
    values_parser.add_argument("--match", action="append", dest="matchers", help="Filter by series selector")
    values_parser.add_argument("--start", help="Start timestamp")
    values_parser.add_argument("--end", help="End timestamp")
    values_parser.add_argument("--limit", type=int, help="Max values to return")

    # Metadata command
    metadata_parser = subparsers.add_parser("metadata", help="Get metric metadata")
    metadata_parser.add_argument("--metric", "-m", help="Filter by metric name")
    metadata_parser.add_argument("--limit", type=int, help="Max metrics to return")
    metadata_parser.add_argument("--limit-per-metric", type=int, help="Max metadata per metric")

    # Targets command
    targets_parser = subparsers.add_parser("targets", help="List scrape targets")
    targets_parser.add_argument("--state", choices=["active", "dropped", "any"], help="Filter by state")
    targets_parser.add_argument("--scrape-pool", help="Filter by scrape pool name")

    args = parser.parse_args()

    if args.command == "series":
        return cmd_series(args.url, args.matchers, args.start, args.end, args.limit)
    elif args.command == "labels":
        return cmd_labels(args.url, args.matchers, args.start, args.end, args.limit)
    elif args.command == "values":
        return cmd_values(args.url, args.label_name, args.matchers, args.start, args.end, args.limit)
    elif args.command == "metadata":
        return cmd_metadata(args.url, args.metric, args.limit, getattr(args, "limit_per_metric", None))
    elif args.command == "targets":
        return cmd_targets(args.url, args.state, getattr(args, "scrape_pool", None))


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

#!/usr/bin/env python3
"""
Prometheus Query Tool

Execute instant and range queries against Prometheus HTTP API.

Usage:
    python prom_query.py <prometheus_url> <query> [options]

Examples:
    # Instant query
    python prom_query.py http://localhost:9090 'up'

    # Range query
    python prom_query.py http://localhost:9090 'rate(http_requests_total[5m])' \
        --start '2024-01-01T00:00:00Z' --end '2024-01-01T01:00:00Z' --step '1m'

    # Output formats
    python prom_query.py http://localhost:9090 'up' --format json
    python prom_query.py http://localhost:9090 'up' --format csv
    python prom_query.py http://localhost:9090 'up' --format table
"""

import argparse
import json
import sys
import urllib.parse
import urllib.request
from datetime import datetime
from typing import Any


def query_prometheus(
    base_url: str,
    query: str,
    time: str | None = None,
    start: str | None = None,
    end: str | None = None,
    step: str | None = None,
    timeout: str | None = None,
    limit: int | None = None,
) -> dict[str, Any]:
    """Execute a Prometheus query."""

    # Determine query type
    is_range = start is not None and end is not None
    endpoint = "/api/v1/query_range" if is_range else "/api/v1/query"

    # Build parameters
    params = {"query": query}

    if is_range:
        params["start"] = start
        params["end"] = end
        params["step"] = step or "1m"
    elif time:
        params["time"] = time

    if timeout:
        params["timeout"] = timeout
    if limit:
        params["limit"] = str(limit)

    # Build URL
    url = f"{base_url.rstrip('/')}{endpoint}?{urllib.parse.urlencode(params)}"

    # Execute request
    try:
        with urllib.request.urlopen(url, timeout=30) as response:
            data = json.loads(response.read().decode())
            return data
    except urllib.error.HTTPError as e:
        error_body = e.read().decode()
        try:
            error_data = json.loads(error_body)
            return error_data
        except json.JSONDecodeError:
            return {"status": "error", "error": error_body, "errorType": str(e.code)}
    except urllib.error.URLError as e:
        return {"status": "error", "error": str(e.reason), "errorType": "connection_error"}


def format_timestamp(ts: float) -> str:
    """Format Unix timestamp as ISO 8601."""
    return datetime.utcfromtimestamp(ts).strftime("%Y-%m-%dT%H:%M:%SZ")


def format_metric(metric: dict[str, str]) -> str:
    """Format metric labels as string."""
    name = metric.get("__name__", "")
    labels = {k: v for k, v in metric.items() if k != "__name__"}

    if not labels:
        return name or "{}"

    label_str = ", ".join(f'{k}="{v}"' for k, v in sorted(labels.items()))
    return f"{name}{{{label_str}}}" if name else f"{{{label_str}}}"


def output_json(data: dict[str, Any]) -> None:
    """Output as JSON."""
    print(json.dumps(data, indent=2))


def output_table(data: dict[str, Any]) -> None:
    """Output as formatted table."""
    if data.get("status") != "success":
        print(f"Error: {data.get('error', 'Unknown error')}")
        return

    result_type = data.get("data", {}).get("resultType")
    results = data.get("data", {}).get("result", [])

    if not results:
        print("No results")
        return

    if result_type == "vector":
        # Instant vector
        print(f"{'METRIC':<60} {'TIMESTAMP':<25} {'VALUE':>15}")
        print("-" * 100)
        for item in results:
            metric = format_metric(item.get("metric", {}))
            ts, value = item.get("value", [0, ""])
            print(f"{metric:<60} {format_timestamp(ts):<25} {value:>15}")

    elif result_type == "matrix":
        # Range vector
        for item in results:
            metric = format_metric(item.get("metric", {}))
            print(f"\n{metric}")
            print(f"{'TIMESTAMP':<25} {'VALUE':>15}")
            print("-" * 40)
            for ts, value in item.get("values", []):
                print(f"{format_timestamp(ts):<25} {value:>15}")

    elif result_type == "scalar":
        ts, value = data.get("data", {}).get("result", [0, ""])
        print(f"Scalar: {value} at {format_timestamp(ts)}")

    elif result_type == "string":
        ts, value = data.get("data", {}).get("result", [0, ""])
        print(f"String: {value} at {format_timestamp(ts)}")

    # Print warnings if any
    warnings = data.get("warnings", [])
    if warnings:
        print("\nWarnings:")
        for w in warnings:
            print(f"  - {w}")


def output_csv(data: dict[str, Any]) -> None:
    """Output as CSV."""
    if data.get("status") != "success":
        print(f"error,{data.get('error', 'Unknown error')}")
        return

    result_type = data.get("data", {}).get("resultType")
    results = data.get("data", {}).get("result", [])

    if result_type == "vector":
        print("metric,timestamp,value")
        for item in results:
            metric = format_metric(item.get("metric", {})).replace(",", ";")
            ts, value = item.get("value", [0, ""])
            print(f'"{metric}",{format_timestamp(ts)},{value}')

    elif result_type == "matrix":
        print("metric,timestamp,value")
        for item in results:
            metric = format_metric(item.get("metric", {})).replace(",", ";")
            for ts, value in item.get("values", []):
                print(f'"{metric}",{format_timestamp(ts)},{value}')

    elif result_type in ("scalar", "string"):
        print("timestamp,value")
        ts, value = data.get("data", {}).get("result", [0, ""])
        print(f"{format_timestamp(ts)},{value}")


def main():
    parser = argparse.ArgumentParser(
        description="Query Prometheus HTTP API",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  Instant query:
    %(prog)s http://localhost:9090 'up'
    %(prog)s http://localhost:9090 'rate(http_requests_total[5m])' --time '2024-01-01T00:00:00Z'

  Range query:
    %(prog)s http://localhost:9090 'up' --start '2024-01-01T00:00:00Z' --end '2024-01-01T01:00:00Z' --step '1m'

  Output formats:
    %(prog)s http://localhost:9090 'up' --format table
    %(prog)s http://localhost:9090 'up' --format json
    %(prog)s http://localhost:9090 'up' --format csv
        """
    )

    parser.add_argument("url", help="Prometheus server URL (e.g., http://localhost:9090)")
    parser.add_argument("query", help="PromQL query expression")
    parser.add_argument("--time", "-t", help="Evaluation timestamp (RFC3339 or Unix)")
    parser.add_argument("--start", "-s", help="Range query start time")
    parser.add_argument("--end", "-e", help="Range query end time")
    parser.add_argument("--step", help="Range query step (default: 1m)")
    parser.add_argument("--timeout", help="Query timeout duration")
    parser.add_argument("--limit", type=int, help="Max series to return")
    parser.add_argument(
        "--format", "-f",
        choices=["json", "table", "csv"],
        default="table",
        help="Output format (default: table)"
    )

    args = parser.parse_args()

    # Validate range query args
    if (args.start is None) != (args.end is None):
        parser.error("--start and --end must be used together for range queries")

    # Execute query
    result = query_prometheus(
        base_url=args.url,
        query=args.query,
        time=args.time,
        start=args.start,
        end=args.end,
        step=args.step,
        timeout=args.timeout,
        limit=args.limit,
    )

    # Output result
    if args.format == "json":
        output_json(result)
    elif args.format == "csv":
        output_csv(result)
    else:
        output_table(result)

    # Exit with error code if query failed
    if result.get("status") != "success":
        sys.exit(1)


if __name__ == "__main__":
    main()