structural logging+doc added

Author: HardMax71

backend/app/api/routes/events.py

@@ -291,8 +291,10 @@ async def delete_event(
         raise HTTPException(status_code=404, detail="Event not found")
 
     logger.warning(
-        f"Event {event_id} deleted by admin {admin.email}",
+        "Event deleted by admin",
         extra={
+            "event_id": event_id,
+            "admin_email": admin.email,
             "event_type": result.event_type,
             "aggregate_id": result.aggregate_id,
             "correlation_id": result.correlation_id,

backend/app/db/repositories/execution_repository.py

@@ -21,9 +21,9 @@ def __init__(self, logger: logging.Logger):
 
     async def create_execution(self, create_data: DomainExecutionCreate) -> DomainExecution:
         doc = ExecutionDocument(**asdict(create_data))
-        self.logger.info(f"Inserting execution {doc.execution_id} into MongoDB")
+        self.logger.info("Inserting execution into MongoDB", extra={"execution_id": doc.execution_id})
         await doc.insert()
-        self.logger.info(f"Inserted execution {doc.execution_id}")
+        self.logger.info("Inserted execution", extra={"execution_id": doc.execution_id})
         return DomainExecution(
             **{
                 **doc.model_dump(exclude={"id"}),
@@ -34,13 +34,13 @@ async def create_execution(self, create_data: DomainExecutionCreate) -> DomainEx
         )
 
     async def get_execution(self, execution_id: str) -> DomainExecution | None:
-        self.logger.info(f"Searching for execution {execution_id} in MongoDB")
+        self.logger.info("Searching for execution in MongoDB", extra={"execution_id": execution_id})
         doc = await ExecutionDocument.find_one({"execution_id": execution_id})
         if not doc:
-            self.logger.warning(f"Execution {execution_id} not found in MongoDB")
+            self.logger.warning("Execution not found in MongoDB", extra={"execution_id": execution_id})
             return None
 
-        self.logger.info(f"Found execution {execution_id} in MongoDB")
+        self.logger.info("Found execution in MongoDB", extra={"execution_id": execution_id})
         return DomainExecution(
             **{
                 **doc.model_dump(exclude={"id"}),
@@ -66,7 +66,7 @@ async def update_execution(self, execution_id: str, update_data: DomainExecution
     async def write_terminal_result(self, result: ExecutionResultDomain) -> bool:
         doc = await ExecutionDocument.find_one({"execution_id": result.execution_id})
         if not doc:
-            self.logger.warning(f"No execution found for {result.execution_id}")
+            self.logger.warning("No execution found", extra={"execution_id": result.execution_id})
             return False
 
         await doc.set(

backend/app/dlq/manager.py

@@ -254,7 +254,7 @@ async def _process_dlq_message(self, message: DLQMessage) -> None:
         # Apply filters
         for filter_func in self._filters:
             if not filter_func(message):
-                self.logger.info(f"Message {message.event_id} filtered out")
+                self.logger.info("Message filtered out", extra={"event_id": message.event_id})
                 return
 
         # Store in MongoDB via Beanie
@@ -371,7 +371,7 @@ async def _retry_message(self, message: DLQMessage) -> None:
         # Trigger after_retry callbacks
         await self._trigger_callbacks("after_retry", message, success=True)
 
-        self.logger.info(f"Successfully retried message {message.event_id}")
+        self.logger.info("Successfully retried message", extra={"event_id": message.event_id})
 
     async def _discard_message(self, message: DLQMessage, reason: str) -> None:
         # Update metrics
@@ -390,7 +390,7 @@ async def _discard_message(self, message: DLQMessage, reason: str) -> None:
         # Trigger callbacks
         await self._trigger_callbacks("on_discard", message, reason)
 
-        self.logger.warning(f"Discarded message {message.event_id} due to {reason}")
+        self.logger.warning("Discarded message", extra={"event_id": message.event_id, "reason": reason})
 
     async def _monitor_dlq(self) -> None:
         while self._running:
@@ -453,12 +453,12 @@ async def _trigger_callbacks(self, event_type: str, *args: Any, **kwargs: Any) -
     async def retry_message_manually(self, event_id: str) -> bool:
         doc = await DLQMessageDocument.find_one({"event_id": event_id})
         if not doc:
-            self.logger.error(f"Message {event_id} not found in DLQ")
+            self.logger.error("Message not found in DLQ", extra={"event_id": event_id})
             return False
 
         # Guard against invalid states
         if doc.status in {DLQMessageStatus.DISCARDED, DLQMessageStatus.RETRIED}:
-            self.logger.info(f"Skipping manual retry for {event_id}: status={doc.status}")
+            self.logger.info("Skipping manual retry", extra={"event_id": event_id, "status": str(doc.status)})
             return False
 
         message = self._doc_to_message(doc)

backend/app/services/event_replay/replay_service.py

@@ -42,7 +42,10 @@ async def create_replay_session(self, config: ReplayConfig) -> str:
         state = ReplaySessionState(session_id=str(uuid4()), config=config)
         self._sessions[state.session_id] = state
 
-        self.logger.info(f"Created replay session {state.session_id} type={config.replay_type} target={config.target}")
+        self.logger.info(
+            "Created replay session",
+            extra={"session_id": state.session_id, "type": config.replay_type, "target": config.target},
+        )
 
         return state.session_id
 
@@ -61,7 +64,7 @@ async def start_replay(self, session_id: str) -> None:
         session.started_at = datetime.now(timezone.utc)
 
         self._metrics.increment_active_replays()
-        self.logger.info(f"Started replay session {session_id}")
+        self.logger.info("Started replay session", extra={"session_id": session_id})
 
     async def _run_replay(self, session: ReplaySessionState) -> None:
         start_time = asyncio.get_event_loop().time()
@@ -97,7 +100,10 @@ async def _prepare_session(self, session: ReplaySessionState) -> None:
         total_count = await self._repository.count_events(session.config.filter)
         session.total_events = min(total_count, session.config.max_events or total_count)
 
-        self.logger.info(f"Replay session {session.session_id} will process {session.total_events} events")
+        self.logger.info(
+            "Replay session will process events",
+            extra={"session_id": session.session_id, "total_events": session.total_events},
+        )
 
     async def _handle_progress_callback(self, session: ReplaySessionState) -> None:
         cb = session.config.get_progress_callback()
@@ -119,15 +125,22 @@ async def _complete_session(self, session: ReplaySessionState, start_time: float
         await self._update_session_in_db(session)
 
         self.logger.info(
-            f"Replay session {session.session_id} completed. "
-            f"Replayed: {session.replayed_events}, "
-            f"Failed: {session.failed_events}, "
-            f"Skipped: {session.skipped_events}, "
-            f"Duration: {duration:.2f}s"
+            "Replay session completed",
+            extra={
+                "session_id": session.session_id,
+                "replayed_events": session.replayed_events,
+                "failed_events": session.failed_events,
+                "skipped_events": session.skipped_events,
+                "duration_seconds": round(duration, 2),
+            },
         )
 
     async def _handle_session_error(self, session: ReplaySessionState, error: Exception) -> None:
-        self.logger.error(f"Replay session {session.session_id} failed: {error}", exc_info=True)
+        self.logger.error(
+            "Replay session failed",
+            extra={"session_id": session.session_id, "error": str(error)},
+            exc_info=True,
+        )
         session.status = ReplayStatus.FAILED
         session.completed_at = datetime.now(timezone.utc)
         session.errors.append(
@@ -154,7 +167,7 @@ def _update_replay_metrics(self, session: ReplaySessionState, event: BaseEvent,
         self._metrics.record_event_replayed(session.config.replay_type, event.event_type, status)
 
     async def _handle_replay_error(self, session: ReplaySessionState, event: BaseEvent, error: Exception) -> None:
-        self.logger.error(f"Failed to replay event {event.event_id}: {error}")
+        self.logger.error("Failed to replay event", extra={"event_id": event.event_id, "error": str(error)})
         session.failed_events += 1
         session.errors.append(
             {"timestamp": datetime.now(timezone.utc).isoformat(), "event_id": str(event.event_id), "error": str(error)}
@@ -184,7 +197,7 @@ async def _replay_to_file(self, event: BaseEvent, file_path: str | None) -> bool
         return True
 
     async def _fetch_event_batches(self, session: ReplaySessionState) -> AsyncIterator[List[BaseEvent]]:
-        self.logger.info(f"Fetching events for session {session.session_id}")
+        self.logger.info("Fetching events for session", extra={"session_id": session.session_id})
         events_processed = 0
         max_events = session.config.max_events
 
@@ -250,10 +263,13 @@ async def _replay_event(self, session: ReplaySessionState, event: BaseEvent) ->
                 elif config.target == ReplayTarget.TEST:
                     return True
                 else:
-                    self.logger.error(f"Unknown replay target: {config.target}")
+                    self.logger.error("Unknown replay target", extra={"target": config.target})
                     return False
             except Exception as e:
-                self.logger.error(f"Failed to replay event (attempt {attempt + 1}/{attempts}): {e}")
+                self.logger.error(
+                    "Failed to replay event",
+                    extra={"attempt": attempt + 1, "max_attempts": attempts, "error": str(e)},
+                )
                 if attempt < attempts - 1:
                     await asyncio.sleep(min(2**attempt, 10))
                     continue
@@ -279,7 +295,7 @@ async def pause_replay(self, session_id: str) -> None:
 
         if session.status == ReplayStatus.RUNNING:
             session.status = ReplayStatus.PAUSED
-            self.logger.info(f"Paused replay session {session_id}")
+            self.logger.info("Paused replay session", extra={"session_id": session_id})
 
     async def resume_replay(self, session_id: str) -> None:
         session = self._sessions.get(session_id)
@@ -288,7 +304,7 @@ async def resume_replay(self, session_id: str) -> None:
 
         if session.status == ReplayStatus.PAUSED:
             session.status = ReplayStatus.RUNNING
-            self.logger.info(f"Resumed replay session {session_id}")
+            self.logger.info("Resumed replay session", extra={"session_id": session_id})
 
     async def cancel_replay(self, session_id: str) -> None:
         session = self._sessions.get(session_id)
@@ -301,7 +317,7 @@ async def cancel_replay(self, session_id: str) -> None:
         if task and not task.done():
             task.cancel()
 
-        self.logger.info(f"Cancelled replay session {session_id}")
+        self.logger.info("Cancelled replay session", extra={"session_id": session_id})
 
     def get_session(self, session_id: str) -> ReplaySessionState | None:
         return self._sessions.get(session_id)
@@ -330,7 +346,7 @@ async def cleanup_old_sessions(self, older_than_hours: int = 24) -> int:
                 del self._sessions[session_id]
                 removed += 1
 
-        self.logger.info(f"Cleaned up {removed} old replay sessions")
+        self.logger.info("Cleaned up old replay sessions", extra={"removed_count": removed})
         return removed
 
     async def _update_session_in_db(self, session: ReplaySessionState) -> None:

backend/app/services/execution_service.py

@@ -439,7 +439,7 @@ async def delete_execution(self, execution_id: str) -> bool:
         deleted = await self.execution_repo.delete_execution(execution_id)
 
         if not deleted:
-            self.logger.warning(f"Execution {execution_id} not found for deletion")
+            self.logger.warning("Execution not found for deletion", extra={"execution_id": execution_id})
             raise ExecutionNotFoundError(execution_id)
 
         self.logger.info("Deleted execution", extra={"execution_id": execution_id})

backend/app/services/saga/saga_service.py

@@ -48,24 +48,32 @@ async def check_execution_access(self, execution_id: str, user: User) -> bool:
             return True
 
         self.logger.debug(
-            f"Access denied for user {user.user_id} to execution {execution_id}",
-            extra={"user_role": user.role, "execution_exists": execution is not None},
+            "Access denied to execution",
+            extra={
+                "user_id": user.user_id,
+                "execution_id": execution_id,
+                "user_role": user.role,
+                "execution_exists": execution is not None,
+            },
         )
         return False
 
     async def get_saga_with_access_check(self, saga_id: str, user: User) -> Saga:
         """Get saga with access control."""
-        self.logger.debug(f"Getting saga {saga_id} for user {user.user_id}", extra={"user_role": user.role})
+        self.logger.debug(
+            "Getting saga for user", extra={"saga_id": saga_id, "user_id": user.user_id, "user_role": user.role}
+        )
 
         saga = await self.saga_repo.get_saga(saga_id)
         if not saga:
-            self.logger.warning(f"Saga {saga_id} not found")
+            self.logger.warning("Saga not found", extra={"saga_id": saga_id})
             raise SagaNotFoundError(saga_id)
 
         # Check access permissions
         if not await self.check_execution_access(saga.execution_id, user):
             self.logger.warning(
-                f"Access denied for user {user.user_id} to saga {saga_id}", extra={"execution_id": saga.execution_id}
+                "Access denied to saga",
+                extra={"user_id": user.user_id, "saga_id": saga_id, "execution_id": saga.execution_id},
             )
             raise SagaAccessDeniedError(saga_id, user.user_id)
 
@@ -78,7 +86,8 @@ async def get_execution_sagas(
         # Check access to execution
         if not await self.check_execution_access(execution_id, user):
             self.logger.warning(
-                f"Access denied for user {user.user_id} to execution {execution_id}", extra={"user_role": user.role}
+                "Access denied to execution",
+                extra={"user_id": user.user_id, "execution_id": execution_id, "user_role": user.role},
             )
             raise SagaAccessDeniedError(execution_id, user.user_id)
 
@@ -95,22 +104,31 @@ async def list_user_sagas(
             user_execution_ids = await self.saga_repo.get_user_execution_ids(user.user_id)
             saga_filter.execution_ids = user_execution_ids
             self.logger.debug(
-                f"Filtering sagas for user {user.user_id}",
-                extra={"execution_count": len(user_execution_ids) if user_execution_ids else 0},
+                "Filtering sagas for user",
+                extra={
+                    "user_id": user.user_id,
+                    "execution_count": len(user_execution_ids) if user_execution_ids else 0,
+                },
             )
 
         # Get sagas from repository
         result = await self.saga_repo.list_sagas(saga_filter, limit, skip)
         self.logger.debug(
-            f"Listed {len(result.sagas)} sagas for user {user.user_id}",
-            extra={"total": result.total, "state_filter": str(state) if state else None},
+            "Listed sagas for user",
+            extra={
+                "user_id": user.user_id,
+                "count": len(result.sagas),
+                "total": result.total,
+                "state_filter": str(state) if state else None,
+            },
         )
         return result  # type: ignore[return-value]
 
     async def cancel_saga(self, saga_id: str, user: User) -> bool:
         """Cancel a saga with permission check."""
         self.logger.info(
-            f"User {user.user_id} requesting cancellation of saga {saga_id}", extra={"user_role": user.role}
+            "User requesting saga cancellation",
+            extra={"user_id": user.user_id, "saga_id": saga_id, "user_role": user.role},
         )
         # Get saga with access check
         saga = await self.get_saga_with_access_check(saga_id, user)
@@ -123,10 +141,11 @@ async def cancel_saga(self, saga_id: str, user: User) -> bool:
         success = await self.orchestrator.cancel_saga(saga_id)
         if success:
             self.logger.info(
-                f"User {user.user_id} cancelled saga {saga_id}", extra={"user_role": user.role, "saga_id": saga_id}
+                "User cancelled saga",
+                extra={"user_id": user.user_id, "saga_id": saga_id, "user_role": user.role},
             )
         else:
-            self.logger.error(f"Failed to cancel saga {saga_id} for user {user.user_id}", extra={"saga_id": saga_id})
+            self.logger.error("Failed to cancel saga", extra={"saga_id": saga_id, "user_id": user.user_id})
         return success
 
     async def get_saga_statistics(self, user: User, include_all: bool = False) -> dict[str, object]:
@@ -142,22 +161,22 @@ async def get_saga_statistics(self, user: User, include_all: bool = False) -> di
 
     async def get_saga_status_from_orchestrator(self, saga_id: str, user: User) -> Saga | None:
         """Get saga status from orchestrator with fallback to database."""
-        self.logger.debug(f"Getting live saga status for {saga_id}")
+        self.logger.debug("Getting live saga status", extra={"saga_id": saga_id})
 
         # Try orchestrator first for live status
         saga = await self.orchestrator.get_saga_status(saga_id)
         if saga:
             # Check access
             if not await self.check_execution_access(saga.execution_id, user):
                 self.logger.warning(
-                    f"Access denied for user {user.user_id} to live saga {saga_id}",
-                    extra={"execution_id": saga.execution_id},
+                    "Access denied to live saga",
+                    extra={"user_id": user.user_id, "saga_id": saga_id, "execution_id": saga.execution_id},
                 )
                 raise SagaAccessDeniedError(saga_id, user.user_id)
 
-            self.logger.debug(f"Retrieved live status for saga {saga_id}")
+            self.logger.debug("Retrieved live status for saga", extra={"saga_id": saga_id})
             return saga
 
         # Fall back to repository
-        self.logger.debug(f"No live status found for saga {saga_id}, checking database")
+        self.logger.debug("No live status found for saga, checking database", extra={"saga_id": saga_id})
         return await self.get_saga_with_access_check(saga_id, user)