Merge pull request #417 from aibtcdev/logging-json

add logging json

Author: human058382928

app/lib/logger.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 from typing import Optional
@@ -12,6 +13,51 @@
 }
 
 
+class JSONFormatter(logging.Formatter):
+    """JSON formatter that outputs single-line structured logs."""
+
+    def format(self, record):
+        log_entry = {
+            "timestamp": self.formatTime(record, self.datefmt),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+        }
+
+        # Add exception info if present
+        if record.exc_info:
+            log_entry["exception"] = self.formatException(record.exc_info)
+
+        # Add extra fields if present
+        for key, value in record.__dict__.items():
+            if key not in [
+                "name",
+                "msg",
+                "args",
+                "levelname",
+                "levelno",
+                "pathname",
+                "filename",
+                "module",
+                "exc_info",
+                "exc_text",
+                "stack_info",
+                "lineno",
+                "funcName",
+                "created",
+                "msecs",
+                "relativeCreated",
+                "thread",
+                "threadName",
+                "processName",
+                "process",
+                "message",
+            ]:
+                log_entry[key] = value
+
+        return json.dumps(log_entry, ensure_ascii=False)
+
+
 def configure_logger(name: Optional[str] = None) -> logging.Logger:
     """
     Configure and return a logger instance with consistent formatting and level.
@@ -34,37 +80,37 @@ def configure_logger(name: Optional[str] = None) -> logging.Logger:
     if not logger.handlers:
         console_handler = logging.StreamHandler()
         console_handler.setLevel(log_level)
-        formatter = logging.Formatter("%(levelname)s:     %(message)s")
+        formatter = JSONFormatter()
         console_handler.setFormatter(formatter)
         logger.addHandler(console_handler)
 
     return logger
 
 
 def setup_uvicorn_logging():
-    """Configure uvicorn and other loggers to use normal formatting."""
+    """Configure uvicorn and other loggers to use JSON formatting."""
     # Disable uvicorn access logging since we handle it with middleware
     logging.getLogger("uvicorn.access").disabled = True
 
-    # Create a standard formatter
-    standard_formatter = logging.Formatter("%(levelname)s:     %(message)s")
+    # Create a JSON formatter
+    json_formatter = JSONFormatter()
 
     # Get all existing loggers and configure them
     for logger_name in ["uvicorn", "uvicorn.error", "fastapi"]:
         logger = logging.getLogger(logger_name)
         for handler in logger.handlers:
-            handler.setFormatter(standard_formatter)
+            handler.setFormatter(json_formatter)
 
     # Configure root logger
     root_logger = logging.getLogger()
     for handler in root_logger.handlers:
-        handler.setFormatter(standard_formatter)
+        handler.setFormatter(json_formatter)
 
     # Set up a hook to catch any new handlers that get added later
     original_add_handler = logging.Logger.addHandler
 
     def patched_add_handler(self, hdlr):
-        hdlr.setFormatter(standard_formatter)
+        hdlr.setFormatter(json_formatter)
         return original_add_handler(self, hdlr)
 
     logging.Logger.addHandler = patched_add_handler

app/middleware/logging.py

@@ -37,9 +37,14 @@ async def dispatch(self, request: Request, call_next):
             "process_time_ms": round(process_time * 1000, 2),
         }
 
-        # Log the request/response
+        # Log the request/response with structured data
         logger.info(
-            f"{request_info['method']} {request_info['path']} - {response_info['status_code']} - {response_info['process_time_ms']}ms"
+            f"HTTP {request_info['method']} {request_info['path']} - {response_info['status_code']}",
+            extra={
+                "request": request_info,
+                "response": response_info,
+                "event_type": "http_request",
+            },
         )
 
         return response

app/services/infrastructure/job_management/job_manager.py

@@ -123,6 +123,7 @@ def _get_job_interval(self, job_type, metadata: JobMetadata) -> int:
 
     async def _execute_job_via_executor(self, job_type: str) -> None:
         """Execute a job through the enhanced executor system with proper concurrency control."""
+        logger.info(f"🚀 Scheduled execution triggered for job type: {job_type}")
         try:
             from app.backend.models import QueueMessage, QueueMessageType
 
@@ -131,6 +132,7 @@ async def _execute_job_via_executor(self, job_type: str) -> None:
 
             # Convert job_type string to JobType enum
             job_type_enum = JobType.get_or_create(job_type)
+            logger.debug(f"Converted job type '{job_type}' to enum: {job_type_enum}")
 
             logger.debug(f"🔍 Checking if {job_type} has work to do...")
 
@@ -172,21 +174,22 @@ async def _execute_job_via_executor(self, job_type: str) -> None:
                     "triggered_at": str(datetime.now()),
                 },
                 dao_id=None,
-                tweet_id=None,
-                conversation_id=None,
+                wallet_id=None,
                 is_processed=False,
                 result=None,
                 created_at=datetime.now(),
-                updated_at=datetime.now(),
             )
 
             # Enqueue the synthetic message with the job's priority
+            logger.debug(
+                f"Enqueuing job {job_type} to executor with priority {metadata.priority}"
+            )
             job_id = await self._executor.priority_queue.enqueue(
                 synthetic_message, metadata.priority
             )
 
             logger.info(
-                f"✅ Queued {job_type} job {job_id} for execution (priority: {metadata.priority})"
+                f"✅ Enqueued scheduled job '{job_type}' with ID {job_id} (priority: {metadata.priority})"
             )
 
         except Exception as e:

app/services/infrastructure/startup_service.py

@@ -91,19 +91,34 @@ async def start_enhanced_job_system(self) -> None:
             raise RuntimeError("Job system initialization failed")
 
         # Schedule jobs with the scheduler
+        logger.info("Attempting to schedule jobs with the job manager...")
         any_jobs_scheduled = self.job_manager.schedule_jobs(self.scheduler)
         if any_jobs_scheduled:
             # Start the scheduler
+            logger.info("Starting APScheduler with scheduled jobs...")
             self.scheduler.start()
             logger.info("Job scheduler started successfully")
+
+            # Log scheduler status
+            jobs = self.scheduler.get_jobs()
+            logger.info(
+                f"APScheduler is running with {len(jobs)} active job schedules:"
+            )
+            for job in jobs:
+                logger.info(f"  - Job ID: {job.id}, next run: {job.next_run_time}")
         else:
-            logger.warning("No jobs were scheduled")
+            logger.warning("No jobs were scheduled - scheduler will not be started")
 
         # Start the job executor
+        logger.info("Starting job executor for background job processing...")
         await self.job_manager.start_executor()
-        logger.info("Enhanced job manager executor started successfully")
+        executor_stats = self.job_manager.get_executor_stats()
+        logger.info(
+            f"Enhanced job executor started with {executor_stats.get('worker_count', 0)} workers"
+        )
 
         # Start system metrics collection
+        logger.info("Starting system metrics monitoring...")
         await system_metrics.start_monitoring()
         logger.info("System metrics monitoring started")