Logging updates (#2159)

backend/danswer/main.py

@@ -110,6 +110,7 @@ from danswer.tools.built_in_tools import auto_add_search_tool_to_personas
 from danswer.tools.built_in_tools import load_builtin_tools
 from danswer.tools.built_in_tools import refresh_built_in_tools_cache
 from danswer.utils.logger import setup_logger
+from danswer.utils.logger import setup_uvicorn_logger
 from danswer.utils.telemetry import optional_telemetry
 from danswer.utils.telemetry import RecordType
 from danswer.utils.variable_functionality import fetch_versioned_implementation
@@ -125,6 +126,7 @@ from shared_configs.enums import RerankerProvider
 
 
 logger = setup_logger()
+setup_uvicorn_logger()
 
 
 def validation_exception_handler(request: Request, exc: Exception) -> JSONResponse:

backend/danswer/utils/logger.py

@@ -6,6 +6,9 @@ from typing import Any
 from shared_configs.configs import LOG_LEVEL
 
 
+logging.addLevelName(logging.INFO + 5, "NOTICE")
+
+
 class IndexAttemptSingleton:
     """Used to tell if this process is an indexing job, and if so what is the
     unique identifier for this indexing attempt. For things like the API server,
@@ -27,12 +30,13 @@ def get_log_level_from_str(log_level_str: str = LOG_LEVEL) -> int:
         "CRITICAL": logging.CRITICAL,
         "ERROR": logging.ERROR,
         "WARNING": logging.WARNING,
+        "NOTICE": logging.getLevelName("NOTICE"),
         "INFO": logging.INFO,
         "DEBUG": logging.DEBUG,
         "NOTSET": logging.NOTSET,
     }
 
-    return log_level_dict.get(log_level_str.upper(), logging.INFO)
+    return log_level_dict.get(log_level_str.upper(), logging.getLevelName("NOTICE"))
 
 
 class _IndexAttemptLoggingAdapter(logging.LoggerAdapter):
@@ -51,12 +55,49 @@ class _IndexAttemptLoggingAdapter(logging.LoggerAdapter):
 
         return f"[Attempt ID: {attempt_id}] {msg}", kwargs
 
+    def notice(self, msg: str, *args: Any, **kwargs: Any) -> None:
+        self.log(logging.getLevelName("NOTICE"), msg, *args, **kwargs)
+
+
+class ColoredFormatter(logging.Formatter):
+    """Custom formatter to add colors to log levels."""
+
+    COLORS = {
+        "CRITICAL": "\033[91m",  # Red
+        "ERROR": "\033[91m",  # Red
+        "WARNING": "\033[93m",  # Yellow
+        "NOTICE": "\033[94m",  # Blue
+        "INFO": "\033[92m",  # Green
+        "DEBUG": "\033[96m",  # Light Green
+        "NOTSET": "\033[91m",  # Reset
+    }
+
+    def format(self, record: logging.LogRecord) -> str:
+        levelname = record.levelname
+        if levelname in self.COLORS:
+            prefix = self.COLORS[levelname]
+            suffix = "\033[0m"
+            formatted_message = super().format(record)
+            # Ensure the levelname with colon is 9 characters long
+            # accounts for the extra characters for coloring
+            level_display = f"{prefix}{levelname}{suffix}:"
+            return f"{level_display.ljust(18)} {formatted_message}"
+        return super().format(record)
+
+
+def get_standard_formatter() -> ColoredFormatter:
+    """Returns a standard colored logging formatter."""
+    return ColoredFormatter(
+        "%(asctime)s %(filename)30s %(lineno)4s: %(message)s",
+        datefmt="%m/%d/%Y %I:%M:%S %p",
+    )
+
 
 def setup_logger(
     name: str = __name__,
     log_level: int = get_log_level_from_str(),
     logfile_name: str | None = None,
-) -> logging.LoggerAdapter:
+) -> _IndexAttemptLoggingAdapter:
     logger = logging.getLogger(name)
 
     # If the logger already has handlers, assume it was already configured and return it.
@@ -65,10 +106,7 @@ def setup_logger(
 
     logger.setLevel(log_level)
 
-    formatter = logging.Formatter(
-        "%(asctime)s %(filename)20s%(lineno)4s : %(message)s",
-        datefmt="%m/%d/%Y %I:%M:%S %p",
-    )
+    formatter = get_standard_formatter()
 
     handler = logging.StreamHandler()
     handler.setLevel(log_level)
@@ -84,4 +122,17 @@ def setup_logger(
         file_handler = logging.FileHandler(file_name_template.format(name=logfile_name))
         logger.addHandler(file_handler)
 
+    logger.notice = lambda msg, *args, **kwargs: logger.log(logging.getLevelName("NOTICE"), msg, *args, **kwargs)  # type: ignore
+
     return _IndexAttemptLoggingAdapter(logger)
+
+
+def setup_uvicorn_logger() -> None:
+    logger = logging.getLogger("uvicorn.access")
+    handler = logging.StreamHandler()
+    handler.setLevel(LOG_LEVEL)
+
+    formatter = get_standard_formatter()
+    handler.setFormatter(formatter)
+
+    logger.handlers = [handler]