refactor: event, idempotency and SSE metrics come from spans, and the last metric classes are gone

AGENTS.md

@@ -448,7 +448,7 @@ class ExecutionService:
 
 - Location: `app/services/`
 - Constructor injection: repo, producer, logger, settings, and `trace.Tracer` when the service makes spans
-- New metrics come from spans: put the facts on a span as attributes (`queue.priority`, `queue.outcome`) and define the metric under `signaltometrics` in `backend/otel-collector-config.yaml`. The classes in `app/core/metrics/` go area by area ([#389](https://github.com/HardMax71/Integr8sCode/issues/389)); don't add to them
+- New metrics come from spans: put the facts on a span as attributes (`queue.priority`, `queue.outcome`) and define the metric under `signaltometrics` in `backend/otel-collector-config.yaml`. There are no metric classes: `app/core/metrics/` only sets up the OTLP `MeterProvider` for the instrumentations ([#389](https://github.com/HardMax71/Integr8sCode/issues/389)). A gauge needs a job that reports it on a timer
 - Publish events via `UnifiedProducer`: `await self.producer.produce(event_to_produce=event, key=execution_id)`
 - Services orchestrate domain logic; no direct DB or HTTP calls
 - Always check `None` from repos and raise a domain-specific `NotFoundError` subclass (e.g. `ExecutionNotFoundError`, `SagaNotFoundError`); never return `None` to a route

backend/app/core/container.py

@@ -15,7 +15,6 @@
     KubernetesProvider,
     LoggingProvider,
     MessagingProvider,
-    MetricsProvider,
     PodMonitorProvider,
     RedisProvider,
     RepositoryProvider,
@@ -45,7 +44,6 @@ def create_app_container(settings: Settings) -> AsyncContainer:
         BrokerProvider(),
         RedisProvider(),
         CoreServicesProvider(),
-        MetricsProvider(),
         RepositoryProvider(),
         MessagingProvider(),
         DLQProvider(),
@@ -73,7 +71,6 @@ def create_worker_container(settings: Settings) -> AsyncContainer:
         BrokerProvider(),
         RedisProvider(),
         CoreServicesProvider(),
-        MetricsProvider(),
         RepositoryProvider(),
         MessagingProvider(),
         DLQProvider(),

backend/app/core/dishka_lifespan.py

@@ -17,6 +17,7 @@
 from app.events.handlers import notification_router, sse_router
 from app.services.event_replay import EventReplayService
 from app.services.notification_scheduler import NotificationScheduler
+from app.services.sse import SSERedisBus
 from app.settings import Settings
 
 
@@ -65,9 +66,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     await broker.start()
     logger.info("Kafka broker started")
 
-    # Set up APScheduler for NotificationScheduler and the replay session report
+    # Set up APScheduler for NotificationScheduler and the replay session and SSE stream reports
     notification_scheduler: NotificationScheduler = await container.get(NotificationScheduler)
     replay_service: EventReplayService = await container.get(EventReplayService)
+    sse_bus: SSERedisBus = await container.get(SSERedisBus)
     apscheduler = AsyncIOScheduler()
     apscheduler.add_job(
         notification_scheduler.process_due_notifications,
@@ -85,6 +87,14 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         max_instances=1,
         misfire_grace_time=30,
     )
+    apscheduler.add_job(
+        sse_bus.report_open_streams,
+        trigger="interval",
+        seconds=10,
+        id="report_open_sse_streams",
+        max_instances=1,
+        misfire_grace_time=30,
+    )
     apscheduler.start()
     logger.info("NotificationScheduler started (APScheduler interval=15s)")
 

backend/app/core/metrics/__init__.py

@@ -1,12 +1,5 @@
-from app.core.metrics.base import BaseMetrics, setup_metrics
-from app.core.metrics.connections import ConnectionMetrics
-from app.core.metrics.database import IdempotencyMetrics
-from app.core.metrics.events import EventMetrics
+from app.core.metrics.base import setup_metrics
 
 __all__ = [
-    "BaseMetrics",
-    "ConnectionMetrics",
-    "IdempotencyMetrics",
-    "EventMetrics",
     "setup_metrics",
 ]

backend/app/core/metrics/base.py

@@ -22,8 +22,8 @@
 def setup_metrics(settings: Settings, logger: structlog.stdlib.BoundLogger) -> None:
     """Install the global MeterProvider that exports through OTLP, and this process's CPU and memory metrics.
 
-    ``BaseMetrics`` subclasses and the OpenTelemetry instrumentations take their meters from the global
-    provider, so this runs before them. Without an OTLP endpoint (tests) the no-op provider stays.
+    The OpenTelemetry instrumentations take their meters from the global provider, so this runs before them.
+    Without an OTLP endpoint (tests) the no-op provider stays.
     """
     if not settings.OTEL_EXPORTER_OTLP_ENDPOINT:
         logger.info("Metrics OTLP export disabled", endpoint=settings.OTEL_EXPORTER_OTLP_ENDPOINT)
@@ -48,30 +48,3 @@ def setup_metrics(settings: Settings, logger: structlog.stdlib.BoundLogger) -> N
     SystemMetricsInstrumentor(config=PROCESS_METRICS).instrument(meter_provider=meter_provider)
 
     logger.info("OpenTelemetry metrics configured with OTLP exporter")
-
-
-class BaseMetrics:
-    # --8<-- [start:init]
-    def __init__(self, settings: Settings, meter_name: str | None = None):
-        """Initialize base metrics with a meter from the global MeterProvider.
-
-        The global MeterProvider is configured once by ``setup_metrics``.
-        If it hasn't been called (e.g. in tests), the default no-op provider is used.
-
-        Args:
-            settings: Application settings (kept for DI compatibility).
-            meter_name: Optional name for the meter. Defaults to class name.
-        """
-        meter_name = meter_name or self.__class__.__name__
-        self._meter = metrics.get_meter(meter_name)
-        self._create_instruments()
-
-    # --8<-- [end:init]
-
-    def _create_instruments(self) -> None:
-        """Create metric instruments. Override in subclasses."""
-        pass
-
-    def close(self) -> None:
-        """Close the metrics collector and clean up resources."""
-        pass

backend/app/core/providers.py

@@ -12,11 +12,6 @@
 from opentelemetry import trace
 
 from app.core.logging import setup_logger
-from app.core.metrics import (
-    ConnectionMetrics,
-    EventMetrics,
-    IdempotencyMetrics,
-)
 from app.core.security import SecurityService
 from app.core.tracing import Tracer
 from app.db import (
@@ -184,9 +179,9 @@ def get_unified_producer(
         self,
         event_repository: EventRepository,
         transport: KafkaEventTransport,
-        event_metrics: EventMetrics,
+        tracer: trace.Tracer,
     ) -> UnifiedProducer:
-        return UnifiedProducer(event_repository, transport, event_metrics)
+        return UnifiedProducer(event_repository, transport, tracer)
 
     @provide
     def get_idempotency_repository(self, redis_client: redis.Redis) -> RedisIdempotencyRepository:
@@ -197,9 +192,9 @@ def get_idempotency_manager(
         self,
         repo: RedisIdempotencyRepository,
         logger: structlog.stdlib.BoundLogger,
-        idempotency_metrics: IdempotencyMetrics,
+        tracer: trace.Tracer,
     ) -> IdempotencyManager:
-        return IdempotencyManager(IdempotencyConfig(), repo, logger, idempotency_metrics)
+        return IdempotencyManager(IdempotencyConfig(), repo, logger, tracer)
 
 
 class DLQProvider(Provider):
@@ -242,24 +237,6 @@ async def get_api_client(
             await api_client.close()
 
 
-class MetricsProvider(Provider):
-    """Provides all metrics instances via DI (no contextvars needed)."""
-
-    scope = Scope.APP
-
-    @provide
-    def get_event_metrics(self, settings: Settings) -> EventMetrics:
-        return EventMetrics(settings)
-
-    @provide
-    def get_connection_metrics(self, settings: Settings) -> ConnectionMetrics:
-        return ConnectionMetrics(settings)
-
-    @provide
-    def get_idempotency_metrics(self, settings: Settings) -> IdempotencyMetrics:
-        return IdempotencyMetrics(settings)
-
-
 class RepositoryProvider(Provider):
     """Provides all repository instances.
 
@@ -328,9 +305,9 @@ def get_sse_redis_bus(
         self,
         redis_client: redis.Redis,
         logger: structlog.stdlib.BoundLogger,
-        connection_metrics: ConnectionMetrics,
+        tracer: trace.Tracer,
     ) -> SSERedisBus:
-        return SSERedisBus(redis_client, logger, connection_metrics)
+        return SSERedisBus(redis_client, logger, tracer)
 
     @provide
     def get_sse_service(

backend/app/events/core/producer.py

@@ -1,17 +1,12 @@
-import time
-
 from aiokafka.errors import KafkaError
 from faststream.exceptions import IncorrectState
 from opentelemetry import trace
 from pymongo.errors import PyMongoError
 
-from app.core.metrics import EventMetrics
 from app.db import EventRepository
 from app.domain.events import DomainEvent
 from app.events.core.transport import KafkaEventTransport
 
-tracer = trace.get_tracer(__name__)
-
 PRODUCE_ERRORS: tuple[type[Exception], ...] = (PyMongoError, KafkaError, IncorrectState)
 
 
@@ -27,11 +22,11 @@ def __init__(
         self,
         event_repository: EventRepository,
         transport: KafkaEventTransport,
-        event_metrics: EventMetrics,
+        tracer: trace.Tracer,
     ):
         self._event_repository = event_repository
         self._transport = transport
-        self._metrics = event_metrics
+        self._tracer = tracer
 
     async def produce(self, event_to_produce: DomainEvent, key: str) -> None:
         """Persist event to MongoDB, then publish to Kafka.
@@ -40,16 +35,13 @@ async def produce(self, event_to_produce: DomainEvent, key: str) -> None:
         in MongoDB before the exception propagates.
         """
         event_type = event_to_produce.event_type
-        with tracer.start_as_current_span("publish_event") as span:
+        with self._tracer.start_as_current_span("publish_event") as span:
             span.set_attribute("event.type", event_type)
             if event_to_produce.aggregate_id:
                 span.set_attribute("aggregate.id", event_to_produce.aggregate_id)
-            start = time.monotonic()
             await self._event_repository.store_event(event_to_produce)
             try:
                 await self._transport.publish(event_to_produce, event_type, key)
             except Exception:
                 await self._event_repository.mark_publish_failed(event_to_produce.event_id)
                 raise
-            self._metrics.record_event_published(event_type)
-            self._metrics.record_event_processing_duration(time.monotonic() - start, event_type)