[INFRA 6] Add observability layer for API and scan worker

api/app.py

@@ -10,17 +10,17 @@
 from flask_cors import CORS
 
 from api.models.finding import DatabaseManager
+from api.observability import configure_logging, get_request_id, init_app, init_sentry
 
 load_dotenv()
 
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
-)
+configure_logging()
+init_sentry()
 logger = logging.getLogger(__name__)
 
-# Paths that are always public regardless of environment or demo mode
-_ALWAYS_PUBLIC = {"/", "/health"}
+# Paths that are always public regardless of environment or demo mode.
+# /ready and /metrics are probe/scrape endpoints and must never require auth.
+_ALWAYS_PUBLIC = {"/", "/health", "/ready", "/metrics"}
 
 _INSECURE_JWT_DEFAULT = "change-me-in-production"
 _MIN_JWT_SECRET_LENGTH = 32
@@ -92,6 +92,14 @@ def create_app() -> Flask:
     """
     app = Flask(__name__)
 
+    # ------------------------------------------------------------------ #
+    # Observability                                                        #
+    # ------------------------------------------------------------------ #
+    # Registered first so request IDs and request timing are available to
+    # every later before_request handler (including JWT auth) and to the
+    # error handlers. Also mounts the public /metrics endpoint.
+    init_app(app)
+
     # ------------------------------------------------------------------ #
     # Configuration & Security                                             #
     # ------------------------------------------------------------------ #
@@ -164,7 +172,12 @@ def verify_jwt() -> None:
 
         auth = request.headers.get("Authorization", "")
         if not auth.startswith("Bearer "):
-            return jsonify({"error": "Missing or malformed Authorization header"}), 401
+            return jsonify(
+                {
+                    "error": "Missing or malformed Authorization header",
+                    "request_id": get_request_id(),
+                }
+            ), 401
 
         token = auth.split(" ", 1)[1]
         try:
@@ -175,9 +188,10 @@ def verify_jwt() -> None:
             )
             g.user = payload
         except jwt.ExpiredSignatureError:
-            return jsonify({"error": "Token has expired"}), 401
+            return jsonify({"error": "Token has expired", "request_id": get_request_id()}), 401
         except jwt.InvalidTokenError as exc:
-            return jsonify({"error": f"Invalid token: {exc}"}), 401
+            logger.warning("Invalid JWT token: %s", exc)
+            return jsonify({"error": "Invalid token", "request_id": get_request_id()}), 401
 
         return None
 
@@ -214,32 +228,44 @@ def index():
 
     @app.get("/health")
     def health():
+        """Liveness probe: process is up. Deliberately does not touch the DB."""
         return jsonify({"status": "ok"})
 
+    @app.get("/ready")
+    def ready():
+        """Readiness probe: 200 when the database is reachable, else 503."""
+        try:
+            db = DatabaseManager()
+            db.ping()
+            return jsonify({"status": "ready"}), 200
+        except Exception as exc:
+            logger.warning("Readiness check failed: %s", exc)
+            return jsonify({"status": "not_ready", "error": "database_unreachable"}), 503
+
     # ------------------------------------------------------------------ #
     # Error handlers                                                        #
     # ------------------------------------------------------------------ #
 
     @app.errorhandler(400)
     def bad_request(exc):
-        return jsonify({"error": "Bad request", "detail": str(exc)}), 400
+        return jsonify({"error": "Bad request", "detail": str(exc), "request_id": get_request_id()}), 400
 
     @app.errorhandler(401)
     def unauthorized(exc):
-        return jsonify({"error": "Unauthorized"}), 401
+        return jsonify({"error": "Unauthorized", "request_id": get_request_id()}), 401
 
     @app.errorhandler(403)
     def forbidden(exc):
-        return jsonify({"error": "Forbidden"}), 403
+        return jsonify({"error": "Forbidden", "request_id": get_request_id()}), 403
 
     @app.errorhandler(404)
     def not_found(exc):
-        return jsonify({"error": "Not found"}), 404
+        return jsonify({"error": "Not found", "request_id": get_request_id()}), 404
 
     @app.errorhandler(500)
     def internal_error(exc):
         logger.error("Unhandled exception: %s", exc)
-        return jsonify({"error": "Internal server error"}), 500
+        return jsonify({"error": "Internal server error", "request_id": get_request_id()}), 500
 
     logger.info("OpenShield API created - %d blueprints registered", len(app.blueprints))
     return app

api/models/finding.py

@@ -103,6 +103,18 @@ def close(self) -> None:
             self.conn = None
             logger.debug("Database connection closed")
 
+    def ping(self) -> bool:
+        """Execute a trivial query to confirm database connectivity.
+
+        Used by the /ready probe. Raises on failure so the caller can map it
+        to a 503; returns True when the database answers.
+        """
+        conn = self._get_conn()
+        with conn.cursor() as cur:
+            cur.execute("SELECT 1")
+            cur.fetchone()
+        return True
+
     # ------------------------------------------------------------------ #
     # Schema                                                                #
     # ------------------------------------------------------------------ #

api/observability.py

@@ -0,0 +1,180 @@
+"""Shared observability layer: structured logging, request IDs, Prometheus
+metrics and optional Sentry error tracking.
+
+This module is intentionally free of any project imports so that it can be
+used from both the API (``api.app``) and the background worker
+(``scanner.worker``) without creating an import cycle.
+
+Metric cardinality note: no metric is ever labelled with a request ID,
+scan ID, subscription ID, resource ID, error message or free-form user
+input. Only low-cardinality, bounded values (HTTP method, view name,
+status code, scan status, rule ID and provider name) are used as labels.
+"""
+
+import logging
+import os
+import time
+import uuid
+
+from flask import Flask, Response, g, request
+from prometheus_client import (
+    CONTENT_TYPE_LATEST,
+    Counter,
+    Gauge,
+    Histogram,
+    generate_latest,
+)
+
+# python-json-logger moved ``JsonFormatter`` from ``pythonjsonlogger.jsonlogger``
+# to ``pythonjsonlogger.json`` in 3.1. Prefer the new path, fall back for <3.1.
+try:  # pragma: no cover - trivial import shim
+    from pythonjsonlogger.json import JsonFormatter
+except ImportError:  # pragma: no cover
+    from pythonjsonlogger.jsonlogger import JsonFormatter
+
+logger = logging.getLogger(__name__)
+
+REQUEST_ID_HEADER = "X-Request-ID"
+
+# --------------------------------------------------------------------------- #
+# Prometheus metrics                                                           #
+# --------------------------------------------------------------------------- #
+# Defined at module import so they are process-wide singletons. Re-importing
+# this module (e.g. across repeated create_app() calls in tests) reuses the
+# same collectors and never double-registers.
+
+HTTP_REQUESTS_TOTAL = Counter(
+    "openshield_http_requests_total",
+    "Total HTTP requests processed by the API.",
+    ["method", "endpoint", "status"],
+)
+HTTP_REQUEST_DURATION_SECONDS = Histogram(
+    "openshield_http_request_duration_seconds",
+    "HTTP request latency in seconds.",
+    ["method", "endpoint"],
+)
+SCANS_TOTAL = Counter(
+    "openshield_scans_total",
+    "Total scans processed by the worker, by terminal status.",
+    ["status"],
+)
+SCAN_DURATION_SECONDS = Histogram(
+    "openshield_scan_duration_seconds",
+    "Wall-clock duration of a single scan execution in seconds.",
+)
+PENDING_SCANS = Gauge(
+    "openshield_pending_scans",
+    "Number of scans currently waiting in the pending queue.",
+)
+RULE_ERRORS_TOTAL = Counter(
+    "openshield_rule_errors_total",
+    "Total number of times a scanner rule raised an exception.",
+    ["rule_id"],
+)
+NVD_REQUEST_LATENCY_SECONDS = Histogram(
+    "openshield_nvd_request_latency_seconds",
+    "Latency of outbound NVD HTTP requests in seconds.",
+)
+LLM_PROVIDER_LATENCY_SECONDS = Histogram(
+    "openshield_llm_provider_latency_seconds",
+    "Latency of outbound LLM provider requests in seconds.",
+    ["provider"],
+)
+
+
+# --------------------------------------------------------------------------- #
+# Structured logging                                                           #
+# --------------------------------------------------------------------------- #
+def configure_logging(level: int = logging.INFO) -> None:
+    """Configure the root logger to emit structured JSON to stderr.
+
+    Safe to call multiple times — it replaces the root handlers each call so
+    repeated invocations (API import, worker start, tests) do not stack
+    duplicate handlers. Any ``extra={...}`` fields passed to log calls are
+    included as top-level JSON keys.
+    """
+    handler = logging.StreamHandler()
+    handler.setFormatter(JsonFormatter("%(asctime)s %(levelname)s %(name)s %(message)s"))
+    root = logging.getLogger()
+    root.handlers = [handler]
+    root.setLevel(level)
+
+
+# --------------------------------------------------------------------------- #
+# Sentry (optional)                                                            #
+# --------------------------------------------------------------------------- #
+def init_sentry() -> bool:
+    """Initialise Sentry error tracking only when ``SENTRY_DSN`` is set.
+
+    Returns ``True`` when Sentry was initialised, ``False`` otherwise (DSN
+    unset, or ``sentry-sdk`` not installed). Never raises.
+    """
+    dsn = os.environ.get("SENTRY_DSN")
+    if not dsn:
+        return False
+    try:
+        import sentry_sdk
+    except ImportError:
+        logger.warning("SENTRY_DSN is set but sentry-sdk is not installed; skipping.")
+        return False
+
+    try:
+        sentry_sdk.init(
+            dsn=dsn,
+            environment=os.environ.get("OPENSHIELD_ENV", "development"),
+            traces_sample_rate=float(os.environ.get("SENTRY_TRACES_SAMPLE_RATE", "0.0")),
+        )
+    except Exception as exc:
+        # A malformed DSN must never take down the API or the worker.
+        logger.warning("Failed to initialise Sentry: %s", exc)
+        return False
+    logger.info("Sentry error tracking initialised.")
+    return True
+
+
+# --------------------------------------------------------------------------- #
+# Request IDs                                                                  #
+# --------------------------------------------------------------------------- #
+def get_request_id() -> str:
+    """Return the request ID bound to the current request context.
+
+    Falls back to a freshly generated UUID if the middleware has not run
+    (e.g. outside a request), so callers always receive a usable value.
+    """
+    rid = getattr(g, "request_id", None)
+    if not rid:
+        rid = str(uuid.uuid4())
+        g.request_id = rid
+    return rid
+
+
+# --------------------------------------------------------------------------- #
+# Flask wiring                                                                 #
+# --------------------------------------------------------------------------- #
+def init_app(app: Flask) -> None:
+    """Register request-ID and Prometheus middleware and the /metrics route.
+
+    Must be called before any other ``before_request`` handlers (such as JWT
+    auth) so that ``g.request_id`` is available to them and to error handlers.
+    """
+
+    @app.before_request
+    def _start_observability() -> None:
+        g.request_id = request.headers.get(REQUEST_ID_HEADER) or str(uuid.uuid4())
+        g.request_start_time = time.perf_counter()
+
+    @app.after_request
+    def _record_observability(response: Response) -> Response:
+        response.headers[REQUEST_ID_HEADER] = get_request_id()
+
+        endpoint = request.endpoint or "unknown"
+        method = request.method
+        HTTP_REQUESTS_TOTAL.labels(method=method, endpoint=endpoint, status=response.status_code).inc()
+        start = getattr(g, "request_start_time", None)
+        if start is not None:
+            HTTP_REQUEST_DURATION_SECONDS.labels(method=method, endpoint=endpoint).observe(time.perf_counter() - start)
+        return response
+
+    @app.get("/metrics")
+    def metrics() -> Response:
+        return Response(generate_latest(), content_type=CONTENT_TYPE_LATEST)

api/services/ai_provider.py

@@ -3,6 +3,8 @@
 import logging
 import requests
 
+from api.observability import LLM_PROVIDER_LATENCY_SECONDS
+
 logger = logging.getLogger(__name__)
 
 PROVIDERS = ("anthropic", "groq", "gemini")
@@ -23,11 +25,12 @@ def get_completion(provider: str, api_key: str, prompt: str, model: str = None)
 
     resolved_model = model or DEFAULT_MODELS[provider]
 
-    if provider == "anthropic":
-        return _anthropic(api_key, prompt, resolved_model)
-    if provider == "groq":
-        return _groq(api_key, prompt, resolved_model)
-    return _gemini(api_key, prompt, resolved_model)
+    with LLM_PROVIDER_LATENCY_SECONDS.labels(provider=provider).time():
+        if provider == "anthropic":
+            return _anthropic(api_key, prompt, resolved_model)
+        if provider == "groq":
+            return _groq(api_key, prompt, resolved_model)
+        return _gemini(api_key, prompt, resolved_model)
 
 
 def _anthropic(api_key: str, prompt: str, model: str) -> str:

requirements.txt

@@ -27,5 +27,8 @@ azure-keyvault-keys==4.9.0
 chromadb==0.4.24
 sentence-transformers==2.7.0
 numpy<2.0
+prometheus-client>=0.19.0
+python-json-logger>=2.0.7
+sentry-sdk>=1.40.0
 pytest>=7.4.0
 pytest-cov>=4.1.0

scanner/engine.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from api.observability import RULE_ERRORS_TOTAL
 from scanner.azure_client import AzureClient
 
 logger = logging.getLogger(__name__)
@@ -121,6 +122,7 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]:
                 findings.extend(rule_findings)
                 logger.info("Rule %s produced %d finding(s)", rule_id, len(rule_findings))
             except Exception as exc:
+                RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc()
                 logger.error("Rule %s raised an exception: %s", rule_id, exc, exc_info=True)
 
         completed_at = datetime.now(timezone.utc).isoformat()

scanner/nvd_client.py

@@ -24,6 +24,8 @@
 import json
 from typing import Optional
 
+from api.observability import NVD_REQUEST_LATENCY_SECONDS
+
 logger = logging.getLogger(__name__)
 
 _NVD_BASE_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
@@ -143,8 +145,9 @@ def query_nvd(keyword: str, results_per_page: int = _RESULTS_PER_PAGE) -> list[d
                 headers={"User-Agent": "OpenShield/0.1 (github.com/openshield-org/openshield)"},
             )
             # URL host is the hardcoded NVD API base, not user-controlled
-            with urllib.request.urlopen(req, timeout=10) as resp:  # nosec B310
-                data = json.loads(resp.read())
+            with NVD_REQUEST_LATENCY_SECONDS.time():
+                with urllib.request.urlopen(req, timeout=10) as resp:  # nosec B310
+                    data = json.loads(resp.read())
 
             vulnerabilities = data.get("vulnerabilities", [])
             results = [parsed for item in vulnerabilities if (parsed := _parse_cve_item(item)) is not None]