Add Metrics endpoint for the api

api/kubeseal_webgui_api/app.py

@@ -1,14 +1,34 @@
 from contextlib import asynccontextmanager
 import logging
+import time
 
 import fastapi
 from fastapi.middleware.cors import CORSMiddleware
+from opentelemetry import metrics
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.exporter.prometheus import PrometheusMetricReader
+from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
+from starlette.responses import Response
 
 from .app_config import fetch_sealed_secrets_cert
 from .routers import config, kubernetes, kubeseal
 
 LOGGER = logging.getLogger("kubeseal-webgui")
 
+exporter = PrometheusMetricReader()
+metrics.set_meter_provider(MeterProvider(metric_readers=[exporter]))
+meter = metrics.get_meter("kubeseal-webgui")
+
+http_requests = meter.create_counter(
+    name="kubeseal_webgui_http_requests_total",
+    description="Total HTTP requests",
+    unit="1",
+)
+http_latency = meter.create_histogram(
+    name="kubeseal_webgui_http_request_latency_seconds",
+    description="HTTP request latency in seconds",
+    unit="seconds",
+)
 
 @asynccontextmanager
 async def lifespan(fastapi_app: fastapi.FastAPI):  # noqa: ANN201 skipcq: PYL-W0613
@@ -17,7 +37,6 @@ async def lifespan(fastapi_app: fastapi.FastAPI):  # noqa: ANN201 skipcq: PYL-W0
     LOGGER.info("Startup tasks complete.")
     yield
 
-
 app = fastapi.FastAPI(lifespan=lifespan)
 
 origins = [
@@ -32,6 +51,22 @@ async def lifespan(fastapi_app: fastapi.FastAPI):  # noqa: ANN201 skipcq: PYL-W0
     allow_headers=["*"],
 )
 
+@app.middleware("http")
+async def prometheus_middleware(request: fastapi.Request, call_next):  # noqa: ANN001, ANN201
+    start_time = time.time()
+    response = await call_next(request)
+    process_time = time.time() - start_time
+
+    # Record metrics
+    http_requests.add(1, {"method": request.method, "endpoint": request.url.path, "http_status": response.status_code})
+    http_latency.record(process_time, {"method": request.method, "endpoint": request.url.path})
+
+    return response
+
+@app.get("/metrics")
+def metrics_endpoint() -> Response:
+    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)
+
 app.include_router(
     kubernetes.router,
 )
@@ -42,7 +77,6 @@ async def lifespan(fastapi_app: fastapi.FastAPI):  # noqa: ANN201 skipcq: PYL-W0
     kubeseal.router,
 )
 
-
 @app.get("/")
 def root() -> dict[str, str]:
     return {"status": "Kubeseal-WebGui API"}

api/kubeseal_webgui_api/routers/kubeseal.py

@@ -3,6 +3,7 @@
 import re
 import subprocess  # noqa: S404 the binary has to be configured by an admin
 from typing import List, Optional, Union, overload
+from opentelemetry import metrics
 
 from fastapi import APIRouter, HTTPException
 
@@ -11,21 +12,38 @@
 
 router = APIRouter()
 LOGGER = logging.getLogger("kubeseal-webgui")
-
+meter = metrics.get_meter("kubeseal-webgui")
+
+SECRETS_ENCRYPTED = meter.create_counter(
+    name="kubeseal_webgui_secrets_encrypted_total",
+    description="Total encrypted secrets",
+    unit="1",
+)
+ENCRYPTION_FAILURES = meter.create_counter(
+    name="kubeseal_webgui_encryption_failures_total",
+    description="Total failed encryption attempts",
+    unit="1",
+)
 
 @router.post("/secrets", response_model=List[KeyValuePair])
 def encrypt(data: Data) -> list[KeyValuePair]:
     try:
-        return run_kubeseal(
+        result = run_kubeseal(
             data.secrets,
             data.namespace,
             data.secret,
             data.scope or Scope.STRICT,
         )
+        SECRETS_ENCRYPTED.add(1, {"scope": data.scope or Scope.STRICT, "namespace": data.namespace})
+        return result
     except (KeyError, ValueError) as e:
-        raise HTTPException(400, f"Invalid data for sealing secrets: {e}")
-    except RuntimeError:
-        raise HTTPException(500, "Server is dreaming...")
+        ENCRYPTION_FAILURES.add(1, {"error": "invalid_input", "namespace": data.namespace})
+        LOGGER.error("Encryption failed due to invalid input: %s", e)
+        raise HTTPException(400, f"Invalid data for sealing secrets: {e}") from e
+    except RuntimeError as e:
+        ENCRYPTION_FAILURES.add(1, {"error": "runtime_error", "namespace": data.namespace})
+        LOGGER.error("Encryption failed due to runtime error: %s", e)
+        raise HTTPException(500, "Server is dreaming...") from e
 
 
 def is_blank(value: Optional[str]) -> bool: