Integrate loguru (#4991)

Author: itallix

application/backend/.gitignore

@@ -17,3 +17,4 @@ data/
 **/openapi.json
 .idea
 .DS_Store
+logs/

application/backend/app/api/dependencies.py

@@ -133,6 +133,11 @@ def get_data_dir(request: Request) -> Path:
     return request.app.state.settings.data_dir
 
 
+def get_job_dir(request: Request) -> Path:
+    """Provides the job log directory path from settings."""
+    return request.app.state.settings.job_dir
+
+
 def get_event_bus(request: Request) -> EventBus:
     """Provides an EventBus instance."""
     return request.app.state.event_bus

application/backend/app/api/routers/jobs.py

@@ -3,12 +3,15 @@
 
 import asyncio
 from collections.abc import AsyncGenerator
+from pathlib import Path
 from typing import Annotated
+from uuid import UUID
 
-from fastapi import APIRouter, Body, Depends, HTTPException, Request, status
-from starlette.responses import StreamingResponse
+import aiofiles
+from fastapi import APIRouter, Body, Depends, HTTPException, status
+from sse_starlette.sse import EventSourceResponse, ServerSentEvent
 
-from app.api.dependencies import get_job_queue, get_project_service
+from app.api.dependencies import get_data_dir, get_job_dir, get_job_queue, get_project_service
 from app.api.validators import JobID
 from app.core.jobs.control_plane import CancellationResult, JobQueue
 from app.core.jobs.models import JobStatus
@@ -33,6 +36,8 @@
 async def submit_job(
     job_request: Annotated[JobRequest, Body()],
     job_queue: Annotated[JobQueue, Depends(get_job_queue)],
+    job_dir: Annotated[Path, Depends(get_job_dir)],
+    data_dir: Annotated[Path, Depends(get_data_dir)],
     project_service: Annotated[ProjectService, Depends(get_project_service)],
 ) -> JobView:
     """
@@ -41,6 +46,8 @@ async def submit_job(
     Args:
         job_request (JobRequest): The Job request payload.
         job_queue (JobQueue): The job queue instance responsible for managing job submissions and tracking job statuses.
+        job_dir (Path): The directory where job log files are stored.
+        data_dir (Path): The base directory for project data storage.
         project_service (ProjectService): The service to interact with project data.
 
     Returns:
@@ -53,6 +60,8 @@ async def submit_job(
             case JobType.TRAIN:
                 job = TrainingJob(
                     project_id=job_request.project_id,
+                    log_dir=job_dir,
+                    data_dir=data_dir,
                     params=TrainingParams(
                         model_architecture_id=job_request.parameters.model_architecture_id,
                         parent_model_revision_id=job_request.parameters.parent_model_revision_id,
@@ -163,49 +172,106 @@ async def cancel_job(job_id: JobID, job_queue: Annotated[JobQueue, Depends(get_j
 
 @router.get("/{job_id}/status")
 async def stream_job_status(
-    job_id: JobID, request: Request, job_queue: Annotated[JobQueue, Depends(get_job_queue)]
-) -> StreamingResponse:
+    job_id: JobID, job_queue: Annotated[JobQueue, Depends(get_job_queue)]
+) -> EventSourceResponse:
     """
     Stream real-time status updates for a specific job.
 
     This endpoint streams job status updates using Server-Sent Events (SSE).
-    It sends periodic updates until the client disconnects or the job reaches
-    terminal state.
+    It sends periodic updates until the job reaches terminal state.
 
     Args:
         job_id (JobID): The unique identifier of the job.
-        request (Request): The HTTP request object to monitor client connection status.
-        job_queue (JobQueue): The job queue instance responsible for managing job submissions and tracking job statuses.
+        job_queue (JobQueue): The job queue instance responsible for tracking job statuses.
 
     Returns:
-        StreamingResponse: A streaming response with job status updates.
+        EventSourceResponse: A streaming response with job status updates.
     """
     if not job_queue.get(job_id):
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found")
 
-    async def gen_job_updates() -> AsyncGenerator[str]:
-        """Generate job status updates."""
-        last = None
-        while True:
-            if await request.is_disconnected():
-                break
-            j = job_queue.get(job_id)
-            if not j:
-                break
-            snap = JobView.of(j).model_dump_json()
-            if snap != last:
-                yield f"{snap}\n"
-                last = snap
-            if j.status >= JobStatus.DONE:
-                break
-            await asyncio.sleep(0.1)
-
-    return StreamingResponse(
-        gen_job_updates(),
-        media_type="text/event-stream",
-        headers={
-            "Content-Type": "text/event-stream",
-            "Connection": "keep-alive",
-            "Cache-Control": "no-cache",
-        },
-    )
+    return EventSourceResponse(__gen_job_updates(job_id, job_queue))
+
+
+@router.get("/{job_id}/logs")
+async def stream_job_logs(
+    job_id: JobID,
+    job_dir: Annotated[Path, Depends(get_job_dir)],
+    job_queue: Annotated[JobQueue, Depends(get_job_queue)],
+) -> EventSourceResponse:
+    """
+    Stream real-time log output for a specific job.
+
+    This endpoint streams job logs using Server-Sent Events (SSE). It reads
+    the job's log file and yields new lines as they are written, allowing clients
+    to follow the job's progress in real-time. The stream continues until the
+    client disconnects or an error occurs.
+
+    Args:
+        job_id (JobID): The unique identifier of the job.
+        job_dir (Path): The directory where job log files are stored.
+        job_queue (JobQueue): The job queue instance for tracking job statuses.
+
+    Returns:
+        EventSourceResponse: A streaming response with log entries sent as SSE events.
+
+    Raises:
+        HTTPException: If the job is not found (404), the log file doesn't exist (404),
+                       or the job has already completed (409).
+    """
+    job = job_queue.get(job_id)
+    if not job:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found")
+
+    if job.status >= JobStatus.DONE:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Job has already completed; logs are no longer available for streaming",
+        )
+
+    log_path = job_dir / job.log_file
+
+    if not log_path.exists():
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Log file not found")
+
+    return EventSourceResponse(__gen_log_stream(job_id, log_path, job_queue))
+
+
+async def __gen_job_updates(job_id: UUID, job_queue: JobQueue) -> AsyncGenerator[ServerSentEvent]:
+    """Generate job status updates."""
+    last = None
+    while True:
+        j = job_queue.get(job_id)
+        if not j:
+            break
+        snap = JobView.of(j).model_dump_json()
+        if snap != last:
+            yield ServerSentEvent(data=snap)
+            last = snap
+        if j.status >= JobStatus.DONE:
+            break
+        await asyncio.sleep(0.1)
+
+
+async def __gen_log_stream(job_id: UUID, log_path: Path, job_queue: JobQueue) -> AsyncGenerator[ServerSentEvent]:
+    """Asynchronously follow a log file and yield new lines as SSE events."""
+    try:
+        async with aiofiles.open(log_path) as f:
+            async for line in f:
+                yield ServerSentEvent(data=line.rstrip("\n"))
+
+            while True:
+                j = job_queue.get(job_id)
+                if not j:
+                    break
+                line = await f.readline()
+                if not line:
+                    await asyncio.sleep(0.3)
+                    continue
+                yield ServerSentEvent(data=line.rstrip("\n"))
+                if j.status >= JobStatus.DONE:
+                    break
+    except asyncio.CancelledError:
+        raise
+    except Exception as e:
+        yield ServerSentEvent(data=f"Error reading log file: {e}")

application/backend/app/api/routers/model_architectures.py

@@ -3,15 +3,12 @@
 
 """Endpoints for managing model architectures"""
 
-import logging
-
 from fastapi import APIRouter, status
 
 from app.schemas import ModelArchitectures
 from app.schemas.model_architecture import ModelArchitecture
 from app.supported_models import SupportedModels
 
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/model_architectures", tags=["Model Architectures"])
 
 

application/backend/app/api/routers/pipelines.py

@@ -3,7 +3,6 @@
 
 """Endpoints for managing pipelines"""
 
-import logging
 from typing import Annotated
 from uuid import UUID
 
@@ -17,7 +16,6 @@
 from app.schemas.pipeline import DataCollectionPolicyAdapter, PipelineStatus, PipelineView
 from app.services import PipelineMetricsService, PipelineService, ResourceNotFoundError
 
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/projects/{project_id}/pipeline", tags=["Pipelines"])
 
 UPDATE_PIPELINE_BODY_DESCRIPTION = """

application/backend/app/api/routers/projects.py

@@ -3,7 +3,6 @@
 
 """Endpoints for managing projects"""
 
-import logging
 from typing import Annotated
 from uuid import UUID
 
@@ -25,7 +24,6 @@
 from app.services.label_service import DuplicateLabelsError
 from app.supported_models.hyperparameters import Hyperparameters
 
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/projects", tags=["Projects"])
 
 CREATE_PROJECT_BODY_DESCRIPTION = """

application/backend/app/api/routers/sinks.py

@@ -3,7 +3,6 @@
 
 """Endpoints for managing pipeline sinks"""
 
-import logging
 from typing import Annotated
 
 import yaml
@@ -24,7 +23,6 @@
     SinkService,
 )
 
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/sinks", tags=["Sinks"])
 
 CREATE_SINK_BODY_DESCRIPTION = """

application/backend/app/api/routers/sources.py

@@ -3,7 +3,6 @@
 
 """Endpoints for managing pipeline sources"""
 
-import logging
 from typing import Annotated
 
 import yaml
@@ -24,7 +23,6 @@
     SourceUpdateService,
 )
 
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/sources", tags=["Sources"])
 
 CREATE_SOURCE_BODY_DESCRIPTION = """

application/backend/app/api/routers/system.py

@@ -3,15 +3,13 @@
 
 """System API Endpoints"""
 
-import logging
 from typing import Annotated
 
 from fastapi import APIRouter, Depends
 
 from app.api.dependencies import get_system_service
 from app.services import SystemService
 
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api")
 
 

application/backend/app/api/routers/webrtc.py

@@ -3,17 +3,16 @@
 
 """WebRTC API Endpoints"""
 
-import logging
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, status
 from fastapi.exceptions import HTTPException
+from loguru import logger
 
 from app.api.dependencies import get_webrtc_manager as get_webrtc
 from app.schemas.webrtc import Answer, InputData, Offer
 from app.webrtc.manager import WebRTCManager
 
-logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/webrtc", tags=["WebRTC"])
 
 
@@ -30,7 +29,7 @@ async def create_webrtc_offer(offer: Offer, webrtc_manager: Annotated[WebRTCMana
     try:
         return await webrtc_manager.handle_offer(offer)
     except Exception as e:
-        logger.error("Error processing WebRTC offer: %s", e)
+        logger.exception("Error processing WebRTC offer")
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))