observability: add tenant analytics dashboard

Author: haasonsaas

src/nimbus/common/metrics.py

@@ -76,5 +76,11 @@ def register(self, metric: object) -> object:
     def render(self) -> str:
         return "\n".join(metric.render() for metric in self._metrics.values()) + "\n"
 
+    def get_metric(self, name: str) -> object:
+        return self._metrics[name]
+
+    def get(self, name: str) -> object | None:
+        return self._metrics.get(name)
+
 
 GLOBAL_REGISTRY = MetricsRegistry()

src/nimbus/common/observability.py

@@ -10,7 +10,10 @@
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
-from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
+try:  # pragma: no cover - optional dependency
+    from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
+except ModuleNotFoundError:  # pragma: no cover - fallback when extra not installed
+    HTTPXClientInstrumentor = None  # type: ignore[assignment]
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor
@@ -111,7 +114,7 @@ def configure_tracing(
     _tracer_configured = True
 
     global _httpx_instrumented
-    if not _httpx_instrumented:
+    if not _httpx_instrumented and HTTPXClientInstrumentor is not None:
         HTTPXClientInstrumentor().instrument()
         _httpx_instrumented = True
 

src/nimbus/control_plane/app.py

@@ -50,6 +50,7 @@
 from . import db
 from .github import GitHubAppClient
 from .jobs import QUEUE_KEY, enqueue_job, lease_job, lease_job_with_fence
+from .observability import build_org_overview
 from ..common.security import mint_cache_token
 from ..common.observability import configure_logging, configure_tracing, instrument_fastapi_app
 from ..common.networking import (
@@ -1207,12 +1208,13 @@ async def job_status(
     @app.get("/api/jobs/recent", response_model=list[JobRecord])
     async def recent_jobs(
         limit: int = 50,
+        org_id: Optional[int] = None,
         _: str = Depends(verify_agent_token),
         session: AsyncSession = Depends(get_session),
     ) -> list[JobRecord]:
         REQUEST_COUNTER.inc()
         limit = max(1, min(limit, 200))
-        rows = await db.list_recent_jobs(session, limit=limit)
+        rows = await db.list_recent_jobs(session, limit=limit, org_id=org_id)
         return [JobRecord.model_validate(row) for row in rows]
 
     @app.get("/api/status", status_code=status.HTTP_200_OK)
@@ -1229,6 +1231,34 @@ async def service_status(
             "jobs_by_status": counts,
         }
 
+    @app.get("/api/observability/orgs")
+    async def observability_orgs_endpoint(
+        limit: int = 50,
+        hours_back: Optional[int] = None,
+        _: str = Depends(verify_admin_token),
+        session: AsyncSession = Depends(get_session),
+    ) -> list[dict]:
+        REQUEST_COUNTER.inc()
+        limit = max(1, min(limit, 200))
+        org_ids = await db.distinct_org_ids(session, hours_back=hours_back)
+        if limit and len(org_ids) > limit:
+            org_ids = org_ids[:limit]
+        status_rows = await db.org_job_status_counts(session, hours_back=hours_back)
+        last_activity = await db.org_last_activity(session, hours_back=hours_back)
+        agents = await db.org_active_agents(session, hours_back=24)
+        failures: dict[int, list[dict]] = {}
+        for org_id in org_ids:
+            failure_rows = await db.list_recent_failures(session, org_id, limit=5)
+            failures[org_id] = failure_rows
+        summaries = build_org_overview(
+            org_ids,
+            status_rows=status_rows,
+            last_activity=last_activity,
+            active_agents=agents,
+            failures=failures,
+        )
+        return summaries
+
     @app.post("/api/agents/token", response_model=AgentTokenResponse)
     async def mint_agent_token_endpoint(
         request_body: AgentTokenMintRequest,

src/nimbus/control_plane/db.py

@@ -138,6 +138,19 @@ def session_factory(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
     return async_sessionmaker(engine, expire_on_commit=False)
 
 
+def _normalise_job_rows(rows: Iterable[dict]) -> list[dict]:
+    normalised: list[dict] = []
+    for row in rows:
+        payload = dict(row)
+        payload["repo_private"] = True if payload.get("repo_private") == "true" else False
+        for key in ("queued_at", "leased_at", "completed_at", "updated_at"):
+            value = payload.get(key)
+            if isinstance(value, datetime):
+                payload[key] = value.isoformat()
+        normalised.append(payload)
+    return normalised
+
+
 async def record_job_queued(session: AsyncSession, assignment: JobAssignment) -> None:
     now = datetime.now(timezone.utc)
     repo = assignment.repository
@@ -209,30 +222,143 @@ async def record_status_update(session: AsyncSession, update_payload: JobStatusU
 
 
 async def list_recent_jobs(
-    session: AsyncSession, limit: int = 50
+    session: AsyncSession, limit: int = 50, org_id: Optional[int] = None
 ) -> Iterable[dict]:
     stmt = (
         select(jobs_table)
         .order_by(jobs_table.c.updated_at.desc())
         .limit(limit)
     )
+    if org_id is not None:
+        stmt = stmt.where(jobs_table.c.org_id == org_id)
     result = await session.execute(stmt)
     rows = [dict(row) for row in result.mappings()]
-    for row in rows:
-        row["repo_private"] = True if row.get("repo_private") == "true" else False
-        for key in ("queued_at", "leased_at", "completed_at", "updated_at"):
-            value = row.get(key)
-            if isinstance(value, datetime):
-                row[key] = value.isoformat()
-    return rows
+    return _normalise_job_rows(rows)
 
 
-async def job_status_counts(session: AsyncSession) -> dict[str, int]:
+async def job_status_counts(
+    session: AsyncSession,
+    *,
+    org_id: Optional[int] = None,
+) -> dict[str, int]:
     stmt = select(jobs_table.c.status, func.count().label("count")).group_by(jobs_table.c.status)
+    if org_id is not None:
+        stmt = stmt.where(jobs_table.c.org_id == org_id)
     result = await session.execute(stmt)
     return {row.status: row.count for row in result}
 
 
+async def distinct_org_ids(
+    session: AsyncSession,
+    *,
+    hours_back: Optional[int] = None,
+) -> list[int]:
+    stmt = select(jobs_table.c.org_id).distinct().where(jobs_table.c.org_id.isnot(None))
+    if hours_back is not None:
+        cutoff = datetime.now(timezone.utc) - timedelta(hours=hours_back)
+        stmt = stmt.where(jobs_table.c.updated_at >= cutoff)
+    result = await session.execute(stmt)
+    org_ids = [int(row.org_id) for row in result if row.org_id is not None]
+    return sorted(set(org_ids))
+
+
+async def org_job_status_counts(
+    session: AsyncSession,
+    *,
+    org_id: Optional[int] = None,
+    hours_back: Optional[int] = None,
+) -> list[dict]:
+    stmt = select(
+        jobs_table.c.org_id,
+        jobs_table.c.status,
+        func.count().label("count"),
+    )
+    if org_id is not None:
+        stmt = stmt.where(jobs_table.c.org_id == org_id)
+    if hours_back is not None:
+        cutoff = datetime.now(timezone.utc) - timedelta(hours=hours_back)
+        stmt = stmt.where(jobs_table.c.updated_at >= cutoff)
+    stmt = stmt.group_by(jobs_table.c.org_id, jobs_table.c.status)
+    result = await session.execute(stmt)
+    return [
+        {
+            "org_id": int(row.org_id),
+            "status": row.status,
+            "count": row.count,
+        }
+        for row in result
+        if row.org_id is not None
+    ]
+
+
+async def org_last_activity(
+    session: AsyncSession,
+    *,
+    org_id: Optional[int] = None,
+    hours_back: Optional[int] = None,
+) -> dict[int, datetime]:
+    stmt = select(
+        jobs_table.c.org_id,
+        func.max(jobs_table.c.updated_at).label("last_updated"),
+    )
+    if org_id is not None:
+        stmt = stmt.where(jobs_table.c.org_id == org_id)
+    if hours_back is not None:
+        cutoff = datetime.now(timezone.utc) - timedelta(hours=hours_back)
+        stmt = stmt.where(jobs_table.c.updated_at >= cutoff)
+    stmt = stmt.group_by(jobs_table.c.org_id)
+    result = await session.execute(stmt)
+    activity: dict[int, datetime] = {}
+    for row in result:
+        if row.org_id is None:
+            continue
+        activity[int(row.org_id)] = row.last_updated
+    return activity
+
+
+async def org_active_agents(
+    session: AsyncSession,
+    *,
+    org_id: Optional[int] = None,
+    hours_back: int = 24,
+) -> dict[int, set[str]]:
+    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours_back)
+    stmt = select(jobs_table.c.org_id, jobs_table.c.agent_id).where(
+        jobs_table.c.agent_id.isnot(None),
+        jobs_table.c.updated_at >= cutoff,
+    )
+    if org_id is not None:
+        stmt = stmt.where(jobs_table.c.org_id == org_id)
+    result = await session.execute(stmt)
+    mapping: dict[int, set[str]] = {}
+    for row in result:
+        if row.org_id is None or not row.agent_id:
+            continue
+        bucket = mapping.setdefault(int(row.org_id), set())
+        bucket.add(str(row.agent_id))
+    return mapping
+
+
+async def list_recent_failures(
+    session: AsyncSession,
+    org_id: int,
+    *,
+    limit: int = 10,
+) -> list[dict]:
+    stmt = (
+        select(jobs_table)
+        .where(
+            jobs_table.c.org_id == org_id,
+            jobs_table.c.status.in_(["failed", "cancelled"]),
+        )
+        .order_by(jobs_table.c.updated_at.desc())
+        .limit(limit)
+    )
+    result = await session.execute(stmt)
+    rows = [dict(row) for row in result.mappings()]
+    return _normalise_job_rows(rows)
+
+
 async def get_job(session: AsyncSession, job_id: int) -> Optional[dict]:
     stmt = select(jobs_table).where(jobs_table.c.job_id == job_id).limit(1)
     result = await session.execute(stmt)

src/nimbus/control_plane/observability.py

@@ -0,0 +1,35 @@
+"""Tenant observability aggregations for the Nimbus control plane."""
+
+from __future__ import annotations
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Iterable
+
+
+def build_org_overview(
+    org_ids: Iterable[int],
+    *,
+    status_rows: list[dict],
+    last_activity: dict[int, datetime],
+    active_agents: dict[int, set[str]],
+    failures: dict[int, list[dict]],
+) -> list[dict]:
+    status_map: dict[int, dict[str, int]] = {}
+    for row in status_rows:
+        org_id = int(row["org_id"])
+        status_map.setdefault(org_id, {})[row["status"]] = row["count"]
+
+    summaries: list[dict] = []
+    for org_id in org_ids:
+        summaries.append(
+            {
+                "org_id": int(org_id),
+                "status_counts": status_map.get(org_id, {}),
+                "last_activity": last_activity.get(org_id),
+                "active_agents": sorted(active_agents.get(org_id, set())),
+                "recent_failures": failures.get(org_id, []),
+            }
+        )
+    return summaries

tests/conftest.py

@@ -0,0 +1,63 @@
+import sys
+from types import ModuleType, SimpleNamespace
+
+
+def _ensure_module(name: str, **attrs) -> None:
+    if name in sys.modules:
+        return
+    module = ModuleType(name)
+    for attr, value in attrs.items():
+        setattr(module, attr, value)
+    sys.modules[name] = module
+
+
+class _SigstoreVerify(ModuleType):
+    def __getattr__(self, name):
+        raise ModuleNotFoundError(f"optional sigstore.verify dependency not installed: {name}")
+
+
+_ensure_module(
+    "boto3",
+    client=lambda *_args, **_kwargs: SimpleNamespace(),
+    session=SimpleNamespace(Session=lambda: SimpleNamespace(client=lambda *_a, **_k: SimpleNamespace())),
+)
+_ensure_module(
+    "botocore.exceptions",
+    ClientError=type(
+        "ClientError",
+        (Exception,),
+        {"__init__": lambda self, error_response=None, operation_name=None: Exception.__init__(self)},
+    ),
+)
+_ensure_module(
+    "pyroute2",
+    IPRoute=SimpleNamespace,
+    NetNS=SimpleNamespace,
+    NetlinkError=Exception,
+    netns=SimpleNamespace,
+)
+sigstore_module = ModuleType("sigstore")
+sigstore_verify = _SigstoreVerify("sigstore.verify")
+sigstore_module.verify = sigstore_verify
+_ensure_module("sigstore", **{"verify": sigstore_verify})
+_ensure_module(
+    "sigstore.verify",
+    VerificationMaterials=object,
+    verifier=lambda *args, **kwargs: SimpleNamespace(verify=lambda *a, **kw: None),
+)
+# SAML optional dependency stubs
+saml_module = ModuleType("nimbus.control_plane.saml")
+saml_module.SamlSettings = lambda **_kwargs: SimpleNamespace(**_kwargs)  # type: ignore[attr-defined]
+
+
+class _DummySamlAuthenticator:
+    def __init__(self, *_args, **_kwargs):
+        pass
+
+    def generate_session_token(self, *_args, **_kwargs):
+        return SimpleNamespace(to_dict=lambda: {"token": "stub"})
+
+
+saml_module.SamlAuthenticator = _DummySamlAuthenticator  # type: ignore[attr-defined]
+saml_module.SamlValidationError = Exception  # type: ignore[attr-defined]
+sys.modules.setdefault("nimbus.control_plane.saml", saml_module)