From a56db99035a7eceb3ec9fd87c8aff8148d18c4c3 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Mon, 19 Oct 2026 01:34:08 +0200 Subject: [PATCH] refactor: replay metrics come from spans, and ReplayMetrics is gone EventReplayService leaves a span where it used to record each metric (replay.session_created, replay.session_started, replay.status_change, replay.event, replay.batch, replay.session_completed, replay.target), and the API's scheduler runs report_sessions every 10 s for the sessions per status and the loaded events still to send. The queue size gauge loses its session_id label (fixes #318), and the active sessions panel sums running and paused from the per-status gauge instead of a separate counter. Part of #389. --- backend/app/core/dishka_lifespan.py | 20 ++- backend/app/core/metrics/__init__.py | 2 - backend/app/core/metrics/replay.py | 130 ---------------- backend/app/core/providers.py | 9 +- .../services/event_replay/replay_service.py | 89 +++++++---- .../provisioning/dashboards/event-replay.json | 10 +- backend/otel-collector-config.yaml | 140 ++++++++++++++++++ backend/tests/unit/conftest.py | 6 - .../services/test_replay_session_metrics.py | 115 ++++++++------ docs/operations/metrics-reference.md | 37 +++-- 10 files changed, 313 insertions(+), 245 deletions(-) delete mode 100644 backend/app/core/metrics/replay.py diff --git a/backend/app/core/dishka_lifespan.py b/backend/app/core/dishka_lifespan.py index 57594a75..c71836c0 100644 --- a/backend/app/core/dishka_lifespan.py +++ b/backend/app/core/dishka_lifespan.py @@ -15,6 +15,7 @@ from app.core.tracing import Tracer from app.db.docs import ALL_DOCUMENTS 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.settings import Settings @@ -64,10 +65,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: await broker.start() logger.info("Kafka broker started") - # Set up APScheduler for NotificationScheduler + # Set up APScheduler for NotificationScheduler and the replay session report notification_scheduler: NotificationScheduler = await container.get(NotificationScheduler) - notification_apscheduler = AsyncIOScheduler() - notification_apscheduler.add_job( + replay_service: EventReplayService = await container.get(EventReplayService) + apscheduler = AsyncIOScheduler() + apscheduler.add_job( notification_scheduler.process_due_notifications, trigger="interval", seconds=15, @@ -75,13 +77,21 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: max_instances=1, misfire_grace_time=60, ) - notification_apscheduler.start() + apscheduler.add_job( + replay_service.report_sessions, + trigger="interval", + seconds=10, + id="report_replay_sessions", + max_instances=1, + misfire_grace_time=30, + ) + apscheduler.start() logger.info("NotificationScheduler started (APScheduler interval=15s)") try: yield finally: - notification_apscheduler.shutdown(wait=False) + apscheduler.shutdown(wait=False) logger.info("NotificationScheduler stopped") await broker.stop() logger.info("Kafka broker stopped") diff --git a/backend/app/core/metrics/__init__.py b/backend/app/core/metrics/__init__.py index 86ac4f77..3f2d08b0 100644 --- a/backend/app/core/metrics/__init__.py +++ b/backend/app/core/metrics/__init__.py @@ -3,7 +3,6 @@ from app.core.metrics.database import IdempotencyMetrics from app.core.metrics.events import EventMetrics from app.core.metrics.rate_limit import RateLimitMetrics -from app.core.metrics.replay import ReplayMetrics from app.core.metrics.security import SecurityMetrics __all__ = [ @@ -12,7 +11,6 @@ "IdempotencyMetrics", "EventMetrics", "RateLimitMetrics", - "ReplayMetrics", "SecurityMetrics", "setup_metrics", ] diff --git a/backend/app/core/metrics/replay.py b/backend/app/core/metrics/replay.py deleted file mode 100644 index 3c46fb6c..00000000 --- a/backend/app/core/metrics/replay.py +++ /dev/null @@ -1,130 +0,0 @@ -from app.core.metrics.base import BaseMetrics -from app.domain.enums import ReplayStatus - - -class ReplayMetrics(BaseMetrics): - """Metrics for event replay operations.""" - - def _create_instruments(self) -> None: - # Core replay metrics - self.replay_sessions_created = self._meter.create_counter( - name="replay.sessions.created.total", description="Total number of replay sessions created", unit="1" - ) - - self.replay_sessions_active = self._meter.create_up_down_counter( - name="replay.sessions.active", description="Number of currently active replay sessions", unit="1" - ) - - self.replay_events_processed = self._meter.create_counter( - name="replay.events.processed.total", description="Total number of events replayed", unit="1" - ) - - self.replay_events_failed = self._meter.create_counter( - name="replay.events.failed.total", description="Total number of failed replay events", unit="1" - ) - - # Performance metrics - self.replay_duration = self._meter.create_histogram( - name="replay.duration", description="Duration of replay sessions in seconds", unit="s" - ) - - self.replay_event_processing_time = self._meter.create_histogram( - name="replay.event.processing.time", - description="Time to process individual replay events in seconds", - unit="s", - ) - - self.replay_throughput = self._meter.create_histogram( - name="replay.throughput", description="Events replayed per second", unit="event/s" - ) - - self.replay_batch_size = self._meter.create_histogram( - name="replay.batch.size", description="Size of replay batches", unit="1" - ) - - # Status tracking - self.replay_status_changes = self._meter.create_counter( - name="replay.status.changes.total", description="Total replay session status changes", unit="1" - ) - - self.replay_sessions_by_status = self._meter.create_up_down_counter( - name="replay.sessions.by.status", description="Number of replay sessions by status", unit="1" - ) - - # Target metrics - self.replay_by_target = self._meter.create_counter( - name="replay.by.target.total", description="Total replays by target type", unit="1" - ) - - self.replay_target_errors = self._meter.create_counter( - name="replay.target.errors.total", description="Errors by replay target", unit="1" - ) - - # Speed control metrics - self.replay_speed_multiplier = self._meter.create_histogram( - name="replay.speed.multiplier", description="Speed multiplier used for replay sessions", unit="x" - ) - - self.replay_delay_applied = self._meter.create_histogram( - name="replay.delay.applied", description="Delay applied between replay events in seconds", unit="s" - ) - - # Queue metrics - self.replay_queue_size = self._meter.create_up_down_counter( - name="replay.queue.size", description="Size of replay event queue", unit="1" - ) - - def record_session_created(self, replay_type: str, target: str) -> None: - self.replay_sessions_created.add(1, attributes={"replay_type": replay_type, "target": target}) - self.update_sessions_by_status(ReplayStatus.CREATED, 1) - - def increment_active_replays(self) -> None: - self.replay_sessions_active.add(1) - - def decrement_active_replays(self) -> None: - self.replay_sessions_active.add(-1) - - def record_event_replayed(self, replay_type: str, event_type: str, success: bool) -> None: - counter = self.replay_events_processed if success else self.replay_events_failed - counter.add(1, attributes={"replay_type": replay_type, "event_type": event_type}) - - def record_replay_duration(self, duration_seconds: float, replay_type: str, total_events: int = 0) -> None: - self.replay_duration.record(duration_seconds, attributes={"replay_type": replay_type}) - - if total_events > 0 and duration_seconds > 0: - throughput = total_events / duration_seconds - self.replay_throughput.record(throughput, attributes={"replay_type": replay_type}) - - def record_event_processing_time(self, duration_seconds: float, event_type: str) -> None: - self.replay_event_processing_time.record(duration_seconds, attributes={"event_type": event_type}) - - def record_status_change(self, session_id: str, from_status: str, to_status: str) -> None: - self.replay_status_changes.add(1, attributes={"from_status": from_status, "to_status": to_status}) - self.update_sessions_by_status(from_status, -1) - self.update_sessions_by_status(to_status, 1) - - def update_sessions_by_status(self, status: str, delta: int) -> None: - if delta != 0: - self.replay_sessions_by_status.add(delta, attributes={"status": status}) - - def record_replay_by_target(self, target: str, success: bool) -> None: - self.replay_by_target.add(1, attributes={"target": target, "success": str(success)}) - if not success: - self.replay_target_errors.add(1, attributes={"target": target}) - - def record_speed_multiplier(self, multiplier: float, replay_type: str) -> None: - self.replay_speed_multiplier.record(multiplier, attributes={"replay_type": replay_type}) - - def record_delay_applied(self, delay_seconds: float) -> None: - self.replay_delay_applied.record(delay_seconds) - - def record_batch_size(self, size: int, replay_type: str) -> None: - self.replay_batch_size.record(size, attributes={"replay_type": replay_type}) - - def update_replay_queue_size(self, session_id: str, size: int) -> None: - key = f"_queue_{session_id}" - current_val = getattr(self, key, 0) - delta = size - current_val - if delta != 0: - self.replay_queue_size.add(delta, attributes={"session_id": session_id}) - setattr(self, key, size) diff --git a/backend/app/core/providers.py b/backend/app/core/providers.py index 530d5727..cd757089 100644 --- a/backend/app/core/providers.py +++ b/backend/app/core/providers.py @@ -17,7 +17,6 @@ EventMetrics, IdempotencyMetrics, RateLimitMetrics, - ReplayMetrics, SecurityMetrics, ) from app.core.security import SecurityService @@ -266,10 +265,6 @@ def get_rate_limit_metrics(self, settings: Settings) -> RateLimitMetrics: def get_idempotency_metrics(self, settings: Settings) -> IdempotencyMetrics: return IdempotencyMetrics(settings) - @provide - def get_replay_metrics(self, settings: Settings) -> ReplayMetrics: - return ReplayMetrics(settings) - @provide def get_security_metrics(self, settings: Settings) -> SecurityMetrics: return SecurityMetrics(settings) @@ -708,14 +703,14 @@ def get_event_replay_service( self, replay_repository: ReplayRepository, kafka_producer: UnifiedProducer, - replay_metrics: ReplayMetrics, + tracer: trace.Tracer, logger: structlog.stdlib.BoundLogger, sse_bus: SSERedisBus, ) -> EventReplayService: return EventReplayService( repository=replay_repository, producer=kafka_producer, - replay_metrics=replay_metrics, + tracer=tracer, logger=logger, sse_bus=sse_bus, ) diff --git a/backend/app/services/event_replay/replay_service.py b/backend/app/services/event_replay/replay_service.py index b03a85a6..167a98f2 100644 --- a/backend/app/services/event_replay/replay_service.py +++ b/backend/app/services/event_replay/replay_service.py @@ -1,6 +1,7 @@ import asyncio import json import time +from collections import Counter from collections.abc import AsyncIterator from datetime import datetime, timedelta, timezone from uuid import uuid4 @@ -8,12 +9,13 @@ import aiofiles import structlog from apscheduler.schedulers.asyncio import AsyncIOScheduler +from opentelemetry import trace +from opentelemetry.util.types import AttributeValue from pydantic import ValidationError from pymongo.errors import PyMongoError from redis.exceptions import RedisError from tenacity import AsyncRetrying, RetryCallState, retry_if_exception_type, stop_after_attempt, wait_exponential -from app.core.metrics import ReplayMetrics from app.db import ReplayRepository from app.domain.admin import ReplaySessionUpdate from app.domain.enums import ReplayStatus, ReplayTarget @@ -39,7 +41,7 @@ def __init__( self, repository: ReplayRepository, producer: UnifiedProducer, - replay_metrics: ReplayMetrics, + tracer: trace.Tracer, logger: structlog.stdlib.BoundLogger, sse_bus: SSERedisBus, ) -> None: @@ -52,7 +54,7 @@ def __init__( self._producer = producer self.logger = logger self._file_locks: dict[str, asyncio.Lock] = {} - self._metrics = replay_metrics + self._tracer = tracer self._sse_bus = sse_bus async def create_session_from_config(self, config: ReplayConfig) -> ReplayOperationResult: @@ -60,7 +62,8 @@ async def create_session_from_config(self, config: ReplayConfig) -> ReplayOperat state = ReplaySessionState(session_id=str(uuid4()), config=config) self._sessions[state.session_id] = state await self._repository.save_session(state) - self._metrics.record_session_created(config.replay_type, config.target) + attributes = {"replay.type": config.replay_type, "replay.target": config.target} + self._tracer.start_span("replay.session_created", attributes=attributes).end() return ReplayOperationResult( session_id=state.session_id, status=ReplayStatus.CREATED, @@ -84,7 +87,7 @@ async def start_session(self, session_id: str) -> ReplayOperationResult: if not await self._load_next_batch(session_id): session.status = ReplayStatus.COMPLETED session.completed_at = datetime.now(timezone.utc) - self._metrics.record_status_change(session_id, ReplayStatus.CREATED, ReplayStatus.COMPLETED) + self._report_status_change(ReplayStatus.CREATED, ReplayStatus.COMPLETED) await self._update_session_in_db(session) return ReplayOperationResult( session_id=session_id, status=ReplayStatus.COMPLETED, message="No events to replay" @@ -105,9 +108,12 @@ async def start_session(self, session_id: str) -> ReplayOperationResult: previous_status = session.status session.status = ReplayStatus.RUNNING session.started_at = datetime.now(timezone.utc) - self._metrics.increment_active_replays() - self._metrics.record_status_change(session_id, previous_status, ReplayStatus.RUNNING) - self._metrics.record_speed_multiplier(session.config.speed_multiplier, session.config.replay_type) + self._report_status_change(previous_status, ReplayStatus.RUNNING) + attributes: dict[str, AttributeValue] = { + "replay.type": session.config.replay_type, + "replay.speed_multiplier": session.config.speed_multiplier, + } + self._tracer.start_span("replay.session_started", attributes=attributes).end() await self._repository.update_session_status(session_id, ReplayStatus.RUNNING) await self._publish_replay_status(session) return ReplayOperationResult( @@ -120,7 +126,7 @@ async def pause_session(self, session_id: str) -> ReplayOperationResult: raise ReplayOperationError(session_id, "pause", "Session is not running") previous_status = session.status session.status = ReplayStatus.PAUSED - self._metrics.record_status_change(session_id, previous_status, ReplayStatus.PAUSED) + self._report_status_change(previous_status, ReplayStatus.PAUSED) scheduler = self._schedulers.get(session_id) if scheduler: scheduler.remove_all_jobs() @@ -133,7 +139,7 @@ async def resume_session(self, session_id: str) -> ReplayOperationResult: raise ReplayOperationError(session_id, "resume", "Session is not paused") previous_status = session.status session.status = ReplayStatus.RUNNING - self._metrics.record_status_change(session_id, previous_status, ReplayStatus.RUNNING) + self._report_status_change(previous_status, ReplayStatus.RUNNING) scheduler = self._schedulers.get(session_id) if scheduler: scheduler.add_job( @@ -178,7 +184,6 @@ async def cleanup_old_sessions(self, older_than_hours: int = 24) -> CleanupResul session = self._sessions[session_id] if session.status.is_terminal and session.created_at < cutoff_time: del self._sessions[session_id] - self._metrics.update_sessions_by_status(session.status, -1) removed_memory += 1 removed_db = await self._repository.delete_old_sessions(cutoff_time) @@ -187,6 +192,33 @@ async def cleanup_old_sessions(self, older_than_hours: int = 24) -> CleanupResul self.logger.info("Cleaned up old replay sessions", removed_count=total_removed) return CleanupResult(removed_sessions=total_removed, message=f"Removed {total_removed} old sessions") + async def report_sessions(self) -> None: + """Report the sessions per status and the loaded events still to send, for the collector's gauges.""" + with self._tracer.start_as_current_span("replay.report") as span: + queued = sum(len(buf) - self._buffer_indices.get(sid, 0) for sid, buf in self._event_buffers.items()) + span.set_attribute("replay.queued_events", queued) + counts = Counter(session.status for session in self._sessions.values()) + for status in ReplayStatus: + attributes: dict[str, AttributeValue] = {"replay.status": status, "replay.sessions": counts[status]} + self._tracer.start_span("replay.sessions", attributes=attributes).end() + + def _report_status_change(self, from_status: ReplayStatus, to_status: ReplayStatus) -> None: + attributes = {"replay.from_status": from_status, "replay.to_status": to_status} + self._tracer.start_span("replay.status_change", attributes=attributes).end() + + def _report_event( + self, session: ReplaySessionState, event: DomainEvent, replayed: bool, seconds: float, delay: float + ) -> None: + attributes: dict[str, AttributeValue] = { + "replay.type": session.config.replay_type, + "replay.event_type": event.event_type, + "replay.result": "replayed" if replayed else "failed", + "replay.processing_seconds": seconds, + } + if delay > 0: + attributes["replay.delay_seconds"] = delay + self._tracer.start_span("replay.event", attributes=attributes).end() + async def _dispatch_next(self, session: ReplaySessionState) -> None: if session.status != ReplayStatus.RUNNING: return @@ -201,34 +233,27 @@ async def _dispatch_next(self, session: ReplaySessionState) -> None: await self._finalize_session(session, ReplayStatus.COMPLETED) return - buf = self._event_buffers.get(session.session_id, []) - idx = self._buffer_indices.get(session.session_id, 0) - self._metrics.update_replay_queue_size(session.session_id, len(buf) - idx) - success = False t0 = time.monotonic() try: success = await self._replay_event(session, event) except Exception as e: processing_time = time.monotonic() - t0 - self._metrics.record_event_processing_time(processing_time, event.event_type) session.errors.append( ReplayError(timestamp=datetime.now(timezone.utc), event_id=str(event.event_id), error=str(e)) ) if not session.config.skip_errors: session.failed_events += 1 - self._metrics.record_event_replayed(session.config.replay_type, event.event_type, False) + self._report_event(session, event, False, processing_time, 0.0) await self._finalize_session(session, ReplayStatus.FAILED) return else: processing_time = time.monotonic() - t0 - self._metrics.record_event_processing_time(processing_time, event.event_type) if success: session.replayed_events += 1 else: session.failed_events += 1 - self._metrics.record_event_replayed(session.config.replay_type, event.event_type, success) session.last_event_at = event.timestamp await self._update_session_in_db(session) @@ -238,8 +263,7 @@ async def _dispatch_next(self, session: ReplaySessionState) -> None: time_diff = (next_event.timestamp - session.last_event_at).total_seconds() delay = max(time_diff / session.config.speed_multiplier, 0) - if delay > 0: - self._metrics.record_delay_applied(delay) + self._report_event(session, event, success, processing_time, delay) scheduler = self._schedulers.get(session.session_id) if scheduler and scheduler.running and session.status == ReplayStatus.RUNNING: @@ -278,8 +302,11 @@ async def _load_next_batch(self, session_id: str) -> bool: self._buffer_indices[session_id] = 0 session = self._sessions.get(session_id) if session: - self._metrics.record_batch_size(len(batch), session.config.replay_type) - self._metrics.update_replay_queue_size(session_id, len(batch)) + attributes: dict[str, AttributeValue] = { + "replay.type": session.config.replay_type, + "replay.batch_size": len(batch), + } + self._tracer.start_span("replay.batch", attributes=attributes).end() return True except StopAsyncIteration: return False @@ -291,11 +318,14 @@ async def _finalize_session(self, session: ReplaySessionState, final_status: Rep if final_status == ReplayStatus.COMPLETED and session.started_at: duration = (session.completed_at - session.started_at).total_seconds() total_events = session.replayed_events + session.failed_events - self._metrics.record_replay_duration(duration, session.config.replay_type, total_events=total_events) - self._metrics.record_status_change(session.session_id, previous_status, final_status) - if previous_status in (ReplayStatus.RUNNING, ReplayStatus.PAUSED): - self._metrics.decrement_active_replays() - self._metrics.update_replay_queue_size(session.session_id, 0) + attributes: dict[str, AttributeValue] = { + "replay.type": session.config.replay_type, + "replay.duration_seconds": duration, + } + if total_events > 0 and duration > 0: + attributes["replay.throughput"] = total_events / duration + self._tracer.start_span("replay.session_completed", attributes=attributes).end() + self._report_status_change(previous_status, final_status) await self._update_session_in_db(session) scheduler = self._schedulers.pop(session.session_id, None) if scheduler: @@ -365,7 +395,8 @@ async def _replay_event(self, session: ReplaySessionState, event: DomainEvent) - return False finally: # Unexpected errors still propagate to _dispatch_next, but count against the target first - self._metrics.record_replay_by_target(config.target, success=success) + attributes: dict[str, AttributeValue] = {"replay.target": config.target, "replay.target_success": success} + self._tracer.start_span("replay.target", attributes=attributes).end() return True def _log_replay_retry(self, state: RetryCallState) -> None: diff --git a/backend/grafana/provisioning/dashboards/event-replay.json b/backend/grafana/provisioning/dashboards/event-replay.json index c99732e0..13b4a55a 100644 --- a/backend/grafana/provisioning/dashboards/event-replay.json +++ b/backend/grafana/provisioning/dashboards/event-replay.json @@ -178,7 +178,7 @@ "pluginVersion": "8.3.3", "targets": [ { - "expr": "replay_sessions_active or vector(0)", + "expr": "sum(replay_sessions_by_status{replay_status=~\"running|paused\"}) or vector(0)", "refId": "A", "datasource": "Victoria Metrics" } @@ -240,10 +240,10 @@ "pluginVersion": "8.3.3", "targets": [ { - "expr": "sum by (status) (replay_sessions_by_status) or vector(0)", + "expr": "sum by (replay_status) (replay_sessions_by_status) or vector(0)", "refId": "A", "datasource": "Victoria Metrics", - "legendFormat": "{{status}}" + "legendFormat": "{{replay_status}}" } ], "title": "By Status", @@ -841,7 +841,7 @@ "targets": [ { "expr": "rate(replay_by_target_total[5m]) or vector(0)", - "legendFormat": "{{target}}", + "legendFormat": "{{replay_target}}", "refId": "A", "datasource": "Victoria Metrics" }, @@ -993,7 +993,7 @@ "pluginVersion": "8.3.3", "targets": [ { - "expr": "replay_queue_size or vector(0)", + "expr": "sum(replay_queue_size) or vector(0)", "refId": "A", "datasource": "Victoria Metrics" } diff --git a/backend/otel-collector-config.yaml b/backend/otel-collector-config.yaml index 09ed405f..3063e00b 100644 --- a/backend/otel-collector-config.yaml +++ b/backend/otel-collector-config.yaml @@ -480,6 +480,146 @@ connectors: sum: value: "1" monotonic: true + - name: replay.sessions.created + description: Replay sessions created + unit: "{session}" + attributes: + - key: replay.type + - key: replay.target + conditions: + - span.name == "replay.session_created" + sum: + value: "1" + monotonic: true + - name: replay.status.changes + description: Replay session status changes + unit: "{change}" + attributes: + - key: replay.from_status + - key: replay.to_status + conditions: + - span.name == "replay.status_change" + sum: + value: "1" + monotonic: true + - name: replay.events.processed + description: Events replayed + unit: "{event}" + attributes: + - key: replay.type + - key: replay.event_type + conditions: + - span.name == "replay.event" and span.attributes["replay.result"] == "replayed" + sum: + value: "1" + monotonic: true + - name: replay.events.failed + description: Events that failed to replay + unit: "{event}" + attributes: + - key: replay.type + - key: replay.event_type + conditions: + - span.name == "replay.event" and span.attributes["replay.result"] == "failed" + sum: + value: "1" + monotonic: true + - name: replay.event.processing.time + description: Time to replay one event, retries included + unit: s + attributes: + - key: replay.event_type + conditions: + - span.name == "replay.event" + histogram: + buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5] + value: Double(span.attributes["replay.processing_seconds"]) + - name: replay.delay.applied + description: Wait before the next event, from the original gap over the speed multiplier + unit: s + conditions: + - span.name == "replay.event" and span.attributes["replay.delay_seconds"] != nil + histogram: + buckets: [0.01, 0.1, 0.5, 1, 5, 10, 30, 60, 300] + value: Double(span.attributes["replay.delay_seconds"]) + - name: replay.batch.size + description: Events loaded per batch + unit: "{event}" + attributes: + - key: replay.type + conditions: + - span.name == "replay.batch" + histogram: + buckets: [1, 10, 50, 100, 250, 500, 1000] + value: Double(span.attributes["replay.batch_size"]) + - name: replay.duration + description: Time from start to completion of a completed session + unit: s + attributes: + - key: replay.type + conditions: + - span.name == "replay.session_completed" + histogram: + buckets: [1, 5, 10, 30, 60, 300, 600, 1800, 3600] + value: Double(span.attributes["replay.duration_seconds"]) + - name: replay.throughput + description: Events per second of a completed session + unit: event/s + attributes: + - key: replay.type + conditions: + - span.name == "replay.session_completed" and span.attributes["replay.throughput"] != nil + histogram: + buckets: [1, 5, 10, 50, 100, 500, 1000, 5000] + value: Double(span.attributes["replay.throughput"]) + - name: replay.speed.multiplier + description: Speed multiplier of each started session + unit: x + attributes: + - key: replay.type + conditions: + - span.name == "replay.session_started" + histogram: + buckets: [0.5, 1, 2, 5, 10, 50, 100] + value: Double(span.attributes["replay.speed_multiplier"]) + - name: replay.by.target + description: Events sent to each replay target, by success + unit: "{event}" + attributes: + - key: replay.target + - key: replay.target_success + conditions: + - span.name == "replay.target" + sum: + value: "1" + monotonic: true + - name: replay.target.errors + description: Events a replay target failed to take + unit: "{event}" + attributes: + - key: replay.target + conditions: + - span.name == "replay.target" and span.attributes["replay.target_success"] == false + sum: + value: "1" + monotonic: true + # The API reports these every 10 s from the sessions it holds in memory + - name: replay.sessions.by.status + description: Replay sessions the API holds, by status + unit: "{session}" + attributes: + - key: replay.status + conditions: + - span.name == "replay.sessions" + gauge: + value: Int(span.attributes["replay.sessions"]) + - name: replay.queue.size + description: Loaded events still to send, over all sessions + unit: "{event}" + conditions: + - span.name == "replay.report" + gauge: + value: Int(span.attributes["replay.queued_events"]) processors: batch: diff --git a/backend/tests/unit/conftest.py b/backend/tests/unit/conftest.py index 4124ab0e..35f3e81d 100644 --- a/backend/tests/unit/conftest.py +++ b/backend/tests/unit/conftest.py @@ -27,7 +27,6 @@ EventMetrics, IdempotencyMetrics, RateLimitMetrics, - ReplayMetrics, SecurityMetrics, ) from app.domain.enums import ExecutionStatus, QueuePriority @@ -243,11 +242,6 @@ def rate_limit_metrics(test_settings: Settings) -> RateLimitMetrics: return RateLimitMetrics(test_settings) -@pytest.fixture -def replay_metrics(test_settings: Settings) -> ReplayMetrics: - return ReplayMetrics(test_settings) - - @pytest.fixture def security_metrics(test_settings: Settings) -> SecurityMetrics: return SecurityMetrics(test_settings) diff --git a/backend/tests/unit/services/test_replay_session_metrics.py b/backend/tests/unit/services/test_replay_session_metrics.py index 35d060f5..6ef85824 100644 --- a/backend/tests/unit/services/test_replay_session_metrics.py +++ b/backend/tests/unit/services/test_replay_session_metrics.py @@ -6,8 +6,9 @@ import pytest import structlog from aiokafka.errors import KafkaError +from opentelemetry import trace +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from app.core.metrics import ReplayMetrics from app.db import ReplayRepository from app.domain.enums import ReplayStatus, ReplayTarget, ReplayType from app.domain.replay import ReplayConfig, ReplaySessionState @@ -25,51 +26,56 @@ async def _no_batches(*_: Any, **__: Any) -> AsyncIterator[list[dict[str, Any]]] yield batch -def _service(metrics: ReplayMetrics, producer: AsyncMock | None = None) -> tuple[EventReplayService, AsyncMock]: +def _service(tracer: trace.Tracer, producer: AsyncMock | None = None) -> EventReplayService: repo = AsyncMock(spec=ReplayRepository) repo.count_events.return_value = 0 repo.fetch_events = MagicMock(side_effect=_no_batches) repo.delete_old_sessions.return_value = 0 - service = EventReplayService( + return EventReplayService( repository=repo, producer=producer or AsyncMock(spec=UnifiedProducer), - replay_metrics=metrics, + tracer=tracer, logger=structlog.get_logger("test.replay_session_metrics"), sse_bus=AsyncMock(spec=SSERedisBus), ) - return service, repo -async def test_cancelling_a_session_that_never_started_leaves_active_alone() -> None: - metrics = MagicMock(spec=ReplayMetrics) - service, _ = _service(metrics) +def _spans(exporter: InMemorySpanExporter) -> list[tuple[str, dict[str, Any]]]: + return [(span.name, dict(span.attributes or {})) for span in exporter.get_finished_spans()] + + +async def test_cancelling_a_session_that_never_started_reports_one_status_change( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + service = _service(tracer) created = await service.create_session_from_config(ReplayConfig(replay_type=ReplayType.EXECUTION)) await service.cancel_session(created.session_id) - metrics.decrement_active_replays.assert_not_called() - metrics.record_status_change.assert_called_once_with( - created.session_id, ReplayStatus.CREATED, ReplayStatus.CANCELLED - ) + assert _spans(span_exporter) == [ + ("replay.session_created", {"replay.type": "execution", "replay.target": "kafka"}), + ("replay.status_change", {"replay.from_status": "created", "replay.to_status": "cancelled"}), + ] -async def test_starting_with_nothing_to_replay_records_the_completion() -> None: - metrics = MagicMock(spec=ReplayMetrics) - service, _ = _service(metrics) +async def test_starting_with_nothing_to_replay_reports_the_completion( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + service = _service(tracer) created = await service.create_session_from_config(ReplayConfig(replay_type=ReplayType.EXECUTION)) result = await service.start_session(created.session_id) assert result.status == ReplayStatus.COMPLETED - metrics.record_status_change.assert_called_once_with( - created.session_id, ReplayStatus.CREATED, ReplayStatus.COMPLETED - ) - metrics.increment_active_replays.assert_not_called() + assert [name for name, _ in _spans(span_exporter)] == ["replay.session_created", "replay.status_change"] + assert _spans(span_exporter)[1][1] == {"replay.from_status": "created", "replay.to_status": "completed"} -async def test_cleanup_takes_removed_sessions_out_of_the_status_gauge() -> None: - metrics = MagicMock(spec=ReplayMetrics) - service, _ = _service(metrics) +async def test_report_counts_held_sessions_by_status_and_events_still_to_send( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + service = _service(tracer) + created = await service.create_session_from_config(ReplayConfig(replay_type=ReplayType.EXECUTION)) old = ReplaySessionState( session_id="old", config=ReplayConfig(replay_type=ReplayType.EXECUTION), @@ -77,35 +83,29 @@ async def test_cleanup_takes_removed_sessions_out_of_the_status_gauge() -> None: created_at=datetime.now(timezone.utc) - timedelta(hours=72), ) service._sessions[old.session_id] = old + service._event_buffers[created.session_id] = [make_execution_requested_event() for _ in range(5)] + service._buffer_indices[created.session_id] = 2 + await service.cleanup_old_sessions(older_than_hours=48) + span_exporter.clear() - result = await service.cleanup_old_sessions(older_than_hours=48) - - assert result.removed_sessions == 1 - metrics.update_sessions_by_status.assert_called_once_with(ReplayStatus.COMPLETED, -1) - + await service.report_sessions() -def test_session_created_counts_towards_created_status() -> None: - metrics = ReplayMetrics(MagicMock()) - metrics.replay_sessions_by_status = MagicMock() - - metrics.record_session_created(ReplayType.EXECUTION, "kafka") - metrics.record_status_change("s1", ReplayStatus.CREATED, ReplayStatus.RUNNING) - - calls = [ - (c.args[0], c.kwargs["attributes"]["status"]) for c in metrics.replay_sessions_by_status.add.call_args_list - ] - assert calls == [(1, ReplayStatus.CREATED), (-1, ReplayStatus.CREATED), (1, ReplayStatus.RUNNING)] + *sessions, (report_name, report) = _spans(span_exporter) + assert (report_name, report) == ("replay.report", {"replay.queued_events": 3}) + counts = {attributes["replay.status"]: attributes["replay.sessions"] for _, attributes in sessions} + assert counts == {status: 1 if status == ReplayStatus.CREATED else 0 for status in ReplayStatus} @pytest.mark.parametrize( ("produce_error", "success"), [(None, True), (KafkaError(), False), (TypeError("bug"), False)], ) -async def test_every_dispatch_outcome_counts_against_the_target(produce_error: Exception | None, success: bool) -> None: - metrics = MagicMock(spec=ReplayMetrics) +async def test_every_dispatch_outcome_counts_against_the_target( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter, produce_error: Exception | None, success: bool +) -> None: producer = AsyncMock(spec=UnifiedProducer) producer.produce.side_effect = produce_error - service, _ = _service(metrics, producer) + service = _service(tracer, producer) session = ReplaySessionState(session_id="s1", config=ReplayConfig(replay_type=ReplayType.EXECUTION)) if isinstance(produce_error, TypeError): @@ -114,4 +114,35 @@ async def test_every_dispatch_outcome_counts_against_the_target(produce_error: E else: assert await service._replay_event(session, make_execution_requested_event()) is success - metrics.record_replay_by_target.assert_called_once_with(ReplayTarget.KAFKA, success=success) + assert _spans(span_exporter) == [("replay.target", {"replay.target": "kafka", "replay.target_success": success})] + + +async def test_dispatch_reports_each_event_and_the_completed_session( + tracer: trace.Tracer, span_exporter: InMemorySpanExporter +) -> None: + service = _service(tracer) + config = ReplayConfig(replay_type=ReplayType.EXECUTION, target=ReplayTarget.TEST, speed_multiplier=2.0) + session = ReplaySessionState( + session_id="s1", + config=config, + status=ReplayStatus.RUNNING, + started_at=datetime.now(timezone.utc) - timedelta(seconds=4), + ) + first, second = make_execution_requested_event(), make_execution_requested_event() + second.timestamp = first.timestamp + timedelta(seconds=2) + service._sessions[session.session_id] = session + service._event_buffers[session.session_id] = [first, second] + + for _ in range(3): + await service._dispatch_next(session) + + spans = _spans(span_exporter) + events = [attributes for name, attributes in spans if name == "replay.event"] + assert [(e["replay.result"], e.get("replay.delay_seconds")) for e in events] == [ + ("replayed", 1.0), + ("replayed", None), + ] + [completed] = [attributes for name, attributes in spans if name == "replay.session_completed"] + assert completed["replay.duration_seconds"] >= 4 + assert completed["replay.throughput"] == pytest.approx(2 / completed["replay.duration_seconds"]) + assert spans[-1] == ("replay.status_change", {"replay.from_status": "running", "replay.to_status": "completed"}) diff --git a/docs/operations/metrics-reference.md b/docs/operations/metrics-reference.md index 320e9786..5167a16b 100644 --- a/docs/operations/metrics-reference.md +++ b/docs/operations/metrics-reference.md @@ -24,7 +24,7 @@ The collector's metrics pipeline (`backend/otel-collector-config.yaml`) reads th ## 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 and notification 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. The execution, pod, queue, dead letter queue, notification and replay metrics work this way so far ([#389](https://github.com/HardMax71/Integr8sCode/issues/389)). ## Names and labels in VictoriaMetrics @@ -135,25 +135,24 @@ The status snapshot a stream sends first comes from the database, not Redis, and ### Event replay (backend) -`EventReplayService`. The replay routes drive it inside backend. A TTL index in MongoDB drops finished sessions, so nothing outside the API touches them. +`EventReplayService`. The replay routes drive it inside backend. A TTL index in MongoDB drops finished sessions, so nothing outside the API touches them. Each step leaves a span where the service used to record a metric, and the API's scheduler runs `report_sessions` every 10 seconds for the two gauges. -| Metric | Type | Labels | Notes | -|--------|------|--------|-------| -| `replay_sessions_created_total` | counter | replay_type, target | | -| `replay_sessions_active` | gauge, in process | none | running or paused sessions | -| `replay_sessions_by_status` | gauge, in process | status | sessions held in memory | -| `replay_status_changes_total` | counter | from_status, to_status | | -| `replay_events_processed_total` | counter | replay_type, event_type | | -| `replay_events_failed_total` | counter | replay_type, event_type | | -| `replay_event_processing_time_seconds` | histogram | event_type | dispatching one event | -| `replay_duration_seconds` | histogram | replay_type | completed sessions only | -| `replay_throughput_event_per_second` | histogram | replay_type | events over duration, completed sessions only | -| `replay_batch_size` | histogram | replay_type | | -| `replay_by_target_total` | counter | target, success | | -| `replay_target_errors_total` | counter | target | | -| `replay_speed_multiplier_x` | histogram | replay_type | one sample per session start | -| `replay_delay_applied_seconds` | histogram | none | one sample per delayed event | -| `replay_queue_size` | gauge, in process | session_id | events left in the loaded batch | +| Metric | Type | Labels | Recorded from | +|--------|------|--------|---------------| +| `replay_sessions_created_total` | counter | replay_type, replay_target | span `replay.session_created` | +| `replay_sessions_by_status` | gauge | replay_status | span `replay.sessions`, one per status with zeros, every 10 s: sessions the process holds in memory. Running plus paused is the active count | +| `replay_status_changes_total` | counter | replay_from_status, replay_to_status | span `replay.status_change` | +| `replay_events_processed_total` | counter | replay_type, replay_event_type | span `replay.event` with `replay.result` `replayed` | +| `replay_events_failed_total` | counter | replay_type, replay_event_type | the same span with `replay.result` `failed` | +| `replay_event_processing_time_seconds` | histogram | replay_event_type | `replay.processing_seconds` on `replay.event`: dispatching one event | +| `replay_duration_seconds` | histogram | replay_type | span `replay.session_completed`, completed sessions only | +| `replay_throughput_event_per_second` | histogram | replay_type | events over duration on the same span | +| `replay_batch_size` | histogram | replay_type | span `replay.batch` | +| `replay_by_target_total` | counter | replay_target, replay_target_success | span `replay.target` | +| `replay_target_errors_total` | counter | replay_target | the same span when the target failed | +| `replay_speed_multiplier_x` | histogram | replay_type | span `replay.session_started`, one sample per session start | +| `replay_delay_applied_seconds` | histogram | none | `replay.delay_seconds` on `replay.event`, one sample per delayed event | +| `replay_queue_size` | gauge | none | `replay.queued_events` on span `replay.report`, every 10 s: loaded events still to send, over all sessions | ### Authentication and authorization (backend)