Improve logging, Add logging per VM

pyproject.toml

@@ -191,15 +191,16 @@ lint.ignore = [
   # Allow the use of assert statements
   "S101",
 ]
-# Tests can use magic values, assertions, and relative imports
-lint.per-file-ignores."tests/**/*" = [ "PLR2004", "S101", "TID252" ]
 #[tool.ruff.flake8-tidy-imports]
 #ban-relative-imports = "all"
 #unfixable = [
 #  # Don't touch unused imports
 #  "F401",
 #]
-lint.isort = [ "aleph.vm" ]
+#lint.isort = [ "aleph.vm" ]
+
+# Tests can use magic values, assertions, and relative imports
+lint.per-file-ignores."tests/**/*" = [ "PLR2004", "S101", "TID252" ]
 
 [tool.pytest.ini_options]
 pythonpath = [

src/aleph/vm/conf.py

@@ -136,6 +136,7 @@ class Settings(BaseSettings):
     # System logs make boot ~2x slower
     PRINT_SYSTEM_LOGS = False
     IGNORE_TRACEBACK_FROM_DIAGNOSTICS = True
+    LOG_LEVEL = "WARNING"
     DEBUG_ASYNCIO = False
 
     # Networking does not work inside Docker/Podman
@@ -396,8 +397,6 @@ def setup(self):
         STREAM_CHAINS[Chain.AVAX].rpc = str(self.RPC_AVAX)
         STREAM_CHAINS[Chain.BASE].rpc = str(self.RPC_BASE)
 
-        logger.info(STREAM_CHAINS)
-
         os.makedirs(self.MESSAGE_CACHE, exist_ok=True)
         os.makedirs(self.CODE_CACHE, exist_ok=True)
         os.makedirs(self.RUNTIME_CACHE, exist_ok=True)

src/aleph/vm/orchestrator/cli.py

@@ -23,6 +23,7 @@
 from aleph.vm.version import __version__, get_version_from_apt, get_version_from_git
 
 from . import metrics, supervisor
+from .custom_logs import setup_handlers
 from .pubsub import PubSub
 from .run import run_code_on_event, run_code_on_request, start_persistent_vm
 
@@ -65,7 +66,7 @@
         help="set loglevel to INFO",
         action="store_const",
         const=logging.INFO,
-        default=logging.WARNING,
+        default=settings.LOG_LEVEL,
     )
     parser.add_argument(
         "-vv",
@@ -282,7 +283,7 @@
 
 
 async def run_async_db_migrations():
-    async_engine = create_async_engine(make_db_url(), echo=True)
+    async_engine = create_async_engine(make_db_url(), echo=False)
     async with async_engine.begin() as conn:
         await conn.run_sync(run_db_migrations)
 
@@ -293,13 +294,20 @@
     log_format = (
         "%(relativeCreated)4f | %(levelname)s | %(message)s"
         if args.profile
-        else "%(asctime)s | %(levelname)s | %(message)s"
+        else "%(asctime)s | %(levelname)s %(name)s:%(lineno)s | %(message)s"
     )
+    # log_format = "[%(asctime)s] p%(process)s {%(pathname)s:%(lineno)d} %(levelname)s - %(message)s"
+
+    handlers = setup_handlers(args, log_format)
     logging.basicConfig(
         level=args.loglevel,
         format=log_format,
+        handlers=handlers,
     )
 
+    logging.getLogger("aiosqlite").setLevel(settings.LOG_LEVEL)
+    logging.getLogger("sqlalchemy.engine").setLevel(settings.LOG_LEVEL)
+
     settings.update(
         USE_JAILER=args.use_jailer,
         PRINT_SYSTEM_LOGS=args.system_logs,

src/aleph/vm/orchestrator/custom_logs.py

@@ -0,0 +1,54 @@
+import contextlib
+import logging
+from contextvars import ContextVar
+
+from aleph_message.models import ItemHash
+
+from aleph.vm.models import VmExecution
+
+ctx_current_execution: ContextVar[VmExecution | None] = ContextVar("current_execution")
+ctx_current_execution_hash: ContextVar[ItemHash | None] = ContextVar("current_execution_hash")
+
+
+@contextlib.contextmanager
+def set_vm_for_logging(vm_hash):
+    token = ctx_current_execution_hash.set(vm_hash)
+    try:
+        yield
+    finally:
+        ctx_current_execution_hash.reset(token)
+
+
+class InjectingFilter(logging.Filter):
+    """
+    A filter which injects context-specific information into logs
+    """
+
+    def filter(self, record):
+
+        vm_hash = ctx_current_execution_hash.get(None)
+        if not vm_hash:
+            vm_execution: VmExecution | None = ctx_current_execution.get(None)
+            if vm_execution:
+                vm_hash = vm_execution.vm_hash
+
+        if not vm_hash:
+            return False
+
+        record.vm_hash = vm_hash
+        return True
+
+
+def setup_handlers(args, log_format):
+    # Set up two custom handler, one that will add the VM information if present and the other print if not
+    execution_handler = logging.StreamHandler()
+    execution_handler.addFilter(InjectingFilter())
+    execution_handler.setFormatter(
+        logging.Formatter("%(asctime)s | %(levelname)s %(name)s:%(lineno)s | {%(vm_hash)s} %(message)s ")
+    )
+    non_execution_handler = logging.StreamHandler()
+    non_execution_handler.addFilter(lambda x: ctx_current_execution_hash.get(None) is None)
+    non_execution_handler.setFormatter(
+        logging.Formatter("%(asctime)s | %(levelname)s %(name)s:%(lineno)s | %(message)s ")
+    )
+    return [non_execution_handler, execution_handler]

src/aleph/vm/orchestrator/metrics.py

@@ -38,7 +38,7 @@
 
 def setup_engine():
     global AsyncSessionMaker
-    engine = create_async_engine(make_db_url(), echo=True)
+    engine = create_async_engine(make_db_url(), echo=False)
     AsyncSessionMaker = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
     return engine
 

src/aleph/vm/orchestrator/views/__init__.py

@@ -1,4 +1,5 @@
 import binascii
+import contextlib
 import logging
 from decimal import Decimal
 from hashlib import sha256
@@ -25,6 +26,7 @@
 from aleph.vm.hypervisors.firecracker.microvm import MicroVMFailedInitError
 from aleph.vm.orchestrator import payment, status
 from aleph.vm.orchestrator.chain import STREAM_CHAINS, ChainInfo
+from aleph.vm.orchestrator.custom_logs import set_vm_for_logging
 from aleph.vm.orchestrator.messages import try_get_message
 from aleph.vm.orchestrator.metrics import get_execution_records
 from aleph.vm.orchestrator.payment import (
@@ -75,7 +77,8 @@
         ) from e
 
     pool: VmPool = request.app["vm_pool"]
-    return await run_code_on_request(message_ref, path, pool, request)
+    with set_vm_for_logging(vm_hash=message_ref):
+        return await run_code_on_request(message_ref, path, pool, request)
 
 
 async def run_code_from_hostname(request: web.Request) -> web.Response:
@@ -112,7 +115,8 @@
                 return HTTPNotFound(reason="Invalid message reference")
 
     pool = request.app["vm_pool"]
-    return await run_code_on_request(message_ref, path, pool, request)
+    with set_vm_for_logging(vm_hash=message_ref):
+        return await run_code_on_request(message_ref, path, pool, request)
 
 
 def authenticate_request(request: web.Request) -> None: