feat(infrahubctl): add telemetry commands

Signed-off-by: Fatih Acar <fatih@opsmill.com>

Author: fatih-acar

docs/docs/infrahubctl/infrahubctl-telemetry.mdx

@@ -0,0 +1,57 @@
+# `infrahubctl telemetry`
+
+**Usage**:
+
+```console
+$ infrahubctl telemetry [OPTIONS] COMMAND [ARGS]...
+```
+
+**Options**:
+
+* `--install-completion`: Install completion for the current shell.
+* `--show-completion`: Show completion for the current shell, to copy it or customize the installation.
+* `--help`: Show this message and exit.
+
+**Commands**:
+
+* `export`: Export telemetry snapshots to a JSON file.
+* `list`: List telemetry snapshots with summary...
+
+## `infrahubctl telemetry export`
+
+Export telemetry snapshots to a JSON file.
+
+Pages through the API automatically so that all matching snapshots are exported,
+not just the first page.
+
+**Usage**:
+
+```console
+$ infrahubctl telemetry export [OPTIONS]
+```
+
+**Options**:
+
+* `--output TEXT`: Output file path  [default: telemetry-export.json]
+* `--start-date [%Y-%m-%d|%Y-%m-%dT%H:%M:%S|%Y-%m-%dT%H:%M:%S%z]`: Start date filter (ISO 8601)
+* `--end-date [%Y-%m-%d|%Y-%m-%dT%H:%M:%S|%Y-%m-%dT%H:%M:%S%z]`: End date filter (ISO 8601)
+* `--config-file TEXT`: [env var: INFRAHUBCTL_CONFIG; default: infrahubctl.toml]
+* `--help`: Show this message and exit.
+
+## `infrahubctl telemetry list`
+
+List telemetry snapshots with summary information.
+
+**Usage**:
+
+```console
+$ infrahubctl telemetry list [OPTIONS]
+```
+
+**Options**:
+
+* `--start-date [%Y-%m-%d|%Y-%m-%dT%H:%M:%S|%Y-%m-%dT%H:%M:%S%z]`: Start date filter (ISO 8601)
+* `--end-date [%Y-%m-%d|%Y-%m-%dT%H:%M:%S|%Y-%m-%dT%H:%M:%S%z]`: End date filter (ISO 8601)
+* `--limit INTEGER`: Maximum number of results  [default: 50]
+* `--config-file TEXT`: [env var: INFRAHUBCTL_CONFIG; default: infrahubctl.toml]
+* `--help`: Show this message and exit.

infrahub_sdk/ctl/cli_commands.py

@@ -35,6 +35,7 @@
 from ..ctl.repository import find_repository_config_file, get_repository_config
 from ..ctl.schema import app as schema_app
 from ..ctl.task import app as task_app
+from ..ctl.telemetry import app as telemetry_app
 from ..ctl.transform import list_transforms
 from ..ctl.utils import (
     catch_exception,
@@ -69,6 +70,7 @@
 app.add_typer(object_app, name="object")
 app.add_typer(graphql_app, name="graphql")
 app.add_typer(task_app, name="task")
+app.add_typer(telemetry_app, name="telemetry")
 
 app.command(name="dump")(dump)
 app.command(name="load")(load)

infrahub_sdk/ctl/telemetry.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlencode
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from ..async_typer import AsyncTyper
+from .client import initialize_client
+from .parameters import CONFIG_PARAM
+from .utils import catch_exception
+
+app = AsyncTyper()
+console = Console()
+
+EXPORT_PAGE_SIZE = 1000
+
+
+@app.command(name="list")
+@catch_exception(console=console)
+async def list_snapshots(
+    start_date: datetime | None = typer.Option(
+        None, help="Start date filter (ISO 8601)", formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"]
+    ),
+    end_date: datetime | None = typer.Option(
+        None, help="End date filter (ISO 8601)", formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"]
+    ),
+    limit: int = typer.Option(50, help="Maximum number of results"),
+    _: str = CONFIG_PARAM,
+) -> None:
+    """List telemetry snapshots with summary information."""
+    client = initialize_client()
+
+    params: dict[str, str | int] = {"limit": limit}
+    if start_date:
+        params["start_date"] = start_date.isoformat()
+    if end_date:
+        params["end_date"] = end_date.isoformat()
+
+    url = f"{client.address}/api/telemetry/snapshots?{urlencode(params)}"
+    response = await client._get(url=url, timeout=client.default_timeout)
+    response.raise_for_status()
+    data = response.json()
+
+    snapshots = data.get("snapshots", [])
+    if not snapshots:
+        console.print("No telemetry snapshots found.")
+        return
+
+    table = Table()
+    table.add_column("Date")
+    table.add_column("Version")
+    table.add_column("Type")
+    table.add_column("Deployment")
+    table.add_column("Remote Status")
+
+    for snap in snapshots:
+        table.add_row(
+            snap.get("created_at", ""),
+            snap.get("infrahub_version", ""),
+            snap.get("kind", ""),
+            snap.get("deployment_id", ""),
+            snap.get("remote_send_status", ""),
+        )
+
+    console.print(table)
+    console.print(f"Showing {len(snapshots)} of {data.get('count', len(snapshots))} total snapshots")
+
+
+@app.command(name="export")
+@catch_exception(console=console)
+async def export_snapshots(
+    output: str = typer.Option("telemetry-export.json", help="Output file path"),
+    start_date: datetime | None = typer.Option(
+        None, help="Start date filter (ISO 8601)", formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"]
+    ),
+    end_date: datetime | None = typer.Option(
+        None, help="End date filter (ISO 8601)", formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"]
+    ),
+    _: str = CONFIG_PARAM,
+) -> None:
+    """Export telemetry snapshots to a JSON file.
+
+    Pages through the API automatically so that all matching snapshots are exported,
+    not just the first page.
+    """
+    client = initialize_client()
+
+    base_params: dict[str, str | int] = {}
+    if start_date:
+        base_params["start_date"] = start_date.isoformat()
+    if end_date:
+        base_params["end_date"] = end_date.isoformat()
+
+    snapshots: list[dict[str, Any]] = []
+    offset = 0
+    total: int | None = None
+
+    while True:
+        params: dict[str, str | int] = {**base_params, "limit": EXPORT_PAGE_SIZE, "offset": offset}
+        url = f"{client.address}/api/telemetry/snapshots?{urlencode(params)}"
+        response = await client._get(url=url, timeout=client.default_timeout)
+        response.raise_for_status()
+        data = response.json()
+
+        page: list[dict[str, Any]] = data.get("snapshots", [])
+        snapshots.extend(page)
+
+        if total is None:
+            total = int(data.get("count", len(page)))
+
+        if len(page) < EXPORT_PAGE_SIZE or len(snapshots) >= total:
+            break
+
+        offset += EXPORT_PAGE_SIZE
+
+    if not snapshots:
+        console.print("No telemetry snapshots found.")
+        raise typer.Exit(code=2)
+
+    output_path = Path(output)
+    output_path.write_text(json.dumps(snapshots, indent=2), encoding="utf-8")
+    console.print(f"Exported {len(snapshots)} snapshots to {output_path}")