perf(npm): cache NPM JWT instead of re-authenticating on every API call

Every NPM helper (proxy host create/update/delete, streams, certs) did a
fresh POST /api/tokens login before its actual request, adding an avoidable
round-trip to every proxy/stream operation.

- Cache the JWT per (api_url, email), sized from its 'exp' claim
- Transparently re-authenticate and retry once on a 401 (e.g. after an NPM
  restart invalidates a cached token), so a stale cache entry can't cause a
  hard failure

Co-Authored-By: Claude Sonnet 5 <[email protected]>

app/main.py

@@ -33,7 +33,7 @@ logger = logging.getLogger(__name__)
 app = FastAPI(
     title="NetBird MSP Appliance",
     description="Multi-tenant NetBird management platform for MSPs",
-    version="1.0.0",
+    version="1.1.1",
     docs_url="/api/docs",
     redoc_url="/api/redoc",
     openapi_url="/api/openapi.json",

app/routers/monitoring.py

@@ -1,7 +1,9 @@
 """Monitoring API — system overview, customer statuses, host resources."""
 
+import asyncio
 import logging
 import platform
+import time
 from typing import Any
 
 import psutil
@@ -16,6 +18,13 @@ from app.services import docker_service, image_service
 logger = logging.getLogger(__name__)
 router = APIRouter()
 
+# Short-lived cache for the local update-status badges. This endpoint is
+# triggered on every customer-table render (i.e. every search keystroke), but
+# the underlying data (which images are outdated) only changes after an image
+# pull + container recreate, so a few seconds of staleness is harmless.
+_update_status_cache: dict[str, Any] = {"data": None, "expires": 0.0}
+_UPDATE_STATUS_TTL_SECONDS = 20
+
 
 @router.get("/status")
 async def system_status(
@@ -58,8 +67,7 @@ async def all_customers_status(
         .all()
     )
 
-    results: list[dict[str, Any]] = []
-    for c in customers:
+    async def _build_entry(c: Customer) -> dict[str, Any]:
         entry: dict[str, Any] = {
             "id": c.id,
             "name": c.name,
@@ -67,7 +75,7 @@ async def all_customers_status(
             "status": c.status,
         }
         if c.deployment:
-            containers = docker_service.get_container_status(c.deployment.container_prefix)
+            containers = await docker_service.get_container_status_async(c.deployment.container_prefix)
             entry["deployment_status"] = c.deployment.deployment_status
             entry["containers"] = containers
             entry["relay_udp_port"] = c.deployment.relay_udp_port
@@ -76,9 +84,11 @@ async def all_customers_status(
         else:
             entry["deployment_status"] = None
             entry["containers"] = []
-        results.append(entry)
+        return entry
 
-    return results
+    # Fetch container status for all customers concurrently instead of one
+    # blocking Docker SDK call at a time.
+    return await asyncio.gather(*[_build_entry(c) for c in customers])
 
 
 @router.get("/resources")
@@ -205,15 +215,28 @@ async def customers_local_update_status(
 
     Compares running container image IDs against locally stored images.
     No network call — safe to call on every dashboard load.
+
+    Results are cached for a few seconds since this is triggered on every
+    customer-table render (including every search keystroke) but the
+    underlying data rarely changes.
     """
+    now = time.monotonic()
+    if _update_status_cache["data"] is not None and now < _update_status_cache["expires"]:
+        return _update_status_cache["data"]
+
     config = db.query(SystemConfig).filter(SystemConfig.id == 1).first()
     if not config:
         return []
     deployments = db.query(Deployment).all()
-    results = []
-    for dep in deployments:
-        cs = image_service.get_customer_container_image_status(dep.container_prefix, config)
-        results.append({"customer_id": dep.customer_id, "needs_update": cs["needs_update"]})
+
+    async def _check(dep: Deployment) -> dict[str, Any]:
+        cs = await image_service.get_customer_container_image_status_async(dep.container_prefix, config)
+        return {"customer_id": dep.customer_id, "needs_update": cs["needs_update"]}
+
+    results = await asyncio.gather(*[_check(dep) for dep in deployments])
+    results = list(results)
+    _update_status_cache["data"] = results
+    _update_status_cache["expires"] = now + _UPDATE_STATUS_TTL_SECONDS
     return results
 
 

app/services/docker_service.py

@@ -27,13 +27,24 @@ async def _run_cmd(cmd: list[str], timeout: int = 120) -> subprocess.CompletedPr
     )
 
 
+_client: Optional[docker.DockerClient] = None
+
+
 def _get_client() -> docker.DockerClient:
-    """Return a Docker client connected via the Unix socket.
+    """Return a shared Docker client connected via the Unix socket.
+
+    The client is created once and reused — creating a new client per call
+    (as `docker.from_env()` does) re-negotiates the API version and opens a
+    fresh connection every time, which is wasteful when called once per
+    customer in a loop.
 
     Returns:
         docker.DockerClient instance.
     """
-    return docker.from_env()
+    global _client
+    if _client is None:
+        _client = docker.from_env()
+    return _client
 
 
 async def compose_up(
@@ -212,6 +223,16 @@ def get_container_status(container_prefix: str) -> list[dict[str, Any]]:
     return results
 
 
+async def get_container_status_async(container_prefix: str) -> list[dict[str, Any]]:
+    """Thread-offloaded wrapper around get_container_status().
+
+    Use this when checking status for multiple customers so the Docker SDK
+    calls run in the thread pool instead of blocking the event loop.
+    """
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(None, get_container_status, container_prefix)
+
+
 def get_container_logs(container_name: str, tail: int = 200) -> str:
     """Retrieve recent logs from a container.
 

app/services/image_service.py

@@ -211,6 +211,44 @@ async def pull_all_images(config) -> dict[str, Any]:
     }
 
 
+async def get_customer_container_image_status_async(container_prefix: str, config) -> dict[str, Any]:
+    """Async, thread-offloaded version of get_customer_container_image_status().
+
+    Runs the per-service `docker inspect` subprocess calls concurrently in the
+    thread pool instead of sequentially blocking the event loop — use this
+    whenever checking status for multiple customers (e.g. dashboard/search
+    badge refresh, monitoring overview).
+
+    Returns:
+        services: dict mapping service name to status info
+        needs_update: True if any service has a different image ID than locally stored
+    """
+    service_images = {
+        "management": config.netbird_management_image,
+        "signal": config.netbird_signal_image,
+        "relay": config.netbird_relay_image,
+        "dashboard": config.netbird_dashboard_image,
+    }
+    loop = asyncio.get_event_loop()
+
+    async def _check(svc: str, image: str) -> tuple[str, dict[str, Any]]:
+        container_name = f"{container_prefix}-{svc}"
+        container_id, local_id = await asyncio.gather(
+            loop.run_in_executor(None, get_container_image_id, container_name),
+            loop.run_in_executor(None, get_local_image_id, image),
+        )
+        if container_id and local_id:
+            up_to_date = container_id == local_id
+        else:
+            up_to_date = None  # container not running or image not pulled
+        return svc, {"container": container_name, "image": image, "up_to_date": up_to_date}
+
+    pairs = await asyncio.gather(*[_check(svc, image) for svc, image in service_images.items()])
+    services = dict(pairs)
+    needs_update = any(s["up_to_date"] is False for s in services.values())
+    return {"services": services, "needs_update": needs_update}
+
+
 def get_customer_container_image_status(container_prefix: str, config) -> dict[str, Any]:
     """Check which service containers are running outdated local images.
 

app/services/npm_service.py

@@ -12,9 +12,12 @@ Let's Encrypt SSL certificates.
 Also manages NPM streams for STUN/TURN relay UDP ports.
 """
 
+import base64
+import json
 import logging
 import os
 import socket
+import time
 from typing import Any
 
 import httpx
@@ -24,6 +27,14 @@ logger = logging.getLogger(__name__)
 # Timeout for NPM API calls (seconds)
 NPM_TIMEOUT = 30
 
+# Cached JWTs, keyed by (api_url, email). NPM issues a token that stays valid
+# for a while (per its 'exp' claim), so re-logging in on every single API
+# call — as this module used to do — adds a full extra round-trip per action
+# for no reason.
+_token_cache: dict[tuple[str, str], dict[str, Any]] = {}
+_TOKEN_SAFETY_MARGIN = 60  # refresh this many seconds before actual expiry
+_DEFAULT_TOKEN_TTL = 3600  # fallback if the 'exp' claim can't be parsed
+
 
 def _get_forward_host() -> str:
     """Get the host machine's real IP address for NPM forwarding.
@@ -90,6 +101,61 @@ async def _npm_login(client: httpx.AsyncClient, api_url: str, email: str, passwo
     )
 
 
+def _decode_jwt_exp(token: str) -> float | None:
+    """Best-effort decode of a JWT's 'exp' claim, without verifying the signature.
+
+    We only use this to size our own cache TTL — NPM itself still enforces
+    the real expiry server-side, so an inaccurate read here is harmless.
+    """
+    try:
+        payload_b64 = token.split(".")[1]
+        padding = "=" * (-len(payload_b64) % 4)
+        payload = json.loads(base64.urlsafe_b64decode(payload_b64 + padding))
+        return payload.get("exp")
+    except Exception:
+        return None
+
+
+async def _get_token(
+    client: httpx.AsyncClient, api_url: str, email: str, password: str, force_refresh: bool = False
+) -> str:
+    """Return a cached NPM JWT if still valid, otherwise log in and cache it."""
+    cache_key = (api_url, email)
+    if not force_refresh:
+        cached = _token_cache.get(cache_key)
+        if cached and time.time() < cached["expires_at"]:
+            return cached["token"]
+
+    token = await _npm_login(client, api_url, email, password)
+    exp = _decode_jwt_exp(token)
+    expires_at = (exp - _TOKEN_SAFETY_MARGIN) if exp else (time.time() + _DEFAULT_TOKEN_TTL)
+    _token_cache[cache_key] = {"token": token, "expires_at": expires_at}
+    return token
+
+
+async def _request_with_reauth(
+    client: httpx.AsyncClient,
+    method: str,
+    api_url: str,
+    email: str,
+    password: str,
+    path: str,
+    headers: dict,
+    **kwargs: Any,
+) -> tuple[httpx.Response, dict]:
+    """Perform a request; if the cached token was rejected, refresh and retry once.
+
+    Returns the response and the (possibly updated) headers dict, so callers
+    can reuse the fresh token for any further requests in the same session.
+    """
+    resp = await client.request(method, f"{api_url}{path}", headers=headers, **kwargs)
+    if resp.status_code == 401:
+        token = await _get_token(client, api_url, email, password, force_refresh=True)
+        headers = {**headers, "Authorization": f"Bearer {token}"}
+        resp = await client.request(method, f"{api_url}{path}", headers=headers, **kwargs)
+    return resp, headers
+
+
 async def test_npm_connection(api_url: str, email: str, password: str) -> dict[str, Any]:
     """Test connectivity to NPM by logging in and listing proxy hosts.
 
@@ -103,9 +169,11 @@ async def test_npm_connection(api_url: str, email: str, password: str) -> dict[s
     """
     try:
         async with httpx.AsyncClient(timeout=NPM_TIMEOUT) as client:
-            token = await _npm_login(client, api_url, email, password)
+            token = await _get_token(client, api_url, email, password)
             headers = {"Authorization": f"Bearer {token}"}
-            resp = await client.get(f"{api_url}/nginx/proxy-hosts", headers=headers)
+            resp, headers = await _request_with_reauth(
+                client, "GET", api_url, email, password, "/nginx/proxy-hosts", headers
+            )
             if resp.status_code == 200:
                 count = len(resp.json())
                 return {"ok": True, "message": f"Connected. Login OK. {count} proxy hosts found."}
@@ -136,9 +204,11 @@ async def list_certificates(api_url: str, email: str, password: str) -> dict[str
     """
     try:
         async with httpx.AsyncClient(timeout=NPM_TIMEOUT) as client:
-            token = await _npm_login(client, api_url, email, password)
+            token = await _get_token(client, api_url, email, password)
             headers = {"Authorization": f"Bearer {token}"}
-            resp = await client.get(f"{api_url}/nginx/certificates", headers=headers)
+            resp, headers = await _request_with_reauth(
+                client, "GET", api_url, email, password, "/nginx/certificates", headers
+            )
             if resp.status_code == 200:
                 result = []
                 for cert in resp.json():
@@ -282,14 +352,15 @@ async def create_proxy_host(
 
     try:
         async with httpx.AsyncClient(timeout=180) as client:  # Long timeout for LE cert
-            token = await _npm_login(client, api_url, npm_email, npm_password)
+            token = await _get_token(client, api_url, npm_email, npm_password)
             headers = {
                 "Authorization": f"Bearer {token}",
                 "Content-Type": "application/json",
             }
 
-            resp = await client.post(
-                f"{api_url}/nginx/proxy-hosts", json=payload, headers=headers
+            resp, headers = await _request_with_reauth(
+                client, "POST", api_url, npm_email, npm_password,
+                "/nginx/proxy-hosts", headers, json=payload,
             )
             if resp.status_code in (200, 201):
                 data = resp.json()
@@ -542,14 +613,15 @@ async def create_stream(
 
     try:
         async with httpx.AsyncClient(timeout=NPM_TIMEOUT) as client:
-            token = await _npm_login(client, api_url, npm_email, npm_password)
+            token = await _get_token(client, api_url, npm_email, npm_password)
             headers = {
                 "Authorization": f"Bearer {token}",
                 "Content-Type": "application/json",
             }
 
-            resp = await client.post(
-                f"{api_url}/nginx/streams", json=payload, headers=headers
+            resp, headers = await _request_with_reauth(
+                client, "POST", api_url, npm_email, npm_password,
+                "/nginx/streams", headers, json=payload,
             )
             if resp.status_code in (200, 201):
                 data = resp.json()
@@ -587,10 +659,11 @@ async def delete_stream(
     """
     try:
         async with httpx.AsyncClient(timeout=NPM_TIMEOUT) as client:
-            token = await _npm_login(client, api_url, npm_email, npm_password)
+            token = await _get_token(client, api_url, npm_email, npm_password)
             headers = {"Authorization": f"Bearer {token}"}
-            resp = await client.delete(
-                f"{api_url}/nginx/streams/{stream_id}", headers=headers
+            resp, headers = await _request_with_reauth(
+                client, "DELETE", api_url, npm_email, npm_password,
+                f"/nginx/streams/{stream_id}", headers,
             )
             if resp.status_code in (200, 204):
                 logger.info("Deleted NPM stream %d", stream_id)
@@ -623,10 +696,11 @@ async def delete_proxy_host(
     """
     try:
         async with httpx.AsyncClient(timeout=NPM_TIMEOUT) as client:
-            token = await _npm_login(client, api_url, npm_email, npm_password)
+            token = await _get_token(client, api_url, npm_email, npm_password)
             headers = {"Authorization": f"Bearer {token}"}
-            resp = await client.delete(
-                f"{api_url}/nginx/proxy-hosts/{proxy_id}", headers=headers
+            resp, headers = await _request_with_reauth(
+                client, "DELETE", api_url, npm_email, npm_password,
+                f"/nginx/proxy-hosts/{proxy_id}", headers,
             )
             if resp.status_code in (200, 204):
                 logger.info("Deleted NPM proxy host %d", proxy_id)