Hoist Gandalf logs and clean up common logs

src/retriever/data_tiers/tier_0/base_query.py

@@ -35,7 +35,6 @@ async def execute(self) -> LookupArtifacts:
         """Execute a lookup against Tier 0 and return what is found."""
         try:
             start_time = time.time()
-            self.job_log.info("Starting lookup against Tier 0...")
 
             timeout = None if self.ctx.timeout < 0 else self.ctx.timeout
             self.job_log.debug(
@@ -50,6 +49,12 @@ async def execute(self) -> LookupArtifacts:
                 backend_results["auxiliary_graphs"],
             )
 
+            if logs := backend_results.get("logs"):
+                self.job_log.debug("<Begin backend-produced logs>")
+                for entry in logs:
+                    self.job_log.replay_entry(entry)
+                self.job_log.debug("<End of backend logs>")
+
             parameters = (self.ctx.body or {}).get("parameters") or ParametersDict()
 
             if not parameters.get("dehydrated"):

src/retriever/data_tiers/tier_0/gandalf/query.py

@@ -32,7 +32,10 @@ async def get_results(self, qgraph: QueryGraphDict) -> BackendResult:
 
         return BackendResult(
             results=result["message"].get("results") or [],
-            knowledge_graph=result["message"].get("knowledge_graph")
-            or KnowledgeGraphDict(nodes={}, edges={}),
-            auxiliary_graphs=result["message"].get("auxiliary_graphs") or {},
+            knowledge_graph=(
+                result["message"].get("knowledge_graph")
+                or KnowledgeGraphDict(nodes={}, edges={})
+            ),
+            auxiliary_graphs=(result["message"].get("auxiliary_graphs") or {}),
+            logs=(result.get("logs") or []),
         )

src/retriever/lookup/lookup.py

@@ -174,7 +174,7 @@ async def lookup(query: QueryInfo) -> tuple[HTTPStatus, ResponseDict]:
     try:
         start_time = time.time()
         job_log.info(
-            f"Begin processing job {job_id} for client {get_submitter(query)}."
+            f"Begin processing tier-{query.tier} job {job_id} for client `{get_submitter(query)}`."
         )
         if (query.body.get("parameters") or {}).get("dehydrated"):
             job_log.info(
@@ -216,8 +216,6 @@ async def lookup(query: QueryInfo) -> tuple[HTTPStatus, ResponseDict]:
 
         job_log.log_deque.extend(logs)
 
-        job_log.info(f"Collected {len(results)} results from query task.")
-
         duration_ms = math.ceil((time.time() - start_time) * 1000)
         finish_msg = _summarize_execution(
             query, status, len(results), duration_ms, job_log

src/retriever/lookup/qgx.py

@@ -161,7 +161,6 @@ async def execute(self) -> LookupArtifacts:
         timeout_task = asyncio.create_task(self.start_timeout_clock())
         try:
             self.start_time = time.time()
-            self.job_log.info(f"Starting lookup against Tier {self.ctx.tier}...")
             supported, operation_plan = await OP_TABLE_MANAGER.create_operation_plan(
                 self.qgraph, (self.ctx.tier or 0)
             )

src/retriever/lookup/utils.py

@@ -39,12 +39,8 @@ def expand_qgraph(qg: QueryGraphDict, job_log: TRAPILogger) -> QueryGraphDict:
         # Not necessary because backends check against descendants
         # new_categories = biolink.expand(categories) - categories
 
-        if "biolink:NamedThing" in categories:
-            job_log.info(
-                f"QNode {qnode_id}: Expanded to all categories (original had NamedThing)."
-            )
-        elif len(new_categories):
-            job_log.info(
+        if len(new_categories):
+            job_log.debug(
                 f"QNode {qnode_id}: Added descendant categories {new_categories}."
             )
 
@@ -61,12 +57,8 @@ def expand_qgraph(qg: QueryGraphDict, job_log: TRAPILogger) -> QueryGraphDict:
         # Not necessary because backends check against descendants
         # new_predicates = biolink.expand(predicates) - predicates
 
-        if "biolink:related_to" in predicates:
-            job_log.info(
-                f"QEdge {qedge_id}: Expanded to all predicates (original had related_to)."
-            )
-        elif len(new_predicates):
-            job_log.info(
+        if len(new_predicates):
+            job_log.debug(
                 f"QEdge {qedge_id}: Added descendant predicates {new_predicates}."
             )
 

src/retriever/types/general.py

@@ -71,6 +71,7 @@ class BackendResult(TypedDict):
     results: list[ResultDict]
     knowledge_graph: KnowledgeGraphDict
     auxiliary_graphs: dict[AuxGraphID, AuxiliaryGraphDict]
+    logs: NotRequired[list[LogEntryDict]]
 
 
 class LookupArtifacts(NamedTuple):

src/retriever/utils/logs.py

@@ -170,6 +170,29 @@ def with_exception(
         """Log with a given exception as an ERROR-level log."""
         self._log("ERROR", message, exception, **kwargs)
 
+    def replay_entry(self, entry: LogEntryDict) -> None:
+        """Emit a TRAPI log through loguru, retaining it verbatim.
+
+        Adopts the entry's level and timestamp.
+        """
+        level = (entry.get("level") or TRAPILogLevel.DEBUG).upper()
+        raw_ts = entry.get("timestamp")
+        try:
+            timestamp = (
+                datetime.fromisoformat(raw_ts)
+                if raw_ts
+                else datetime.now().astimezone()
+            )
+        except ValueError:
+            timestamp = datetime.now().astimezone()
+
+        with logger.contextualize(job_id=self.job_id):
+            logger.patch(lambda record: record.update(time=timestamp)).log(
+                level, entry["message"]
+            )
+
+        self.log_deque.append(entry)
+
     def get_logs(self) -> list[LogEntryDict]:
         """Get a generator of stored TRAPI logs."""
         return list(self.log_deque)