feat: add detailed metrics, active paste count tracking, and Redis integration

- Implemented metrics for paste, cache, storage, and cleanup operations
- Introduced `ActivePastesCounter` for accurate active paste count tracking
- Enable periodic refresh of `active_pastes` gauge from the database
- Added Redis-backed counters and gauges for multi-instance support
- Enhanced cleanup service to update metrics and active pastes count
- Integrated metrics initialization and Redis client setup in app lifecycle
- Updated API routes to collect cache operation metrics
- Updated `PasteService` to increment/decrement counters during operations

Author: Pdzly

backend/app/api/subroutes/pastes.py

@@ -22,6 +22,7 @@
 from app.ratelimit import create_limit_resolver, get_exempt_key, limiter
 from app.services.paste_service import PasteService
 from app.utils.LRUMemoryCache import LRUMemoryCache
+from app.utils.metrics import cache_operations
 
 if TYPE_CHECKING:
     from aiocache import RedisCache
@@ -58,6 +59,7 @@ async def get_paste(
 ):
     cached_result = await cache.get(paste_id)
     if cached_result:
+        cache_operations.labels(operation="get", result="hit").inc()
         return Response(
             cached_result,
             headers={
@@ -66,6 +68,7 @@ async def get_paste(
             },
         )
 
+    cache_operations.labels(operation="get", result="miss").inc()
     paste_result = await paste_service.get_legacy_paste_by_name(paste_id)
     if not paste_result:
         return Response(
@@ -82,6 +85,7 @@ async def get_paste(
     paste_result = paste_result.model_dump_json()
 
     await cache.set(paste_id, paste_result, ttl=config.CACHE_TTL)
+    cache_operations.labels(operation="set", result="success").inc()
 
     return Response(
         paste_result,
@@ -105,6 +109,7 @@ async def get_paste(
 ):
     cached_result = await cache.get(paste_id)
     if cached_result:
+        cache_operations.labels(operation="get", result="hit").inc()
         return Response(
             cached_result,
             headers={
@@ -113,6 +118,7 @@ async def get_paste(
             },
         )
 
+    cache_operations.labels(operation="get", result="miss").inc()
     paste_result = await paste_service.get_paste_by_id(paste_id)
     if not paste_result:
         return Response(
@@ -128,6 +134,7 @@ async def get_paste(
     paste_result = paste_result.model_dump_json()
 
     await cache.set(paste_id, paste_result, ttl=config.CACHE_TTL)
+    cache_operations.labels(operation="set", result="success").inc()
 
     return Response(
         paste_result,
@@ -170,6 +177,7 @@ async def edit_paste(
         )
     # Invalidate cache after successful edit
     await cache.delete(paste_id)
+    cache_operations.labels(operation="delete", result="success").inc()
     return result
 
 
@@ -193,4 +201,5 @@ async def delete_paste(
         )
     # Invalidate cache after successful delete
     await cache.delete(paste_id)
+    cache_operations.labels(operation="delete", result="success").inc()
     return {"message": "Paste deleted successfully"}

backend/app/services/cleanup_service.py

@@ -1,6 +1,7 @@
 import asyncio
 import contextlib
 import logging
+import time
 from collections.abc import Coroutine
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -12,6 +13,8 @@
 from app.config import config
 from app.db.models import PasteEntity
 from app.locks import DistributedLock
+from app.utils.active_pastes_counter import get_active_pastes_counter
+from app.utils.metrics import cleanup_duration
 
 
 class CleanupService:
@@ -76,6 +79,7 @@ async def _cleanup_expired_pastes(self):
         """Remove expired pastes and their files"""
         from app.api.subroutes.pastes import cache
 
+        start_time = time.monotonic()
         try:
             BATCH_SIZE = 100
             total_cleaned = 0
@@ -123,12 +127,19 @@ async def _cleanup_expired_pastes(self):
                     total_cleaned += len(batch)
 
                 if total_cleaned > 0:
+                    # Update active_pastes counter
+                    counter = get_active_pastes_counter()
+                    if counter:
+                        counter.dec(total_cleaned)
                     if not error:
                         self.logger.info("Successfully cleaned up %d expired pastes", total_cleaned)
                     else:
                         self.logger.info("Cleaned up %d expired pastes with some errors", total_cleaned)
         except Exception as exc:
             self.logger.error("Failed to cleanup expired pastes: %s", exc)
+        finally:
+            duration = time.monotonic() - start_time
+            cleanup_duration.observe(duration)
 
     async def _cleanup_deleted_pastes(self):
         """Remove deleted pastes that have been marked for deletion beyond the configured time"""
@@ -139,6 +150,7 @@ async def _cleanup_deleted_pastes(self):
         self.logger.info("Cleaning up deleted pastes")
         from app.api.subroutes.pastes import cache
 
+        start_time = time.monotonic()
         try:
             BATCH_SIZE = 100
             total_cleaned = 0
@@ -195,3 +207,6 @@ async def _cleanup_deleted_pastes(self):
                         self.logger.info("Cleaned up %d deleted pastes with some errors", total_cleaned)
         except Exception as exc:
             self.logger.error("Failed to cleanup deleted pastes: %s", exc)
+        finally:
+            duration = time.monotonic() - start_time
+            cleanup_duration.observe(duration)

backend/app/services/paste_service.py

@@ -28,6 +28,13 @@
 from app.db.models import PasteEntity
 from app.services.cleanup_service import CleanupService
 from app.storage import StorageClient
+from app.utils.active_pastes_counter import get_active_pastes_counter
+from app.utils.metrics import (
+    compressed_pastes,
+    paste_operations,
+    paste_size,
+    storage_operations,
+)
 from app.utils.token_utils import hash_token, is_token_hashed, verify_token
 
 
@@ -44,6 +51,18 @@ def __init__(
         self._cleanup_task: asyncio.Task[Coroutine[None, None, None]] | None = None
         self._lock_file: Path = Path(".cleanup.lock")
         self._cleanup_service: CleanupService = cleanup_service
+        self._storage_backend_name: str = self._get_storage_backend_name()
+
+    def _get_storage_backend_name(self) -> str:
+        """Get storage backend name for metrics labels."""
+        class_name = self.storage_client.__class__.__name__
+        if "Local" in class_name:
+            return "local"
+        elif "S3" in class_name:
+            return "s3"
+        elif "Minio" in class_name or "MinIO" in class_name:
+            return "minio"
+        return "unknown"
 
     async def _read_content(self, paste_path: str, is_compressed: bool = False) -> str | None:
         """
@@ -60,8 +79,15 @@ async def _read_content(self, paste_path: str, is_compressed: bool = False) -> s
             data = await self.storage_client.get_object(paste_path)
             if data is None:
                 self.logger.error("Paste content not found: %s", paste_path)
+                storage_operations.labels(
+                    operation="get", backend=self._storage_backend_name, status="not_found"
+                ).inc()
                 return None
 
+            storage_operations.labels(
+                operation="get", backend=self._storage_backend_name, status="success"
+            ).inc()
+
             if is_compressed:
                 from app.utils.compression import CompressionError, decompress_content
 
@@ -74,6 +100,9 @@ async def _read_content(self, paste_path: str, is_compressed: bool = False) -> s
                 return data.decode("utf-8")
         except Exception as exc:
             self.logger.error("Failed to read paste content: %s", exc)
+            storage_operations.labels(
+                operation="get", backend=self._storage_backend_name, status="error"
+            ).inc()
             return None
 
     async def _save_content(self, paste_id: str, content: str) -> tuple[str, int, bool, int | None] | None:
@@ -126,23 +155,38 @@ async def _save_content(self, paste_id: str, content: str) -> tuple[str, int, bo
             # Write content (compressed or uncompressed)
             if use_compression and compressed_data:
                 await self.storage_client.put_object(storage_key, compressed_data)
+                storage_operations.labels(
+                    operation="put", backend=self._storage_backend_name, status="success"
+                ).inc()
                 content_size = len(compressed_data)
                 return storage_key, content_size, True, original_size
             else:
                 await self.storage_client.put_object(storage_key, content.encode("utf-8"))
+                storage_operations.labels(
+                    operation="put", backend=self._storage_backend_name, status="success"
+                ).inc()
                 content_size = original_size
                 return storage_key, content_size, False, None
 
         except Exception as exc:
             self.logger.error("Failed to save paste content: %s", exc)
+            storage_operations.labels(
+                operation="put", backend=self._storage_backend_name, status="error"
+            ).inc()
             return None
 
     async def _remove_file(self, storage_key: str):
         """Remove paste file from storage."""
         try:
             await self.storage_client.delete_object(storage_key)
+            storage_operations.labels(
+                operation="delete", backend=self._storage_backend_name, status="success"
+            ).inc()
         except Exception as exc:
             self.logger.error("Failed to remove file %s: %s", storage_key, exc)
+            storage_operations.labels(
+                operation="delete", backend=self._storage_backend_name, status="error"
+            ).inc()
 
     def verify_storage_limit(self):
         """Verify storage limit (only applicable for local storage)."""
@@ -202,11 +246,13 @@ async def get_paste_by_id(self, paste_id: UUID4) -> PasteResponse | None:
             )
             result: PasteEntity | None = (await session.execute(stmt)).scalar_one_or_none()
             if result is None:
+                paste_operations.labels(operation="get", status="not_found").inc()
                 return None
             content = await self._read_content(
                 result.content_path,
                 is_compressed=result.is_compressed,
             )
+            paste_operations.labels(operation="get", status="success").inc()
             return PasteResponse(
                 id=result.id,
                 title=result.title,
@@ -233,6 +279,7 @@ async def edit_paste(self, paste_id: UUID4, edit_paste: EditPaste, edit_token: s
             result: PasteEntity | None = (await session.execute(stmt)).scalar_one_or_none()
 
             if result is None:
+                paste_operations.labels(operation="edit", status="not_found").inc()
                 return None
 
             # Verify token - support both hashed (new) and plaintext (legacy)
@@ -249,6 +296,7 @@ async def edit_paste(self, paste_id: UUID4, edit_paste: EditPaste, edit_token: s
                     self.logger.info("Upgraded edit token to hashed format for paste %s", paste_id)
 
             if not token_valid:
+                paste_operations.labels(operation="edit", status="unauthorized").inc()
                 return None
 
             # Update only the fields that are provided (not None)
@@ -290,6 +338,7 @@ async def edit_paste(self, paste_id: UUID4, edit_paste: EditPaste, edit_token: s
                 )
             )
 
+            paste_operations.labels(operation="edit", status="success").inc()
             return PasteResponse(
                 id=result.id,
                 title=result.title,
@@ -316,6 +365,7 @@ async def delete_paste(self, paste_id: UUID4, delete_token: str) -> bool:
             result: PasteEntity | None = (await session.execute(stmt)).scalar_one_or_none()
 
             if result is None:
+                paste_operations.labels(operation="delete", status="not_found").inc()
                 return False
 
             # Verify token - support both hashed (new) and plaintext (legacy)
@@ -329,6 +379,7 @@ async def delete_paste(self, paste_id: UUID4, delete_token: str) -> bool:
                 # No need to upgrade here since we're deleting anyway
 
             if not token_valid:
+                paste_operations.labels(operation="delete", status="unauthorized").inc()
                 return False
 
             # Remove file
@@ -340,10 +391,15 @@ async def delete_paste(self, paste_id: UUID4, delete_token: str) -> bool:
             # Delete from database
             await session.delete(result)
             await session.commit()
+            paste_operations.labels(operation="delete", status="success").inc()
+            counter = get_active_pastes_counter()
+            if counter:
+                counter.dec()
             return True
 
     async def create_paste(self, paste: CreatePaste, user_data: UserMetaData) -> PasteResponse:
         if not self.verify_storage_limit():
+            paste_operations.labels(operation="create", status="storage_limit").inc()
             raise HTTPException(
                 status_code=500,
                 detail="Storage limit reached, contact administration",
@@ -355,6 +411,7 @@ async def create_paste(self, paste: CreatePaste, user_data: UserMetaData) -> Pas
             paste.content,
         )
         if not save_result:
+            paste_operations.labels(operation="create", status="error").inc()
             raise HTTPException(
                 status_code=500,
                 detail="Failed to save paste content",
@@ -391,6 +448,15 @@ async def create_paste(self, paste: CreatePaste, user_data: UserMetaData) -> Pas
                 await session.commit()
                 await session.refresh(entity)
 
+                # Record metrics on successful create
+                paste_operations.labels(operation="create", status="success").inc()
+                paste_size.observe(original_size if original_size else content_size)
+                counter = get_active_pastes_counter()
+                if counter:
+                    counter.inc()
+                if is_compressed:
+                    compressed_pastes.inc()
+
                 return CreatePasteResponse(
                     id=entity.id,
                     title=entity.title,
@@ -405,6 +471,7 @@ async def create_paste(self, paste: CreatePaste, user_data: UserMetaData) -> Pas
         except Exception as exc:
             self.logger.error("Failed to create paste: %s", exc)
             await self._remove_file(paste_path)
+            paste_operations.labels(operation="create", status="error").inc()
             raise HTTPException(
                 status_code=500,
                 detail="Failed to create paste",