diff --git a/AGENTS.md b/AGENTS.md index 2bb8292f..e36b5a08 100644 --- a/AGENTS.md +++ b/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 diff --git a/backend/app/core/container.py b/backend/app/core/container.py index 05ee7852..2f5ff6b4 100644 --- a/backend/app/core/container.py +++ b/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(), diff --git a/backend/app/core/dishka_lifespan.py b/backend/app/core/dishka_lifespan.py index c71836c0..63d51e36 100644 --- a/backend/app/core/dishka_lifespan.py +++ b/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)") diff --git a/backend/app/core/metrics/__init__.py b/backend/app/core/metrics/__init__.py index cbd09148..7ceac459 100644 --- a/backend/app/core/metrics/__init__.py +++ b/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", ] diff --git a/backend/app/core/metrics/base.py b/backend/app/core/metrics/base.py index 7d8a88dc..da8748e0 100644 --- a/backend/app/core/metrics/base.py +++ b/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 diff --git a/backend/app/core/metrics/connections.py b/backend/app/core/metrics/connections.py deleted file mode 100644 index 084577dc..00000000 --- a/backend/app/core/metrics/connections.py +++ /dev/null @@ -1,30 +0,0 @@ -from app.core.metrics.base import BaseMetrics - - -class ConnectionMetrics(BaseMetrics): - """Metrics for SSE connections.""" - - def _create_instruments(self) -> None: - self.sse_active_connections = self._meter.create_up_down_counter( - name="sse.connections.active", description="Number of active Server-Sent Events connections", unit="1" - ) - - self.sse_messages_sent = self._meter.create_counter( - name="sse.messages.sent.total", description="Total number of SSE messages sent", unit="1" - ) - - self.sse_connection_duration = self._meter.create_histogram( - name="sse.connection.duration", description="Duration of SSE connections in seconds", unit="s" - ) - - def increment_sse_connections(self, endpoint: str = "default") -> None: - self.sse_active_connections.add(1, attributes={"endpoint": endpoint}) - - def decrement_sse_connections(self, endpoint: str = "default") -> None: - self.sse_active_connections.add(-1, attributes={"endpoint": endpoint}) - - def record_sse_message_sent(self, endpoint: str, event_type: str) -> None: - self.sse_messages_sent.add(1, attributes={"endpoint": endpoint, "event_type": event_type}) - - def record_sse_connection_duration(self, duration_seconds: float, endpoint: str) -> None: - self.sse_connection_duration.record(duration_seconds, attributes={"endpoint": endpoint}) diff --git a/backend/app/core/metrics/database.py b/backend/app/core/metrics/database.py deleted file mode 100644 index 8d7c969b..00000000 --- a/backend/app/core/metrics/database.py +++ /dev/null @@ -1,79 +0,0 @@ -import redis -from opentelemetry.metrics import CallbackOptions, Observation - -from app.core.metrics.base import BaseMetrics -from app.settings import Settings - -_KEY_PATTERN = "idempotency:*" - - -class IdempotencyMetrics(BaseMetrics): - """Metrics for idempotency operations.""" - - def __init__(self, settings: Settings, meter_name: str | None = None) -> None: - self._redis = redis.Redis( - host=settings.REDIS_HOST, - port=settings.REDIS_PORT, - db=settings.REDIS_DB, - password=settings.REDIS_PASSWORD, - ssl=settings.REDIS_SSL, - socket_connect_timeout=5, - socket_timeout=5, - ) - super().__init__(settings, meter_name) - - def _create_instruments(self) -> None: - self.idempotency_cache_hits = self._meter.create_counter( - name="idempotency.cache.hits.total", description="Total idempotency cache hits", unit="1" - ) - - self.idempotency_cache_misses = self._meter.create_counter( - name="idempotency.cache.misses.total", description="Total idempotency cache misses", unit="1" - ) - - self.idempotency_duplicates_blocked = self._meter.create_counter( - name="idempotency.duplicates.blocked.total", - description="Total duplicate operations blocked by idempotency", - unit="1", - ) - - self.idempotency_check_duration = self._meter.create_histogram( - name="idempotency.check.duration", description="Duration of idempotency key checks in seconds", unit="s" - ) - - self.idempotency_processing_duration = self._meter.create_histogram( - name="idempotency.processing.duration", - description="Time from reserving an idempotency key to marking it completed or failed, in seconds", - unit="s", - ) - - self._meter.create_observable_gauge( - name="idempotency.keys.active", - callbacks=[self._observe_active_keys], - description="Idempotency keys currently stored in Redis", - unit="{key}", - ) - - def _observe_active_keys(self, _: CallbackOptions) -> list[Observation]: - try: - count = sum(1 for _ in self._redis.scan_iter(match=_KEY_PATTERN, count=1000)) - except redis.RedisError: - return [] - return [Observation(count)] - - 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}) - - def record_idempotency_cache_miss(self, event_type: str, operation: str) -> None: - self.idempotency_cache_misses.add(1, attributes={"event_type": event_type, "operation": operation}) - - def record_idempotency_duplicate_blocked(self, event_type: str) -> None: - self.idempotency_duplicates_blocked.add(1, attributes={"event_type": event_type}) - - def record_idempotency_check_duration(self, duration_seconds: float, operation: str) -> None: - self.idempotency_check_duration.record(duration_seconds, attributes={"operation": operation}) - - def record_idempotent_processing_duration(self, duration_seconds: float, event_type: str, status: str) -> None: - self.idempotency_processing_duration.record( - duration_seconds, attributes={"event_type": event_type, "status": status} - ) diff --git a/backend/app/core/metrics/events.py b/backend/app/core/metrics/events.py deleted file mode 100644 index 480e26ae..00000000 --- a/backend/app/core/metrics/events.py +++ /dev/null @@ -1,27 +0,0 @@ -from app.core.metrics.base import BaseMetrics - - -class EventMetrics(BaseMetrics): - """Metrics for domain-level event publishing. - - Kafka publish and consume metrics, consume failures included, come from - KafkaTelemetryMiddleware on the broker. - """ - - def _create_instruments(self) -> None: - self.event_published = self._meter.create_counter( - name="events.published.total", description="Total number of events published", unit="1" - ) - - self.event_processing_duration = self._meter.create_histogram( - name="event.processing.duration", description="Time spent processing events in seconds", unit="s" - ) - - def record_event_published(self, event_type: str, event_category: str | None = None) -> None: - if event_category is None: - event_category = event_type.split(".")[0] if "." in event_type else event_type - - self.event_published.add(1, attributes={"event_type": event_type, "event_category": event_category}) - - def record_event_processing_duration(self, duration_seconds: float, event_type: str) -> None: - self.event_processing_duration.record(duration_seconds, attributes={"event_type": event_type}) diff --git a/backend/app/core/providers.py b/backend/app/core/providers.py index 405fe406..5b5554c6 100644 --- a/backend/app/core/providers.py +++ b/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( diff --git a/backend/app/events/core/producer.py b/backend/app/events/core/producer.py index e634d06d..96861b44 100644 --- a/backend/app/events/core/producer.py +++ b/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) diff --git a/backend/app/services/idempotency/idempotency_manager.py b/backend/app/services/idempotency/idempotency_manager.py index 717eadee..5142f44f 100644 --- a/backend/app/services/idempotency/idempotency_manager.py +++ b/backend/app/services/idempotency/idempotency_manager.py @@ -1,14 +1,14 @@ import hashlib import json -import time from datetime import datetime, timedelta, timezone import structlog +from opentelemetry import trace +from opentelemetry.util.types import AttributeValue from pydantic import BaseModel from pymongo.errors import DuplicateKeyError from redis import RedisError -from app.core.metrics import IdempotencyMetrics from app.domain.enums import EventType from app.domain.events import BaseEvent from app.domain.idempotency import IdempotencyRecord, IdempotencyStatus, KeyStrategy @@ -44,14 +44,19 @@ def __init__( config: IdempotencyConfig, repository: RedisIdempotencyRepository, logger: structlog.stdlib.BoundLogger, - idempotency_metrics: IdempotencyMetrics, + tracer: trace.Tracer, ) -> None: self.config = config - self.metrics = idempotency_metrics + self._tracer = tracer self._repo = repository self.logger = logger self.logger.info("Idempotency manager initialized") + async def report_active_keys(self) -> None: + """Report how many idempotency keys Redis holds, for the collector's gauge. The worker runs it every 30 s.""" + count = await self._repo.count_keys() + self._tracer.start_span("idempotency.keys", attributes={"idempotency.active_keys": count}).end() + # --8<-- [start:generate_key] def _generate_key( self, event: BaseEvent, key_strategy: KeyStrategy, custom_key: str | None = None, fields: set[str] | None = None @@ -83,21 +88,17 @@ async def check_and_reserve( ttl_seconds: int | None = None, fields: set[str] | None = None, ) -> IdempotencyResult: - start = time.monotonic() full_key = self._generate_key(event, key_strategy, custom_key, fields) ttl = ttl_seconds or self.config.default_ttl_seconds - existing = await self._repo.find_by_key(full_key) - if existing: - self.metrics.record_idempotency_cache_hit(event.event_type, "check_and_reserve") - result = await self._handle_existing_key(existing, full_key, event.event_type) - self.metrics.record_idempotency_check_duration(time.monotonic() - start, "check_and_reserve") - return result - - self.metrics.record_idempotency_cache_miss(event.event_type, "check_and_reserve") - result = await self._create_new_key(full_key, event, ttl) - self.metrics.record_idempotency_check_duration(time.monotonic() - start, "check_and_reserve") - return result + attributes = {"idempotency.event_type": event.event_type, "idempotency.operation": "check_and_reserve"} + with self._tracer.start_as_current_span("idempotency.check", attributes=attributes) as span: + existing = await self._repo.find_by_key(full_key) + if existing: + span.set_attribute("idempotency.result", "hit") + return await self._handle_existing_key(existing, full_key, event.event_type) + span.set_attribute("idempotency.result", "miss") + return await self._create_new_key(full_key, event, ttl) async def _handle_existing_key( self, @@ -109,7 +110,9 @@ async def _handle_existing_key( if status == IdempotencyStatus.PROCESSING: return await self._handle_processing_key(existing, full_key, event_type) - self.metrics.record_idempotency_duplicate_blocked(event_type) + self._tracer.start_span( + "idempotency.duplicate_blocked", attributes={"idempotency.event_type": event_type} + ).end() created_at = existing.created_at or datetime.now(timezone.utc) return IdempotencyResult( is_duplicate=True, @@ -140,7 +143,9 @@ async def _handle_processing_key( is_duplicate=False, status=IdempotencyStatus.PROCESSING, created_at=now, key=full_key ) - self.metrics.record_idempotency_duplicate_blocked(event_type) + self._tracer.start_span( + "idempotency.duplicate_blocked", attributes={"idempotency.event_type": event_type} + ).end() return IdempotencyResult( is_duplicate=True, status=IdempotencyStatus.PROCESSING, @@ -194,7 +199,12 @@ async def _update_key_status( existing.result_json = cached_json else: self.logger.warning("Result too large to cache", key=full_key) - self.metrics.record_idempotent_processing_duration(duration_seconds, existing.event_type, status) + attributes: dict[str, AttributeValue] = { + "idempotency.event_type": existing.event_type, + "idempotency.status": status, + "idempotency.processing_seconds": duration_seconds, + } + self._tracer.start_span("idempotency.processed", attributes=attributes).end() return (await self._repo.update_record(existing)) > 0 async def mark_completed( diff --git a/backend/app/services/idempotency/redis_repository.py b/backend/app/services/idempotency/redis_repository.py index a16344a7..30519a52 100644 --- a/backend/app/services/idempotency/redis_repository.py +++ b/backend/app/services/idempotency/redis_repository.py @@ -123,3 +123,7 @@ async def update_record(self, record: IdempotencyRecord) -> int: async def delete_key(self, key: str) -> int: return int(await self._r.delete(self._full_key(key))) + + async def count_keys(self) -> int: + """Count the idempotency keys Redis holds now; SCAN walks the keyspace without blocking Redis.""" + return len([key async for key in self._r.scan_iter(match=f"{self._prefix}:*", count=1000)]) diff --git a/backend/app/services/sse/redis_bus.py b/backend/app/services/sse/redis_bus.py index 8b50a3d5..4dfda6dd 100644 --- a/backend/app/services/sse/redis_bus.py +++ b/backend/app/services/sse/redis_bus.py @@ -1,13 +1,15 @@ from __future__ import annotations +from collections import Counter from collections.abc import AsyncGenerator from datetime import datetime, timezone import redis.asyncio as redis import structlog +from opentelemetry import trace +from opentelemetry.util.types import AttributeValue from pydantic import TypeAdapter -from app.core.metrics import ConnectionMetrics from app.domain.sse import DomainNotificationSSEPayload, DomainReplaySSEPayload, SSEExecutionEventData _sse_event_adapter = TypeAdapter(SSEExecutionEventData) @@ -22,18 +24,37 @@ def __init__( self, redis_client: redis.Redis, logger: structlog.stdlib.BoundLogger, - connection_metrics: ConnectionMetrics, + tracer: trace.Tracer, exec_prefix: str = "sse:exec:", notif_prefix: str = "sse:notif:", replay_prefix: str = "sse:replay:", ) -> None: self._redis = redis_client self.logger = logger - self._metrics = connection_metrics + self._tracer = tracer + self._open_streams: Counter[str] = Counter() self._exec_prefix = exec_prefix self._notif_prefix = notif_prefix self._replay_prefix = replay_prefix + async def report_open_streams(self) -> None: + """Report this process's open SSE streams per endpoint, zeros included. The API runs it every 10 s.""" + for endpoint in ("executions", "notifications", "replay"): + attributes: dict[str, AttributeValue] = { + "sse.endpoint": endpoint, + "sse.open_streams": self._open_streams[endpoint], + } + self._tracer.start_span("sse.streams", attributes=attributes).end() + + def _report_message(self, endpoint: str, event_type: str) -> None: + self._tracer.start_span( + "sse.message", attributes={"sse.endpoint": endpoint, "sse.event_type": event_type} + ).end() + + def _report_stream_closed(self, endpoint: str, seconds: float) -> None: + attributes: dict[str, AttributeValue] = {"sse.endpoint": endpoint, "sse.duration_seconds": seconds} + self._tracer.start_span("sse.stream_closed", attributes=attributes).end() + def _exec_channel(self, execution_id: str) -> str: return f"{self._exec_prefix}{execution_id}" @@ -48,36 +69,36 @@ async def publish_notification(self, user_id: str, notification: DomainNotificat async def listen_execution(self, execution_id: str) -> AsyncGenerator[SSEExecutionEventData, None]: start = datetime.now(timezone.utc) - self._metrics.increment_sse_connections("executions") + self._open_streams["executions"] += 1 self.logger.info("SSE execution stream opened", execution_id=execution_id) try: async with self._redis.pubsub(ignore_subscribe_messages=True) as pubsub: await pubsub.subscribe(self._exec_channel(execution_id)) async for message in pubsub.listen(): event = _sse_event_adapter.validate_json(message["data"]) - self._metrics.record_sse_message_sent("executions", event.event_type) + self._report_message("executions", event.event_type) yield event finally: duration = (datetime.now(timezone.utc) - start).total_seconds() - self._metrics.record_sse_connection_duration(duration, "executions") - self._metrics.decrement_sse_connections("executions") + self._open_streams["executions"] -= 1 + self._report_stream_closed("executions", duration) self.logger.info("SSE execution stream closed", execution_id=execution_id) async def listen_notifications(self, user_id: str) -> AsyncGenerator[DomainNotificationSSEPayload, None]: start = datetime.now(timezone.utc) - self._metrics.increment_sse_connections("notifications") + self._open_streams["notifications"] += 1 self.logger.info("SSE notification stream opened", user_id=user_id) try: async with self._redis.pubsub(ignore_subscribe_messages=True) as pubsub: await pubsub.subscribe(self._notif_channel(user_id)) async for message in pubsub.listen(): notification = _notif_payload_adapter.validate_json(message["data"]) - self._metrics.record_sse_message_sent("notifications", "notification") + self._report_message("notifications", "notification") yield notification finally: duration = (datetime.now(timezone.utc) - start).total_seconds() - self._metrics.record_sse_connection_duration(duration, "notifications") - self._metrics.decrement_sse_connections("notifications") + self._open_streams["notifications"] -= 1 + self._report_stream_closed("notifications", duration) self.logger.info("SSE notification stream closed", user_id=user_id) def _replay_channel(self, session_id: str) -> str: @@ -88,17 +109,17 @@ async def publish_replay_status(self, session_id: str, status: DomainReplaySSEPa async def listen_replay(self, session_id: str) -> AsyncGenerator[DomainReplaySSEPayload, None]: start = datetime.now(timezone.utc) - self._metrics.increment_sse_connections("replay") + self._open_streams["replay"] += 1 self.logger.info("SSE replay stream opened", session_id=session_id) try: async with self._redis.pubsub(ignore_subscribe_messages=True) as pubsub: await pubsub.subscribe(self._replay_channel(session_id)) async for message in pubsub.listen(): status = _replay_adapter.validate_json(message["data"]) - self._metrics.record_sse_message_sent("replay", "replay_status") + self._report_message("replay", "replay_status") yield status finally: duration = (datetime.now(timezone.utc) - start).total_seconds() - self._metrics.record_sse_connection_duration(duration, "replay") - self._metrics.decrement_sse_connections("replay") + self._open_streams["replay"] -= 1 + self._report_stream_closed("replay", duration) self.logger.info("SSE replay stream closed", session_id=session_id) diff --git a/backend/app/worker.py b/backend/app/worker.py index ded82a98..64bb803b 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -16,6 +16,7 @@ from app.db.docs import ALL_DOCUMENTS from app.dlq.manager import DLQManager from app.events.handlers import k8s_worker_router, result_processor_router, saga_router +from app.services.idempotency import IdempotencyManager from app.services.k8s_worker import KubernetesWorker from app.services.pod_monitor import PodMonitor from app.services.saga import SagaOrchestrator @@ -49,7 +50,7 @@ async def run(settings: Settings, logger: structlog.stdlib.BoundLogger) -> None: async def start_jobs(container: AsyncContainer) -> AsyncIOScheduler: - """Prepare the cluster, then schedule the pod watch, the saga queue and timeouts, and DLQ retries.""" + """Prepare the cluster, then schedule the pod watch, saga jobs, DLQ retries and the idempotency key count.""" k8s_worker = await container.get(KubernetesWorker) await k8s_worker.ensure_namespace_security() await k8s_worker.ensure_image_pre_puller_daemonset() @@ -57,6 +58,7 @@ async def start_jobs(container: AsyncContainer) -> AsyncIOScheduler: monitor = await container.get(PodMonitor) orchestrator = await container.get(SagaOrchestrator) dlq_manager = await container.get(DLQManager) + idempotency = await container.get(IdempotencyManager) scheduler = AsyncIOScheduler() jobs = ( @@ -64,6 +66,7 @@ async def start_jobs(container: AsyncContainer) -> AsyncIOScheduler: ("saga_try_schedule", orchestrator.try_schedule_from_queue, 10, 30), ("saga_check_timeouts", orchestrator.check_timeouts, 30, 60), ("dlq_monitor_retries", dlq_manager.process_monitoring_cycle, 10, 60), + ("idempotency_report_keys", idempotency.report_active_keys, 30, 60), ) for job_id, job, seconds, grace in jobs: scheduler.add_job( diff --git a/backend/grafana/provisioning/dashboards/event-stream-monitoring.json b/backend/grafana/provisioning/dashboards/event-stream-monitoring.json index 89223d58..b92f0086 100644 --- a/backend/grafana/provisioning/dashboards/event-stream-monitoring.json +++ b/backend/grafana/provisioning/dashboards/event-stream-monitoring.json @@ -322,7 +322,7 @@ "editorMode": "code", "expr": "sse_connections_active", "instant": false, - "legendFormat": "{{endpoint}}", + "legendFormat": "{{sse_endpoint}}", "range": true, "refId": "A" } @@ -515,9 +515,9 @@ { "datasource": "Victoria Metrics", "editorMode": "code", - "expr": "sum by (endpoint, event_type) (rate(sse_messages_sent_total[5m]))", + "expr": "sum by (sse_endpoint, sse_event_type) (rate(sse_messages_sent_total[5m]))", "instant": false, - "legendFormat": "{{endpoint}} {{event_type}}", + "legendFormat": "{{sse_endpoint}} {{sse_event_type}}", "range": true, "refId": "A" } diff --git a/backend/otel-collector-config.yaml b/backend/otel-collector-config.yaml index 8a14b9de..0eecad7c 100644 --- a/backend/otel-collector-config.yaml +++ b/backend/otel-collector-config.yaml @@ -778,6 +778,118 @@ connectors: sum: value: "1" monotonic: true + - name: events.published + description: Events stored and published by the producer + unit: "{event}" + attributes: + - key: event.type + conditions: + - span.name == "publish_event" and span.status.code != STATUS_CODE_ERROR + sum: + value: "1" + monotonic: true + - name: event.processing.duration + description: Time to store an event and publish it to Kafka + unit: s + attributes: + - key: event.type + conditions: + - span.name == "publish_event" and span.status.code != STATUS_CODE_ERROR + histogram: + buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5] + value: Seconds(span.end_time - span.start_time) + - name: idempotency.cache.hits + description: Idempotency checks that found the key already stored + unit: "{check}" + attributes: + - key: idempotency.event_type + - key: idempotency.operation + conditions: + - span.name == "idempotency.check" and span.attributes["idempotency.result"] == "hit" + sum: + value: "1" + monotonic: true + - name: idempotency.cache.misses + description: Idempotency checks that reserved a new key + unit: "{check}" + attributes: + - key: idempotency.event_type + - key: idempotency.operation + conditions: + - span.name == "idempotency.check" and span.attributes["idempotency.result"] == "miss" + sum: + value: "1" + monotonic: true + - name: idempotency.duplicates.blocked + description: Events skipped because their key was done or still in progress + unit: "{event}" + attributes: + - key: idempotency.event_type + conditions: + - span.name == "idempotency.duplicate_blocked" + sum: + value: "1" + monotonic: true + - name: idempotency.check.duration + description: Time to look up and reserve an idempotency key + unit: s + attributes: + - key: idempotency.operation + conditions: + - span.name == "idempotency.check" + histogram: + buckets: [0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25] + value: Seconds(span.end_time - span.start_time) + - name: idempotency.processing.duration + description: Time from reserving an idempotency key to marking it completed or failed + unit: s + attributes: + - key: idempotency.event_type + - key: idempotency.status + conditions: + - span.name == "idempotency.processed" + histogram: + buckets: [0.01, 0.05, 0.1, 0.5, 1, 5, 10, 30, 60, 300] + value: Double(span.attributes["idempotency.processing_seconds"]) + # The worker counts the keys in Redis every 30 s + - name: idempotency.keys.active + description: Idempotency keys currently stored in Redis + unit: "{key}" + conditions: + - span.name == "idempotency.keys" + gauge: + value: Int(span.attributes["idempotency.active_keys"]) + # Each API process reports its own open streams every 10 s, zeros included + - name: sse.connections.active + description: Open SSE streams, by endpoint + unit: "{connection}" + attributes: + - key: sse.endpoint + conditions: + - span.name == "sse.streams" + gauge: + value: Int(span.attributes["sse.open_streams"]) + - name: sse.messages.sent + description: Messages sent down SSE streams, by endpoint and event type + unit: "{message}" + attributes: + - key: sse.endpoint + - key: sse.event_type + conditions: + - span.name == "sse.message" + sum: + value: "1" + monotonic: true + - name: sse.connection.duration + description: How long an SSE stream stayed open + unit: s + attributes: + - key: sse.endpoint + conditions: + - span.name == "sse.stream_closed" + histogram: + buckets: [1, 5, 10, 30, 60, 300, 600, 1800, 3600, 7200] + value: Double(span.attributes["sse.duration_seconds"]) processors: batch: diff --git a/backend/tests/contract/test_grafana_metrics.py b/backend/tests/contract/test_grafana_metrics.py index c54c7c8c..d74cbd78 100644 --- a/backend/tests/contract/test_grafana_metrics.py +++ b/backend/tests/contract/test_grafana_metrics.py @@ -1,19 +1,15 @@ from __future__ import annotations -import importlib import json -import pkgutil import re from collections.abc import Iterator from pathlib import Path from typing import NotRequired, TypedDict -from unittest.mock import MagicMock import pytest import yaml from fastapi import FastAPI from faststream.kafka.opentelemetry import KafkaTelemetryMiddleware -from opentelemetry import metrics as otel_api from opentelemetry.exporter.prometheus import PrometheusMetricReader from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.system_metrics import SystemMetricsInstrumentor @@ -21,8 +17,7 @@ from opentelemetry.sdk.metrics import MeterProvider from prometheus_client.metrics_core import Metric -import app.core.metrics as metrics_pkg -from app.core.metrics.base import PROCESS_METRICS, BaseMetrics +from app.core.metrics.base import PROCESS_METRICS BACKEND_ROOT = Path(__file__).resolve().parent.parent.parent GRAFANA_DIR = BACKEND_ROOT / "grafana" / "provisioning" @@ -227,7 +222,7 @@ def _families(reader: PrometheusMetricReader) -> dict[str, set[str]]: @pytest.fixture(scope="module") def produced_families() -> Iterator[tuple[dict[str, set[str]], dict[str, set[str]]]]: - """Run every app metric class and the libraries' telemetry through the real OTel SDK. + """Run every span metric the collector defines and the libraries' telemetry through the real OTel SDK. The libraries are FastStream's Kafka middleware and the FastAPI and system metrics instrumentations, configured the way the app configures them. @@ -242,12 +237,7 @@ def produced_families() -> Iterator[tuple[dict[str, set[str]], dict[str, set[str app_provider = MeterProvider(metric_readers=[app_reader]) lib_reader = PrometheusMetricReader() lib_provider = MeterProvider(metric_readers=[lib_reader]) - otel_api.set_meter_provider(app_provider) - for _, mod_name, _ in pkgutil.iter_modules(metrics_pkg.__path__): - importlib.import_module(f"app.core.metrics.{mod_name}") - for cls in BaseMetrics.__subclasses__(): - cls(MagicMock()) _create_span_metrics(app_provider.get_meter("signaltometrics")) KafkaTelemetryMiddleware(meter_provider=lib_provider) api = FastAPI() diff --git a/backend/tests/e2e/idempotency/test_idempotency.py b/backend/tests/e2e/idempotency/test_idempotency.py index 281503b1..2f696af5 100644 --- a/backend/tests/e2e/idempotency/test_idempotency.py +++ b/backend/tests/e2e/idempotency/test_idempotency.py @@ -6,12 +6,11 @@ import pytest import redis.asyncio as redis import structlog +from opentelemetry import trace -from app.core.metrics import IdempotencyMetrics from app.domain.enums import EventType from app.domain.idempotency import IdempotencyRecord, IdempotencyStatus, KeyStrategy from app.services.idempotency import IdempotencyConfig, IdempotencyManager, RedisIdempotencyRepository -from app.settings import Settings from tests.conftest import make_execution_requested_event pytestmark = [pytest.mark.e2e] @@ -24,7 +23,7 @@ class TestIdempotencyManager: """IdempotencyManager backed by real Redis repository (DI-provided client).""" @pytest.fixture - def manager(self, redis_client: redis.Redis, test_settings: Settings) -> IdempotencyManager: + def manager(self, redis_client: redis.Redis) -> IdempotencyManager: prefix = f"idemp_ut:{uuid.uuid4().hex[:6]}" cfg = IdempotencyConfig( key_prefix=prefix, @@ -34,8 +33,7 @@ def manager(self, redis_client: redis.Redis, test_settings: Settings) -> Idempot max_result_size_bytes=1024, ) repo = RedisIdempotencyRepository(redis_client, key_prefix=prefix) - idempotency_metrics = IdempotencyMetrics(test_settings) - return IdempotencyManager(cfg, repo, _test_logger, idempotency_metrics=idempotency_metrics) + return IdempotencyManager(cfg, repo, _test_logger, tracer=trace.NoOpTracer()) async def test_complete_flow_new_event(self, manager: IdempotencyManager) -> None: """Test the complete flow for a new event""" diff --git a/backend/tests/e2e/services/idempotency/test_redis_repository.py b/backend/tests/e2e/services/idempotency/test_redis_repository.py index 98e90b45..19e75bbf 100644 --- a/backend/tests/e2e/services/idempotency/test_redis_repository.py +++ b/backend/tests/e2e/services/idempotency/test_redis_repository.py @@ -133,3 +133,13 @@ async def test_update_record_when_missing( # If key missing, update returns 0 res = await repository.update_record(sample_record) assert res == 0 + + +async def test_count_keys_counts_only_its_own_prefix(redis_client: redis.Redis) -> None: + prefix = f"count-{uuid4().hex[:8]}" + repository = RedisIdempotencyRepository(redis_client, key_prefix=prefix) + for n in range(3): + await redis_client.set(f"{prefix}:key-{n}", "{}", ex=60) + await redis_client.set(f"{prefix}-other:key", "{}", ex=60) + + assert await repository.count_keys() == 3 diff --git a/backend/tests/e2e/services/sse/test_partitioned_event_router.py b/backend/tests/e2e/services/sse/test_partitioned_event_router.py index 995e51ea..c2ebcae9 100644 --- a/backend/tests/e2e/services/sse/test_partitioned_event_router.py +++ b/backend/tests/e2e/services/sse/test_partitioned_event_router.py @@ -1,13 +1,12 @@ import asyncio import contextlib -from unittest.mock import MagicMock from uuid import uuid4 import pytest import redis.asyncio as redis import structlog +from opentelemetry import trace -from app.core.metrics import ConnectionMetrics from app.domain.enums import EventType from app.domain.sse import SSEExecutionEventData from app.services.sse import SSERedisBus @@ -23,7 +22,7 @@ async def test_bus_routes_event_to_redis(redis_client: redis.Redis, test_setting bus = SSERedisBus( redis_client, logger=_test_logger, - connection_metrics=MagicMock(spec=ConnectionMetrics), + tracer=trace.NoOpTracer(), exec_prefix=f"sse:exec:{suffix}:", notif_prefix=f"sse:notif:{suffix}:", ) diff --git a/backend/tests/unit/app/test_worker_entrypoint.py b/backend/tests/unit/app/test_worker_entrypoint.py index 9df45f10..db8dfd3c 100644 --- a/backend/tests/unit/app/test_worker_entrypoint.py +++ b/backend/tests/unit/app/test_worker_entrypoint.py @@ -49,5 +49,6 @@ async def test_start_jobs_prepares_the_cluster_and_schedules_the_jobs(monkeypatc "saga_try_schedule", "saga_check_timeouts", "dlq_monitor_retries", + "idempotency_report_keys", ] scheduler.start.assert_called_once() diff --git a/backend/tests/unit/conftest.py b/backend/tests/unit/conftest.py index 4c1a4982..c9343d83 100644 --- a/backend/tests/unit/conftest.py +++ b/backend/tests/unit/conftest.py @@ -22,15 +22,9 @@ from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from app.core.metrics import ( - ConnectionMetrics, - EventMetrics, - IdempotencyMetrics, -) from app.domain.enums import ExecutionStatus, QueuePriority from app.domain.execution import DomainExecution from app.runtime_registry import DEFAULT_LANG, DEFAULT_LANG_VERSION -from app.settings import Settings def make_domain_execution( @@ -206,11 +200,6 @@ def mock_watch() -> MagicMock: return make_mock_watch([]) -@pytest.fixture -def connection_metrics(test_settings: Settings) -> ConnectionMetrics: - return ConnectionMetrics(test_settings) - - @pytest.fixture def span_exporter(monkeypatch: pytest.MonkeyPatch) -> InMemorySpanExporter: """Collects the spans a test finishes. pytest-env sets OTEL_SDK_DISABLED, so the SDK is turned back on here.""" @@ -225,16 +214,6 @@ def tracer(span_exporter: InMemorySpanExporter) -> trace.Tracer: return provider.get_tracer("test") -@pytest.fixture -def idempotency_metrics(test_settings: Settings) -> IdempotencyMetrics: - return IdempotencyMetrics(test_settings) - - -@pytest.fixture -def event_metrics(test_settings: Settings) -> EventMetrics: - return EventMetrics(test_settings) - - @pytest.fixture def db() -> NoReturn: raise RuntimeError("Unit tests should not access DB; use mocks, or move the test to tests/e2e") diff --git a/backend/tests/unit/core/metrics/__init__.py b/backend/tests/unit/core/metrics/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/backend/tests/unit/core/metrics/test_base_metrics.py b/backend/tests/unit/core/metrics/test_base_metrics.py deleted file mode 100644 index eab1efd2..00000000 --- a/backend/tests/unit/core/metrics/test_base_metrics.py +++ /dev/null @@ -1,28 +0,0 @@ -import pytest - -from app.core.metrics import BaseMetrics -from app.settings import Settings - -pytestmark = pytest.mark.unit - - -class DummyMetrics(BaseMetrics): - def __init__(self, settings: Settings) -> None: - self.created = False - super().__init__(settings, meter_name="dummy") - - def _create_instruments(self) -> None: - self.created = True - - -def test_base_metrics_initializes_meter_and_instruments(test_settings: Settings) -> None: - """Test that BaseMetrics initializes properly with no-op metrics.""" - # Create DummyMetrics instance - will use NoOpMeterProvider automatically - m = DummyMetrics(test_settings) - - # Verify that the BaseMetrics init was called and instruments were created - assert m.created is True - assert m._meter is not None # Meter exists (will be NoOpMeter) - - # close is no-op - m.close() diff --git a/backend/tests/unit/core/metrics/test_database_and_dlq_metrics.py b/backend/tests/unit/core/metrics/test_database_and_dlq_metrics.py deleted file mode 100644 index f345fec1..00000000 --- a/backend/tests/unit/core/metrics/test_database_and_dlq_metrics.py +++ /dev/null @@ -1,31 +0,0 @@ -from unittest.mock import MagicMock - -import pytest -import redis - -from app.core.metrics import IdempotencyMetrics -from app.settings import Settings - -pytestmark = pytest.mark.unit - - -@pytest.mark.parametrize( - ("keys", "error", "expected"), - [ - (["idempotency:a", "idempotency:b", "idempotency:c"], None, [3]), - ([], None, [0]), - ([], redis.ConnectionError("down"), []), - ], -) -async def test_idempotency_active_keys_gauge_counts_redis_keys( - test_settings: Settings, keys: list[str], error: redis.RedisError | None, expected: list[int] -) -> None: - m = IdempotencyMetrics(test_settings) - m._redis = MagicMock(spec=redis.Redis) - m._redis.scan_iter.return_value = iter(keys) - m._redis.scan_iter.side_effect = error - - observations = m._observe_active_keys(MagicMock()) - - assert [o.value for o in observations] == expected - m._redis.scan_iter.assert_called_once_with(match="idempotency:*", count=1000) diff --git a/backend/tests/unit/events/test_producer.py b/backend/tests/unit/events/test_producer.py index 2aa7df15..5dceae3e 100644 --- a/backend/tests/unit/events/test_producer.py +++ b/backend/tests/unit/events/test_producer.py @@ -1,9 +1,11 @@ -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock import pytest from aiokafka.errors import KafkaError +from opentelemetry import trace +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import StatusCode -from app.core.metrics import EventMetrics from app.db import EventRepository from app.domain.events import EventMetadata, UserRegisteredEvent from app.events import UnifiedProducer @@ -21,27 +23,35 @@ def _event() -> UserRegisteredEvent: ) -def _producer() -> tuple[UnifiedProducer, AsyncMock, AsyncMock, MagicMock]: +def _producer(tracer: trace.Tracer) -> tuple[UnifiedProducer, AsyncMock, AsyncMock]: repo = AsyncMock(spec=EventRepository) transport = AsyncMock(spec=KafkaEventTransport) - metrics = MagicMock(spec=EventMetrics) - return UnifiedProducer(repo, transport, metrics), repo, transport, metrics + return UnifiedProducer(repo, transport, tracer), repo, transport -async def test_produce_stores_publishes_and_counts_every_event() -> None: - producer, repo, transport, metrics = _producer() +async def test_produce_stores_publishes_and_reports_a_clean_span( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + producer, repo, transport = _producer(tracer) event = _event() await producer.produce(event_to_produce=event, key="u1") repo.store_event.assert_awaited_once_with(event) transport.publish.assert_awaited_once_with(event, event.event_type, "u1") - metrics.record_event_published.assert_called_once_with(event.event_type) - metrics.record_event_processing_duration.assert_called_once() + [span] = span_exporter.get_finished_spans() + assert span.attributes is not None + assert (span.name, span.attributes["event.type"], span.status.status_code) == ( + "publish_event", + event.event_type, + StatusCode.UNSET, + ) -async def test_failed_publish_is_marked_and_not_counted() -> None: - producer, repo, transport, metrics = _producer() +async def test_failed_publish_is_marked_and_its_span_is_an_error( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + producer, repo, transport = _producer(tracer) transport.publish.side_effect = KafkaError("broker down") event = _event() @@ -49,4 +59,5 @@ async def test_failed_publish_is_marked_and_not_counted() -> None: await producer.produce(event_to_produce=event, key="u1") repo.mark_publish_failed.assert_awaited_once_with(event.event_id) - metrics.record_event_published.assert_not_called() + [span] = span_exporter.get_finished_spans() + assert span.status.status_code == StatusCode.ERROR diff --git a/backend/tests/unit/services/idempotency/test_idempotency_manager.py b/backend/tests/unit/services/idempotency/test_idempotency_manager.py index ae9cc93d..bbdacfc5 100644 --- a/backend/tests/unit/services/idempotency/test_idempotency_manager.py +++ b/backend/tests/unit/services/idempotency/test_idempotency_manager.py @@ -3,8 +3,9 @@ import pytest import structlog +from opentelemetry import trace +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from app.core.metrics import IdempotencyMetrics from app.domain.enums import EventType from app.domain.events import BaseEvent from app.domain.idempotency import IdempotencyRecord, IdempotencyStatus, KeyStrategy @@ -16,9 +17,9 @@ _test_logger = structlog.get_logger("test.idempotency_manager") -def test_manager_generate_key_variants(idempotency_metrics: IdempotencyMetrics) -> None: +def test_manager_generate_key_variants() -> None: repo = MagicMock() - mgr = IdempotencyManager(IdempotencyConfig(), repo, _test_logger, idempotency_metrics=idempotency_metrics) + mgr = IdempotencyManager(IdempotencyConfig(), repo, _test_logger, tracer=trace.NoOpTracer()) ev = MagicMock(spec=BaseEvent) ev.event_type = "t" ev.event_id = "e" @@ -39,8 +40,19 @@ def _event() -> MagicMock: return ev -def _manager(repo: AsyncMock, metrics: MagicMock) -> IdempotencyManager: - return IdempotencyManager(IdempotencyConfig(), repo, _test_logger, idempotency_metrics=metrics) +def _manager(repo: AsyncMock, tracer: trace.Tracer) -> IdempotencyManager: + return IdempotencyManager(IdempotencyConfig(), repo, _test_logger, tracer=tracer) + + +def _record(status: IdempotencyStatus) -> IdempotencyRecord: + return IdempotencyRecord( + key="idempotency:create_pod_command:evt-1", + status=status, + event_type=EventType.CREATE_POD_COMMAND, + event_id="evt-1", + created_at=datetime.now(timezone.utc) - timedelta(seconds=2), + ttl_seconds=3600, + ) @pytest.mark.parametrize( @@ -52,51 +64,87 @@ def _manager(repo: AsyncMock, metrics: MagicMock) -> IdempotencyManager: ], ) async def test_marking_records_processing_duration( - method: str, kwargs: dict[str, str], expected_status: IdempotencyStatus + tracer: trace.Tracer, + span_exporter: InMemorySpanExporter, + method: str, + kwargs: dict[str, str], + expected_status: IdempotencyStatus, ) -> None: repo = AsyncMock(spec=RedisIdempotencyRepository) - repo.find_by_key.return_value = IdempotencyRecord( - key="idempotency:create_pod_command:evt-1", - status=IdempotencyStatus.PROCESSING, - event_type=EventType.CREATE_POD_COMMAND, - event_id="evt-1", - created_at=datetime.now(timezone.utc) - timedelta(seconds=2), - ttl_seconds=3600, - ) + repo.find_by_key.return_value = _record(IdempotencyStatus.PROCESSING) repo.update_record.return_value = 1 - metrics = MagicMock(spec=IdempotencyMetrics) - assert await getattr(_manager(repo, metrics), method)(_event(), **kwargs) is True - - metrics.record_idempotent_processing_duration.assert_called_once() - duration, event_type, status = metrics.record_idempotent_processing_duration.call_args.args - assert duration >= 2.0 - assert event_type == EventType.CREATE_POD_COMMAND - assert status == expected_status + assert await getattr(_manager(repo, tracer), method)(_event(), **kwargs) is True + + [span] = span_exporter.get_finished_spans() + assert span.name == "idempotency.processed" + assert span.attributes is not None + seconds = span.attributes["idempotency.processing_seconds"] + assert isinstance(seconds, float) + assert seconds >= 2.0 + assert (span.attributes["idempotency.event_type"], span.attributes["idempotency.status"]) == ( + EventType.CREATE_POD_COMMAND, + expected_status, + ) -async def test_marking_missing_key_records_nothing() -> None: +async def test_marking_missing_key_records_nothing(tracer: trace.Tracer, span_exporter: InMemorySpanExporter) -> None: repo = AsyncMock(spec=RedisIdempotencyRepository) repo.find_by_key.return_value = None - metrics = MagicMock(spec=IdempotencyMetrics) - assert await _manager(repo, metrics).mark_completed(_event()) is False + assert await _manager(repo, tracer).mark_completed(_event()) is False - metrics.record_idempotent_processing_duration.assert_not_called() + assert span_exporter.get_finished_spans() == () repo.update_record.assert_not_called() -async def test_check_and_reserve_records_check_duration_only() -> None: +async def test_check_and_reserve_of_a_new_key_reports_a_miss( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: repo = AsyncMock(spec=RedisIdempotencyRepository) repo.find_by_key.return_value = None - metrics = MagicMock(spec=IdempotencyMetrics) - result = await _manager(repo, metrics).check_and_reserve(_event()) + result = await _manager(repo, tracer).check_and_reserve(_event()) assert result.is_duplicate is False assert result.status == IdempotencyStatus.PROCESSING repo.insert_processing.assert_awaited_once() - metrics.record_idempotency_cache_miss.assert_called_once_with(EventType.CREATE_POD_COMMAND, "check_and_reserve") - metrics.record_idempotency_check_duration.assert_called_once() - assert metrics.record_idempotency_check_duration.call_args.args[1] == "check_and_reserve" - metrics.record_idempotent_processing_duration.assert_not_called() + [span] = span_exporter.get_finished_spans() + assert (span.name, span.attributes) == ( + "idempotency.check", + { + "idempotency.event_type": EventType.CREATE_POD_COMMAND, + "idempotency.operation": "check_and_reserve", + "idempotency.result": "miss", + }, + ) + + +async def test_check_and_reserve_of_a_done_key_reports_a_hit_and_a_blocked_duplicate( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + repo = AsyncMock(spec=RedisIdempotencyRepository) + repo.find_by_key.return_value = _record(IdempotencyStatus.COMPLETED) + + result = await _manager(repo, tracer).check_and_reserve(_event()) + + assert result.is_duplicate is True + blocked, check = span_exporter.get_finished_spans() + assert (blocked.name, blocked.attributes) == ( + "idempotency.duplicate_blocked", + {"idempotency.event_type": EventType.CREATE_POD_COMMAND}, + ) + assert check.attributes is not None + assert (check.name, check.attributes["idempotency.result"]) == ("idempotency.check", "hit") + + +async def test_report_active_keys_puts_the_redis_count_on_a_span( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + repo = AsyncMock(spec=RedisIdempotencyRepository) + repo.count_keys.return_value = 3 + + await _manager(repo, tracer).report_active_keys() + + [span] = span_exporter.get_finished_spans() + assert (span.name, span.attributes) == ("idempotency.keys", {"idempotency.active_keys": 3}) diff --git a/backend/tests/unit/services/sse/test_redis_bus.py b/backend/tests/unit/services/sse/test_redis_bus.py index dedfe10f..d2542b01 100644 --- a/backend/tests/unit/services/sse/test_redis_bus.py +++ b/backend/tests/unit/services/sse/test_redis_bus.py @@ -5,13 +5,13 @@ from contextlib import aclosing from datetime import datetime, timezone from typing import Any, cast -from unittest.mock import MagicMock import pytest import redis.asyncio as redis_async import structlog +from opentelemetry import trace +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from app.core.metrics import ConnectionMetrics from app.domain.enums import EventType, NotificationSeverity, NotificationStatus, ReplayStatus from app.domain.sse import DomainNotificationSSEPayload, DomainReplaySSEPayload, SSEExecutionEventData from app.services.sse import SSERedisBus @@ -69,10 +69,13 @@ def pubsub(self, ignore_subscribe_messages: bool = False) -> _FakePubSub: return self._pubsub -async def test_publish_and_subscribe_round_trip() -> None: +def _spans(exporter: InMemorySpanExporter, name: str) -> list[dict[str, Any]]: + return [dict(span.attributes or {}) for span in exporter.get_finished_spans() if span.name == name] + + +async def test_publish_and_subscribe_round_trip(tracer: trace.Tracer, span_exporter: InMemorySpanExporter) -> None: r = _FakeRedis() - metrics = MagicMock(spec=ConnectionMetrics) - bus = SSERedisBus(cast(redis_async.Redis, r), logger=_test_logger, connection_metrics=metrics) + bus = SSERedisBus(cast(redis_async.Redis, r), logger=_test_logger, tracer=tracer) # Publish event as SSEExecutionEventData (field projection happens in handlers.py) evt = SSEExecutionEventData( @@ -105,14 +108,14 @@ async def test_publish_and_subscribe_round_trip() -> None: await r._pubsub.push(ch, good_payload) msg2 = await asyncio.wait_for(messages.__anext__(), timeout=2.0) assert msg2.event_type == EventType.EXECUTION_COMPLETED - assert metrics.record_sse_message_sent.call_count == 2 - metrics.record_sse_message_sent.assert_called_with("executions", EventType.EXECUTION_COMPLETED) + assert _spans(span_exporter, "sse.message") == 2 * [ + {"sse.endpoint": "executions", "sse.event_type": EventType.EXECUTION_COMPLETED} + ] -async def test_notifications_channels() -> None: +async def test_notifications_channels(tracer: trace.Tracer, span_exporter: InMemorySpanExporter) -> None: r = _FakeRedis() - metrics = MagicMock(spec=ConnectionMetrics) - bus = SSERedisBus(cast(redis_async.Redis, r), logger=_test_logger, connection_metrics=metrics) + bus = SSERedisBus(cast(redis_async.Redis, r), logger=_test_logger, tracer=tracer) notif = DomainNotificationSSEPayload( notification_id="n1", @@ -136,13 +139,16 @@ async def test_notifications_channels() -> None: # Subscription happened inside __anext__ assert "sse:notif:user-1" in r._pubsub.subscribed assert got.notification_id == "n1" - metrics.record_sse_message_sent.assert_called_once_with("notifications", "notification") + assert _spans(span_exporter, "sse.message") == [ + {"sse.endpoint": "notifications", "sse.event_type": "notification"} + ] -async def test_replay_publish_and_subscribe_round_trip() -> None: +async def test_replay_publish_and_subscribe_round_trip( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: r = _FakeRedis() - metrics = MagicMock(spec=ConnectionMetrics) - bus = SSERedisBus(cast(redis_async.Redis, r), logger=_test_logger, connection_metrics=metrics) + bus = SSERedisBus(cast(redis_async.Redis, r), logger=_test_logger, tracer=tracer) status = DomainReplaySSEPayload( session_id="sess-1", @@ -167,4 +173,35 @@ async def test_replay_publish_and_subscribe_round_trip() -> None: assert got.session_id == "sess-1" assert got.status == ReplayStatus.RUNNING assert got.replayed_events == 3 - metrics.record_sse_message_sent.assert_called_once_with("replay", "replay_status") + assert _spans(span_exporter, "sse.message") == [{"sse.endpoint": "replay", "sse.event_type": "replay_status"}] + + +async def test_open_streams_are_reported_per_endpoint_and_closing_records_the_duration( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + r = _FakeRedis() + bus = SSERedisBus(cast(redis_async.Redis, r), logger=_test_logger, tracer=tracer) + await r._pubsub.push( + "sse:exec:exec-1", + _sse_event_adapter.dump_json( + SSEExecutionEventData(event_type=EventType.EXECUTION_COMPLETED, execution_id="exec-1") + ), + ) + + async with aclosing(bus.listen_execution("exec-1")) as messages: + await asyncio.wait_for(messages.__anext__(), timeout=2.0) + await bus.report_open_streams() + await bus.report_open_streams() + + reports = [(a["sse.endpoint"], a["sse.open_streams"]) for a in _spans(span_exporter, "sse.streams")] + assert reports == [ + ("executions", 1), + ("notifications", 0), + ("replay", 0), + ("executions", 0), + ("notifications", 0), + ("replay", 0), + ] + [closed] = _spans(span_exporter, "sse.stream_closed") + assert closed["sse.endpoint"] == "executions" + assert closed["sse.duration_seconds"] >= 0 diff --git a/backend/tests/unit/services/test_execution_service.py b/backend/tests/unit/services/test_execution_service.py index 4376086f..a2888450 100644 --- a/backend/tests/unit/services/test_execution_service.py +++ b/backend/tests/unit/services/test_execution_service.py @@ -1,5 +1,5 @@ from datetime import datetime, timezone -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock import pytest import structlog @@ -7,7 +7,6 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from app.core.exceptions.handlers import _map_to_status_code -from app.core.metrics import IdempotencyMetrics from app.db import ExecutionRepository from app.domain.admin import SystemSettings from app.domain.enums import EventType, ExecutionStatus, UserRole @@ -197,7 +196,7 @@ def idem_service( runtime_settings = AsyncMock(spec=RuntimeSettingsLoader) runtime_settings.get_effective_settings.return_value = SystemSettings() logger = structlog.get_logger("test") - manager = IdempotencyManager(IdempotencyConfig(), idem_repo, logger, MagicMock(spec=IdempotencyMetrics)) + manager = IdempotencyManager(IdempotencyConfig(), idem_repo, logger, trace.NoOpTracer()) return ExecutionService( execution_repo=repo, producer=producer, diff --git a/docs/architecture/execution-queue.md b/docs/architecture/execution-queue.md index bd6c880a..149ec2ae 100644 --- a/docs/architecture/execution-queue.md +++ b/docs/architecture/execution-queue.md @@ -75,15 +75,4 @@ The queue status reads Redis directly: `queue_depth` is the sum of `ZCARD` over ## Metrics -`QueueMetrics` in `backend/app/core/metrics/queue.py` defines these. The second column is the series name in VictoriaMetrics. - -| Instrument | Series | Meaning | -|------------|--------|---------| -| `queue.depth` | `queue_depth` | Up-down counter: up on enqueue, down on schedule | -| `queue.active` | `queue_active` | Up-down counter: up on schedule, down when `release` or `remove` takes an id out of the running set; slots dropped for expiry don't bring it down | -| `queue.enqueue.total` | `queue_enqueue_total` | Enqueues, including re-enqueues after a failed saga start | -| `queue.schedule.total` | `queue_schedule_total` | Executions popped from the queue | -| `queue.wait_time` | `queue_wait_time_seconds` | Histogram of enqueue to schedule, by `priority` | -| `queue.event_data_lost` | `queue_event_data_lost_total` | Scheduled ids whose stored event had expired | - -`queue_depth` and `queue_active` are counted in the orchestrator process, so they start from zero after a restart. `queue_depth` also drifts: an id popped with an expired event, or removed by a cancel, leaves the pending set without decrementing it. The admin queue endpoint is the number to trust. The [metrics reference](../operations/metrics-reference.md) lists the rest. +The queue's six series come from spans: `queue.enqueue` and `queue.schedule` from `ExecutionQueueService`, and `queue.schedule_pending` from the saga scheduler's 10-second job, which reads the depth and the running slots from Redis. So `queue_depth` and `queue_active` match the admin queue endpoint and survive restarts. The [metrics reference](../operations/metrics-reference.md#executions-and-queue) lists the series and their labels. diff --git a/docs/architecture/idempotency.md b/docs/architecture/idempotency.md index 5338ca7d..60cedf97 100644 --- a/docs/architecture/idempotency.md +++ b/docs/architecture/idempotency.md @@ -92,13 +92,4 @@ The DI provider builds `IdempotencyConfig()` with these defaults; nothing in `co ## Metrics -`IdempotencyMetrics` in `backend/app/core/metrics/database.py` defines these. The second column is the series name in VictoriaMetrics after the collector's Prometheus conversion. - -| Instrument | Series | Meaning | -|------------|--------|---------| -| `idempotency.cache.hits.total` | `idempotency_cache_hits_total` | Lookups that found a record, by `event_type` and `operation` | -| `idempotency.cache.misses.total` | `idempotency_cache_misses_total` | Lookups that found no record and went on to `SET NX`, whether or not they won it | -| `idempotency.duplicates.blocked.total` | `idempotency_duplicates_blocked_total` | Calls rejected as duplicates, by `event_type` | -| `idempotency.check.duration` | `idempotency_check_duration_seconds` | Histogram of `check_and_reserve` time | -| `idempotency.processing.duration` | `idempotency_processing_duration_seconds` | Histogram of reserve to completed or failed, by `event_type` and `status` | -| `idempotency.keys.active` | `idempotency_keys_active` | Observable gauge: `SCAN` count of `idempotency:*` keys in Redis at each export | +`IdempotencyManager` puts each `check_and_reserve` on an `idempotency.check` span with a hit or miss, and leaves `idempotency.duplicate_blocked` and `idempotency.processed` spans for blocked duplicates and finished keys. The worker counts the `idempotency:*` keys in Redis with `SCAN` every 30 seconds for `idempotency_keys_active`. A miss is a lookup that found no record and went on to `SET NX`, whether or not it won it. The [metrics reference](../operations/metrics-reference.md#events-and-idempotency) lists the series and their labels. diff --git a/docs/operations/metrics-reference.md b/docs/operations/metrics-reference.md index 66d27dfa..d3c451e3 100644 --- a/docs/operations/metrics-reference.md +++ b/docs/operations/metrics-reference.md @@ -14,17 +14,13 @@ flowchart LR VM --> Grafana ``` -`create_app` and the worker's `main()` each call `setup_metrics` (`app/core/metrics/base.py`) once. It installs a global `MeterProvider` whose `PeriodicExportingMetricReader` exports every 60 000 ms through an OTLP gRPC exporter. The exporter targets `OTEL_EXPORTER_OTLP_ENDPOINT`, which `config.toml` sets to `http://otel-collector:4317`. It also starts the system metrics instrumentation for the process's own CPU and memory. Without an endpoint it returns early and the no-op provider stays. Each metrics class takes its meter from the global provider when DI builds it, so `setup_metrics` has to run before the container resolves anything: - -```python ---8<-- "backend/app/core/metrics/base.py:init" -``` +`create_app` and the worker's `main()` each call `setup_metrics` (`app/core/metrics/base.py`) once. It installs a global `MeterProvider` whose `PeriodicExportingMetricReader` exports every 60 000 ms through an OTLP gRPC exporter. The exporter targets `OTEL_EXPORTER_OTLP_ENDPOINT`, which `config.toml` sets to `http://otel-collector:4317`. It also starts the system metrics instrumentation for the process's own CPU and memory. Without an endpoint it returns early and the no-op provider stays. The OpenTelemetry instrumentations take their meters from that provider. The app's own metrics have no meter at all: they come from spans, described next. The collector's metrics pipeline (`backend/otel-collector-config.yaml`) reads the `otlp`, `host_metrics`, `kafka_metrics` and `mongodb` receivers, runs `memory_limiter`, `batch` and `resource`, and exports with `prometheus_remote_write` to `http://victoria-metrics:8428/api/v1/write`. VictoriaMetrics keeps 30 days. The collector, VictoriaMetrics and Grafana sit in the `observability` compose profile, so none of this runs unless you start the stack with `./deploy.sh dev --observability`. Counters and histograms made from spans take another way, described next. ## Metrics made from spans -Some metrics have no instrument in the code. A service puts the facts on a span, and the collector's `signaltometrics` connector (alpha in collector v0.162.0) turns them into counters, histograms and gauges, defined under `connectors` in `backend/otel-collector-config.yaml`. The SDK sends every span (`TRACING_SAMPLING_RATE = 1.0`), and the traces pipeline gives each span to the connector before `tail_sampling` decides what Jaeger keeps, so the counts don't depend on sampling. The connector emits a point only when a matching span arrives. Sent straight to VictoriaMetrics, a counter that sees one event every half hour would get one point every half hour, and `increase` and `rate` stop finding the previous point once two points are more than about 8 to 10 minutes apart (measured on v1.153.0): `increase` then counts the whole total again, and `rate` returns nothing. So counters and histograms go another way. `deltatocumulative` adds up the connector's deltas and keeps each total through 24 hours without spans (its default is 5 minutes, after which an idle counter would start again from zero). The collector's `prometheus` exporter serves the totals on `otel-collector:8889`, and VictoriaMetrics scrapes it every 15 seconds (`backend/victoria-metrics/scrape.yaml`, with `honor_labels` so each service's `job` and `instance` stay). A series the exporter hasn't updated for 24 hours drops out until its next span. Gauges still go by remote write, so a gauge goes stale when the job that reports it on a timer stops. Panels wrap counter rates in `or vector(0)` so a series that hasn't started yet reads 0. The execution, pod, queue, dead letter queue, notification, replay, security and rate limit metrics work this way so far ([#389](https://github.com/HardMax71/Integr8sCode/issues/389)). +Some metrics have no instrument in the code. A service puts the facts on a span, and the collector's `signaltometrics` connector (alpha in collector v0.162.0) turns them into counters, histograms and gauges, defined under `connectors` in `backend/otel-collector-config.yaml`. The SDK sends every span (`TRACING_SAMPLING_RATE = 1.0`), and the traces pipeline gives each span to the connector before `tail_sampling` decides what Jaeger keeps, so the counts don't depend on sampling. The connector emits a point only when a matching span arrives. Sent straight to VictoriaMetrics, a counter that sees one event every half hour would get one point every half hour, and `increase` and `rate` stop finding the previous point once two points are more than about 8 to 10 minutes apart (measured on v1.153.0): `increase` then counts the whole total again, and `rate` returns nothing. So counters and histograms go another way. `deltatocumulative` adds up the connector's deltas and keeps each total through 24 hours without spans (its default is 5 minutes, after which an idle counter would start again from zero). The collector's `prometheus` exporter serves the totals on `otel-collector:8889`, and VictoriaMetrics scrapes it every 15 seconds (`backend/victoria-metrics/scrape.yaml`, with `honor_labels` so each service's `job` and `instance` stay). A series the exporter hasn't updated for 24 hours drops out until its next span. Gauges still go by remote write, so a gauge goes stale when the job that reports it on a timer stops. Panels wrap counter rates in `or vector(0)` so a series that hasn't started yet reads 0. Every app metric works this way since [#389](https://github.com/HardMax71/Integr8sCode/issues/389). ## Names and labels in VictoriaMetrics @@ -36,7 +32,7 @@ Code names instruments with dots, and they arrive with underscores. The collecto The contract test `backend/tests/contract/test_grafana_metrics.py` applies the same rule. It fails when a dashboard or alert queries a name nothing emits, and when code defines a metric that no dashboard uses. -Resource attributes become labels on every series: `job` (`integr8scode/integr8scode-backend`), `service_name`, `service_namespace`, `service_version`, `deployment_environment`, the `telemetry_sdk_*` set, and `instance` with `service_instance_id`, a fresh UUID each time the process starts. `otel_scope_name` names the metrics class, such as `KubernetesMetrics`. Since `instance` changes on every restart, wrap queries in `sum` or `max` instead of reading single series. +Resource attributes become labels on every series: `job` (`integr8scode/integr8scode-backend`), `service_name`, `service_namespace`, `service_version`, `deployment_environment`, the `telemetry_sdk_*` set, and `instance` with `service_instance_id`, a fresh UUID each time the process starts. `otel_scope_name` names what made the series: an instrumentation such as `opentelemetry.instrumentation.fastapi`, or the collector's signaltometrics connector for every metric made from spans. Since `instance` changes on every restart, wrap queries in `sum` or `max` instead of reading single series. Gauges listed as "in process" are up-down counters the process keeps itself. They start at zero when the process starts, whatever the real state is. @@ -75,28 +71,28 @@ All of these come from spans (see [Metrics made from spans](#metrics-made-from-s ### Events and idempotency -| Metric | Type | Labels | Recorded in | -|--------|------|--------|-------------| -| `events_published_total` | counter | event_type, event_category | `UnifiedProducer.produce`, after the event is stored and published, in every process that publishes | -| `event_processing_duration_seconds` | histogram | event_type | the same call: persisting the event and publishing it | -| `idempotency_cache_hits_total` | counter | event_type, operation | `IdempotencyManager.check_and_reserve` | -| `idempotency_cache_misses_total` | counter | event_type, operation | `IdempotencyManager.check_and_reserve` | -| `idempotency_duplicates_blocked_total` | counter | event_type | `IdempotencyManager`, when the key is already completed or still processing | -| `idempotency_check_duration_seconds` | histogram | operation | `IdempotencyManager.check_and_reserve` | -| `idempotency_processing_duration_seconds` | histogram | event_type, status | `IdempotencyManager`: reserving the key to marking it completed or failed | -| `idempotency_keys_active` | gauge, observed at export | none | `IdempotencyMetrics` scans `idempotency:*` in Redis | +| Metric | Type | Labels | Recorded from | +|--------|------|--------|---------------| +| `events_published_total` | counter | event_type | span `publish_event` from `UnifiedProducer.produce` without an error status: the event was stored and published, in every process that publishes | +| `event_processing_duration_seconds` | histogram | event_type | the length of those spans: persisting the event and publishing it | +| `idempotency_cache_hits_total` | counter | idempotency_event_type, idempotency_operation | span `idempotency.check` with `idempotency.result` `hit` | +| `idempotency_cache_misses_total` | counter | idempotency_event_type, idempotency_operation | the same span with `idempotency.result` `miss` | +| `idempotency_duplicates_blocked_total` | counter | idempotency_event_type | span `idempotency.duplicate_blocked`, when the key is already completed or still processing | +| `idempotency_check_duration_seconds` | histogram | idempotency_operation | the length of `idempotency.check` | +| `idempotency_processing_duration_seconds` | histogram | idempotency_event_type, idempotency_status | `idempotency.processing_seconds` on span `idempotency.processed`: reserving the key to marking it completed or failed | +| `idempotency_keys_active` | gauge | none | span `idempotency.keys` from the worker's job every 30 s, which counts `idempotency:*` in Redis with `SCAN` | -Every Kafka publish also shows up in `messaging_publish_messages_message_total`; a publish that fails is marked on the stored event and not counted here. `event_category` equals `event_type` for the current event names. A handler that raises has no counter of its own: it's a `messaging_process_duration_seconds` sample with an `error_type` label, in the FastStream families below. `IdempotencyManager` runs in backend for `Idempotency-Key` requests and in the worker's k8s worker, result processor and saga orchestrator consumers. +Every Kafka publish also shows up in `messaging_publish_messages_message_total`; a publish that fails is marked on the stored event and not counted here. A handler that raises has no counter of its own: it's a `messaging_process_duration_seconds` sample with an `error_type` label, in the FastStream families below. `IdempotencyManager` runs in backend for `Idempotency-Key` requests and in the worker's k8s worker, result processor and saga orchestrator consumers. ### Server-sent events (backend) `SSERedisBus.listen_execution`, `listen_notifications` and `listen_replay`. -| Metric | Type | Labels | Notes | -|--------|------|--------|-------| -| `sse_connections_active` | gauge, in process | endpoint: executions, notifications, replay | | -| `sse_connection_duration_seconds` | histogram | endpoint | recorded when the stream closes | -| `sse_messages_sent_total` | counter | endpoint, event_type | each message relayed from Redis; `event_type` is the execution event type, `notification` or `replay_status` | +| Metric | Type | Labels | Recorded from | +|--------|------|--------|---------------| +| `sse_connections_active` | gauge | sse_endpoint: executions, notifications, replay | span `sse.streams`, one per endpoint with zeros, from the API's job every 10 s: the streams that process has open | +| `sse_connection_duration_seconds` | histogram | sse_endpoint | `sse.duration_seconds` on span `sse.stream_closed`, when the stream closes | +| `sse_messages_sent_total` | counter | sse_endpoint, sse_event_type | span `sse.message` for each message relayed from Redis; the event type is the execution event type, `notification` or `replay_status` | The status snapshot a stream sends first comes from the database, not Redis, and isn't counted. diff --git a/docs/testing/contract-testing.md b/docs/testing/contract-testing.md index b54a77ec..f2821c6b 100644 --- a/docs/testing/contract-testing.md +++ b/docs/testing/contract-testing.md @@ -1,10 +1,10 @@ # Contract testing -The Grafana dashboards and the Python metric code name the same series, but nothing ties them together at runtime. A renamed metric leaves a dead panel, and a new metric nobody charts goes unnoticed. [`tests/contract/test_grafana_metrics.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/tests/contract/test_grafana_metrics.py) checks both directions without running the stack. +The Grafana dashboards and the metric definitions name the same series, but nothing ties them together at runtime. A renamed metric leaves a dead panel, and a new metric nobody charts goes unnoticed. [`tests/contract/test_grafana_metrics.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/tests/contract/test_grafana_metrics.py) checks both directions without running the stack. ## How it gets the real names -A module-scoped fixture sets up OTel `MeterProvider`s with a `PrometheusMetricReader`, so the SDK does the name conversion instead of the test guessing it. It imports every module in `app.core.metrics` and instantiates every `BaseMetrics` subclass on one provider. On the same provider it creates an instrument for each metric the collector's `signaltometrics` connector makes from spans, with the name, unit and kind from `otel-collector-config.yaml`, since those replace app metrics and must be charted too. The libraries go on a second one, set up the way the app sets them up: FastStream's `KafkaTelemetryMiddleware`, the FastAPI instrumentation on a bare app whose middleware stack it builds (that's when the instrumentation creates its instruments), and the system metrics instrumentation with the app's `PROCESS_METRICS`. Then it gives every instrument one data point: observable instruments get a stub callback, and synchronous ones get whichever of `add`, `record` or `set` they have. The result is two maps from Prometheus family name to sample names (`_total`, `_bucket`, `_count`, `_sum` and so on), one for app metrics and one for the libraries'. +A module-scoped fixture sets up OTel `MeterProvider`s with a `PrometheusMetricReader`, so the SDK does the name conversion instead of the test guessing it. On one provider it creates an instrument for each metric the collector's `signaltometrics` connector makes from spans, with the name, unit and kind from `otel-collector-config.yaml`; these are all the app's metrics. The libraries go on a second one, set up the way the app sets them up: FastStream's `KafkaTelemetryMiddleware`, the FastAPI instrumentation on a bare app whose middleware stack it builds (that's when the instrumentation creates its instruments), and the system metrics instrumentation with the app's `PROCESS_METRICS`. Then it gives every instrument one data point: observable instruments get a stub callback, and synchronous ones get whichever of `add`, `record` or `set` they have. The result is two maps from Prometheus family name to sample names (`_total`, `_bucket`, `_count`, `_sum` and so on), one for app metrics and one for the libraries'. Family names follow the collector, not the Python exporter. The Python exporter appends the unit unless the name already ends with it; the collector's translator skips the unit when it's any token of the name. So an up-down counter named `jobs_active` with unit `jobs` stays `jobs_active` in Victoria Metrics, while the Python exporter alone would call it `jobs_active_jobs`. `_collector_name` applies the collector's rule, and dashboards must use the collector's name. @@ -33,4 +33,4 @@ The pytest-env setting `OTEL_SDK_DISABLED=true` is popped for the fixture's life ## Adding a metric -Add the instrument to a `BaseMetrics` subclass and the reverse check fails until a dashboard panel or alert rule uses it. Add a panel for a series that doesn't exist and the forward check fails, naming the dashboard file and the metric. A new collector receiver needs its prefix added to `COLLECTOR_PREFIXES` before dashboards can use its series. +Define a metric under `signaltometrics` in `otel-collector-config.yaml` and the reverse check fails until a dashboard panel or alert rule uses it. Add a panel for a series that doesn't exist and the forward check fails, naming the dashboard file and the metric. A new collector receiver needs its prefix added to `COLLECTOR_PREFIXES` before dashboards can use its series.