fix: wip

Author: ServerSideHannes

pyproject.toml

@@ -22,6 +22,8 @@ dependencies = [
     "redis[hiredis]>=5.0.0",
     # Memory monitoring
     "psutil>=5.9.0",
+    # Prometheus metrics
+    "prometheus-client>=0.20.0",
 ]
 
 [project.optional-dependencies]

s3proxy/app.py

@@ -12,8 +12,10 @@
 import structlog
 from fastapi import FastAPI, HTTPException, Request, Response
 from fastapi.responses import PlainTextResponse
+from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
 from structlog.stdlib import BoundLogger
 
+from . import metrics as _  # noqa: F401 - Import to register metrics
 from .config import Settings
 from .errors import S3Error, get_s3_error_code
 from .handlers import S3ProxyHandler
@@ -151,6 +153,10 @@ def _register_routes(app: FastAPI) -> None:
     async def health():
         return PlainTextResponse("ok")
 
+    @app.get("/metrics")
+    async def metrics():
+        return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)
+
     @app.api_route(
         "/{path:path}",
         methods=["GET", "PUT", "POST", "DELETE", "HEAD"],

s3proxy/concurrency.py

@@ -12,6 +12,7 @@
 import structlog
 
 from s3proxy.errors import S3Error
+from s3proxy.metrics import MEMORY_LIMIT_BYTES, MEMORY_REJECTIONS, MEMORY_RESERVED_BYTES
 
 logger = structlog.get_logger(__name__)
 
@@ -25,6 +26,9 @@
 _active_bytes = 0
 _lock: asyncio.Lock | None = None
 
+# Initialize memory limit metric
+MEMORY_LIMIT_BYTES.set(_limit_bytes)
+
 
 def _create_malloc_release() -> Callable[[], int] | None:
     """Create platform-specific function to release memory back to OS.
@@ -91,10 +95,12 @@ async def try_acquire_memory(bytes_needed: int) -> int:
             limit_mb = _limit_bytes / 1024 / 1024
             logger.warning("MEMORY_REJECTED", active_mb=round(active_mb, 2),
                          requested_mb=round(request_mb, 2), limit_mb=round(limit_mb, 2))
+            MEMORY_REJECTIONS.inc()
             raise S3Error.slow_down(
                 f"Memory limit: {active_mb:.0f}MB + {request_mb:.0f}MB > {limit_mb:.0f}MB"
             )
         _active_bytes += to_reserve
+        MEMORY_RESERVED_BYTES.set(_active_bytes)
         return to_reserve
 
 
@@ -108,6 +114,7 @@ async def release_memory(bytes_reserved: int) -> None:
     lock = await _get_lock()
     async with lock:
         _active_bytes = max(0, _active_bytes - bytes_reserved)
+        MEMORY_RESERVED_BYTES.set(_active_bytes)
 
     # Run garbage collection and release memory to OS
     gc.collect(0)
@@ -136,6 +143,7 @@ def set_memory_limit(limit_mb: int) -> None:
     global _limit_mb, _limit_bytes
     _limit_mb = limit_mb
     _limit_bytes = limit_mb * 1024 * 1024
+    MEMORY_LIMIT_BYTES.set(_limit_bytes)
 
 
 def set_active_memory(bytes_val: int) -> None:

s3proxy/crypto.py

@@ -15,6 +15,8 @@
 )
 from structlog import BoundLogger
 
+from .metrics import BYTES_DECRYPTED, BYTES_ENCRYPTED, ENCRYPTION_OPERATIONS
+
 logger: BoundLogger = structlog.get_logger(__name__)
 
 # Constants
@@ -187,6 +189,11 @@ def encrypt(plaintext: bytes, dek: bytes, nonce: bytes | None = None) -> bytes:
         aesgcm = AESGCM(dek)
         ciphertext_with_tag = aesgcm.encrypt(nonce, plaintext, None)
         result = nonce + ciphertext_with_tag
+
+        # Track metrics
+        ENCRYPTION_OPERATIONS.labels(operation="encrypt").inc()
+        BYTES_ENCRYPTED.inc(len(plaintext))
+
         logger.debug(
             "Data encrypted",
             plaintext_size=len(plaintext),
@@ -222,6 +229,11 @@ def decrypt(ciphertext: bytes, dek: bytes) -> bytes:
     try:
         aesgcm = AESGCM(dek)
         plaintext = aesgcm.decrypt(nonce, ct_with_tag, None)
+
+        # Track metrics
+        ENCRYPTION_OPERATIONS.labels(operation="decrypt").inc()
+        BYTES_DECRYPTED.inc(len(plaintext))
+
         logger.debug(
             "Data decrypted",
             ciphertext_size=len(ciphertext),

s3proxy/metrics.py

@@ -0,0 +1,135 @@
+"""Prometheus metrics for S3Proxy."""
+
+from __future__ import annotations
+
+from prometheus_client import Counter, Gauge, Histogram
+
+# Request metrics
+REQUEST_COUNT = Counter(
+    "s3proxy_requests_total",
+    "Total number of requests",
+    ["method", "operation", "status"],
+)
+
+REQUEST_DURATION = Histogram(
+    "s3proxy_request_duration_seconds",
+    "Request duration in seconds",
+    ["method", "operation"],
+    buckets=(0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0),
+)
+
+REQUESTS_IN_FLIGHT = Gauge(
+    "s3proxy_requests_in_flight",
+    "Number of requests currently being processed",
+    ["method"],
+)
+
+# Memory/Concurrency metrics
+MEMORY_RESERVED_BYTES = Gauge(
+    "s3proxy_memory_reserved_bytes",
+    "Currently reserved memory in bytes",
+)
+
+MEMORY_LIMIT_BYTES = Gauge(
+    "s3proxy_memory_limit_bytes",
+    "Configured memory limit in bytes",
+)
+
+MEMORY_REJECTIONS = Counter(
+    "s3proxy_memory_rejections_total",
+    "Total number of requests rejected due to memory limits",
+)
+
+# Encryption metrics
+ENCRYPTION_OPERATIONS = Counter(
+    "s3proxy_encryption_operations_total",
+    "Total number of encryption/decryption operations",
+    ["operation"],
+)
+
+BYTES_ENCRYPTED = Counter(
+    "s3proxy_bytes_encrypted_total",
+    "Total bytes encrypted",
+)
+
+BYTES_DECRYPTED = Counter(
+    "s3proxy_bytes_decrypted_total",
+    "Total bytes decrypted",
+)
+
+
+def get_operation_name(method: str, path: str, query: str) -> str:
+    """Derive S3 operation name from request attributes.
+
+    Args:
+        method: HTTP method (GET, PUT, POST, DELETE, HEAD).
+        path: Request path.
+        query: Query string.
+
+    Returns:
+        S3 operation name for metrics labeling.
+    """
+    is_bucket_only = "/" not in path.strip("/") and bool(path.strip("/"))
+    is_root = path.strip("/") == ""
+
+    # Root path
+    if is_root:
+        return "ListBuckets"
+
+    # Batch delete
+    if "delete" in query and method == "POST":
+        return "DeleteObjects"
+
+    # Multipart operations
+    if "uploadId" in query:
+        if method == "GET" and "partNumber" not in query:
+            return "ListParts"
+        if method == "PUT":
+            if "x-amz-copy-source" in query:
+                return "UploadPartCopy"
+            return "UploadPart"
+        if method == "POST":
+            return "CompleteMultipartUpload"
+        if method == "DELETE":
+            return "AbortMultipartUpload"
+
+    # List/Create multipart uploads
+    if "uploads" in query:
+        if method == "GET":
+            return "ListMultipartUploads"
+        if method == "POST":
+            return "CreateMultipartUpload"
+
+    # Bucket operations
+    if is_bucket_only:
+        if "location" in query and method == "GET":
+            return "GetBucketLocation"
+        if method == "PUT":
+            return "CreateBucket"
+        if method == "DELETE":
+            return "DeleteBucket"
+        if method == "HEAD":
+            return "HeadBucket"
+        if method == "GET":
+            return "ListObjects"
+
+    # Object tagging
+    if "tagging" in query:
+        if method == "GET":
+            return "GetObjectTagging"
+        if method == "PUT":
+            return "PutObjectTagging"
+        if method == "DELETE":
+            return "DeleteObjectTagging"
+
+    # Standard object operations
+    if method == "GET":
+        return "GetObject"
+    if method == "PUT":
+        return "PutObject"
+    if method == "HEAD":
+        return "HeadObject"
+    if method == "DELETE":
+        return "DeleteObject"
+
+    return "Unknown"

s3proxy/request_handler.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+import time
 from urllib.parse import parse_qs
 
 import structlog
@@ -12,6 +13,12 @@
 from structlog.stdlib import BoundLogger
 
 from . import concurrency, crypto
+from .metrics import (
+    REQUEST_COUNT,
+    REQUEST_DURATION,
+    REQUESTS_IN_FLIGHT,
+    get_operation_name,
+)
 from .errors import S3Error, raise_for_client_error, raise_for_exception
 from .handlers import S3ProxyHandler
 from .routing import RequestDispatcher
@@ -68,22 +75,32 @@ async def handle_proxy_request(
     Raises:
         S3Error: For authentication failures or S3-compatible errors.
     """
+    # Track metrics
+    method = request.method
+    path = request.url.path
+    query = str(request.url.query)
+    operation = get_operation_name(method, path, query)
+    start_time = time.perf_counter()
+    status_code = 200
+
+    REQUESTS_IN_FLIGHT.labels(method=method).inc()
+
     # Check memory limit BEFORE reading body data - reject if at capacity
     reserved_memory = 0
-    needs_limit = request.method in ("PUT", "POST", "GET")
+    needs_limit = method in ("PUT", "POST", "GET")
     memory_limit = concurrency.get_memory_limit()
 
     if memory_limit > 0 and needs_limit:
         content_length = int(request.headers.get("content-length", "0"))
-        memory_needed = concurrency.estimate_memory_footprint(request.method, content_length)
+        memory_needed = concurrency.estimate_memory_footprint(method, content_length)
 
         logger.info(
             "REQUEST_ARRIVED - attempting to acquire memory",
             memory_needed_mb=round(memory_needed / 1024 / 1024, 2),
             active_mb=round(concurrency.get_active_memory() / 1024 / 1024, 2),
             limit_mb=round(memory_limit / 1024 / 1024, 2),
-            method=request.method,
-            path=request.url.path,
+            method=method,
+            path=path,
             content_length=content_length,
         )
         reserved_memory = await concurrency.try_acquire_memory(memory_needed)
@@ -92,22 +109,37 @@ async def handle_proxy_request(
             reserved_mb=round(reserved_memory / 1024 / 1024, 2),
             active_mb=round(concurrency.get_active_memory() / 1024 / 1024, 2),
             limit_mb=round(memory_limit / 1024 / 1024, 2),
-            method=request.method,
-            path=request.url.path,
+            method=method,
+            path=path,
         )
 
     try:
-        return await _handle_proxy_request_impl(request, handler, verifier)
+        response = await _handle_proxy_request_impl(request, handler, verifier)
+        if response is not None:
+            status_code = response.status_code
+        return response
+    except HTTPException as e:
+        status_code = e.status_code
+        raise
+    except Exception:
+        status_code = 500
+        raise
     finally:
+        # Record metrics
+        duration = time.perf_counter() - start_time
+        REQUESTS_IN_FLIGHT.labels(method=method).dec()
+        REQUEST_COUNT.labels(method=method, operation=operation, status=status_code).inc()
+        REQUEST_DURATION.labels(method=method, operation=operation).observe(duration)
+
         if reserved_memory > 0:
             await concurrency.release_memory(reserved_memory)
             logger.info(
                 "MEMORY_RELEASED",
                 released_mb=round(reserved_memory / 1024 / 1024, 2),
                 active_mb=round(concurrency.get_active_memory() / 1024 / 1024, 2),
                 limit_mb=round(memory_limit / 1024 / 1024, 2),
-                method=request.method,
-                path=request.url.path,
+                method=method,
+                path=path,
             )