fix: data sources for grafana dashboards + correct metrics acquisition

backend/app/core/metrics/__init__.py

@@ -1,6 +1,6 @@
 from app.core.metrics.base import BaseMetrics
 from app.core.metrics.connections import ConnectionMetrics
-from app.core.metrics.database import DatabaseMetrics
+from app.core.metrics.database import IdempotencyMetrics
 from app.core.metrics.dlq import DLQMetrics
 from app.core.metrics.events import EventMetrics
 from app.core.metrics.execution import ExecutionMetrics
@@ -14,7 +14,7 @@
 __all__ = [
     "BaseMetrics",
     "ConnectionMetrics",
-    "DatabaseMetrics",
+    "IdempotencyMetrics",
     "DLQMetrics",
     "EventMetrics",
     "ExecutionMetrics",

backend/app/core/metrics/database.py

@@ -1,29 +1,10 @@
 from app.core.metrics.base import BaseMetrics
 
 
-class DatabaseMetrics(BaseMetrics):
-    """Metrics for database operations."""
+class IdempotencyMetrics(BaseMetrics):
+    """Metrics for idempotency operations."""
 
     def _create_instruments(self) -> None:
-        # MongoDB operation metrics
-        self.mongodb_event_operations = self._meter.create_counter(
-            name="mongodb.event.operations.total", description="Total MongoDB operations for events", unit="1"
-        )
-
-        self.mongodb_event_query_duration = self._meter.create_histogram(
-            name="mongodb.event.query.duration", description="Duration of MongoDB event queries in seconds", unit="s"
-        )
-
-        # Event store specific metrics
-        self.event_store_operations = self._meter.create_counter(
-            name="event.store.operations.total", description="Total event store operations", unit="1"
-        )
-
-        self.event_store_failures = self._meter.create_counter(
-            name="event.store.failures.total", description="Total event store operation failures", unit="1"
-        )
-
-        # Idempotency metrics
         self.idempotency_cache_hits = self._meter.create_counter(
             name="idempotency.cache.hits.total", description="Total idempotency cache hits", unit="1"
         )
@@ -46,37 +27,6 @@ def _create_instruments(self) -> None:
             name="idempotency.keys.active", description="Number of active idempotency keys", unit="1"
         )
 
-        # Database connection metrics
-        self.database_connections_active = self._meter.create_up_down_counter(
-            name="database.connections.active", description="Number of active database connections", unit="1"
-        )
-
-        self.database_connection_errors = self._meter.create_counter(
-            name="database.connection.errors.total", description="Total database connection errors", unit="1"
-        )
-
-    def record_mongodb_operation(self, operation: str, status: str) -> None:
-        self.mongodb_event_operations.add(1, attributes={"operation": operation, "status": status})
-
-    def record_mongodb_query_duration(self, duration_seconds: float, operation: str) -> None:
-        self.mongodb_event_query_duration.record(duration_seconds, attributes={"operation": operation})
-
-    def record_event_store_duration(self, duration_seconds: float, operation: str, collection: str) -> None:
-        self.mongodb_event_query_duration.record(
-            duration_seconds, attributes={"operation": f"store_{operation}", "collection": collection}
-        )
-
-        # Also record in event store specific counter
-        self.event_store_operations.add(1, attributes={"operation": operation, "collection": collection})
-
-    def record_event_query_duration(self, duration_seconds: float, operation: str, collection: str) -> None:
-        self.mongodb_event_query_duration.record(
-            duration_seconds, attributes={"operation": f"query_{operation}", "collection": collection}
-        )
-
-    def record_event_store_failed(self, event_type: str, error_type: str) -> None:
-        self.event_store_failures.add(1, attributes={"event_type": event_type, "error_type": error_type})
-
     def record_idempotency_cache_hit(self, event_type: str, operation: str) -> None:
         self.idempotency_cache_hits.add(1, attributes={"event_type": event_type, "operation": operation})
 
@@ -97,16 +47,5 @@ def decrement_idempotency_keys(self, prefix: str) -> None:
         """Decrement active idempotency keys count when a key is removed."""
         self.idempotency_keys_active.add(-1, attributes={"key_prefix": prefix})
 
-    def record_idempotent_event_processed(self, event_type: str, result: str) -> None:
-        self.event_store_operations.add(
-            1, attributes={"operation": "idempotent_process", "event_type": event_type, "result": result}
-        )
-
     def record_idempotent_processing_duration(self, duration_seconds: float, event_type: str) -> None:
         self.idempotency_processing_duration.record(duration_seconds, attributes={"event_type": event_type})
-
-    def update_database_connections(self, delta: int) -> None:
-        self.database_connections_active.add(delta)
-
-    def record_database_connection_error(self, error_type: str) -> None:
-        self.database_connection_errors.add(1, attributes={"error_type": error_type})

backend/app/core/providers.py

@@ -13,10 +13,10 @@
 from app.core.logging import setup_logger
 from app.core.metrics import (
     ConnectionMetrics,
-    DatabaseMetrics,
     DLQMetrics,
     EventMetrics,
     ExecutionMetrics,
+    IdempotencyMetrics,
     KubernetesMetrics,
     NotificationMetrics,
     QueueMetrics,
@@ -184,9 +184,9 @@ def get_idempotency_manager(
             self,
             repo: RedisIdempotencyRepository,
             logger: structlog.stdlib.BoundLogger,
-            database_metrics: DatabaseMetrics,
+            idempotency_metrics: IdempotencyMetrics,
     ) -> IdempotencyManager:
-        return IdempotencyManager(IdempotencyConfig(), repo, logger, database_metrics)
+        return IdempotencyManager(IdempotencyConfig(), repo, logger, idempotency_metrics)
 
 
 class DLQProvider(Provider):
@@ -251,8 +251,8 @@ def get_execution_metrics(self, settings: Settings) -> ExecutionMetrics:
         return ExecutionMetrics(settings)
 
     @provide
-    def get_database_metrics(self, settings: Settings) -> DatabaseMetrics:
-        return DatabaseMetrics(settings)
+    def get_idempotency_metrics(self, settings: Settings) -> IdempotencyMetrics:
+        return IdempotencyMetrics(settings)
 
     @provide
     def get_kubernetes_metrics(self, settings: Settings) -> KubernetesMetrics:

backend/app/dlq/manager.py

@@ -1,3 +1,4 @@
+import time
 from datetime import datetime, timezone
 from typing import Callable
 
@@ -78,6 +79,7 @@ async def process_monitoring_cycle(self) -> None:
 
     async def handle_message(self, message: DLQMessage) -> None:
         """Process a single DLQ message: filter -> store -> decide retry/discard."""
+        start = time.monotonic()
         for filter_func in self._filters:
             if not filter_func(message):
                 self.logger.info("Message filtered out", event_id=message.event.event_id)
@@ -86,6 +88,9 @@ async def handle_message(self, message: DLQMessage) -> None:
         message.status = DLQMessageStatus.PENDING
         message.last_updated = datetime.now(timezone.utc)
         await self.repository.save_message(message)
+        self.metrics.record_dlq_message_received(message.original_topic, message.event.event_type)
+        age_seconds = (datetime.now(timezone.utc) - message.failed_at).total_seconds()
+        self.metrics.record_dlq_message_age(age_seconds)
 
         await self._broker.publish(
             DLQMessageReceivedEvent(
@@ -120,11 +125,14 @@ async def handle_message(self, message: DLQMessage) -> None:
         if retry_policy.strategy == RetryStrategy.IMMEDIATE:
             await self.retry_message(message)
 
+        self.metrics.record_dlq_processing_duration(time.monotonic() - start, "handle")
+
     async def retry_message(self, message: DLQMessage) -> None:
         """Retry a DLQ message by republishing to the original topic.
 
         FastStream handles JSON serialization of Pydantic models natively.
         """
+        start = time.monotonic()
         await self._broker.publish(
             message=message.event,
             topic=message.original_topic,
@@ -158,10 +166,12 @@ async def retry_message(self, message: DLQMessage) -> None:
             ),
             topic=EventType.DLQ_MESSAGE_RETRIED,
         )
+        self.metrics.record_dlq_processing_duration(time.monotonic() - start, "retry")
         self.logger.info("Successfully retried message", event_id=message.event.event_id)
 
     async def discard_message(self, message: DLQMessage, reason: str) -> None:
         """Discard a DLQ message, updating status and emitting an event."""
+        start = time.monotonic()
         self.metrics.record_dlq_message_discarded(message.original_topic, message.event.event_type, reason)
 
         await self.repository.update_status(
@@ -188,6 +198,7 @@ async def discard_message(self, message: DLQMessage, reason: str) -> None:
             ),
             topic=EventType.DLQ_MESSAGE_DISCARDED,
         )
+        self.metrics.record_dlq_processing_duration(time.monotonic() - start, "discard")
         self.logger.warning("Discarded message", event_id=message.event.event_id, reason=reason)
 
     async def process_due_retries(self) -> int: