Merge pull request #198 from pauldg/feat/explain-on-failure

Add a explain on failure option to log scheduling trace for job mapping failures

Author: nuwang

tests/test_explain.py

@@ -3,11 +3,14 @@
 from unittest.mock import MagicMock
 
 import yaml
+from galaxy.jobs import JobDestination
 from galaxy.jobs.mapper import JobMappingException
 
 from tpv.commands.dryrunner import TPVDryRunner
+from tpv.commands.test import mock_galaxy
 from tpv.core.entities import SchedulingTags
 from tpv.core.explain import ExplainCollector, ExplainPhase
+from tpv.rules import gateway
 
 
 class TestExplainCollectorUnit(unittest.TestCase):
@@ -496,3 +499,59 @@ def test_from_params_without_tool_id_sets_tool_none(self):
             tpv_confs=[self._fixture_path("mapping-basic.yml")],
         )
         self.assertIsNone(runner.tool)
+
+
+
+class TestGatewayExplainOnFailure(unittest.TestCase):
+    """Tests for the tpv_explain_on_failure gateway config option."""
+
+    @staticmethod
+    def _fixture_path(name):
+        return os.path.join(os.path.dirname(__file__), f"fixtures/{name}")
+
+    def _call_gateway(self, tool_id, referrer=None, explain_collector=None):
+        gateway.ACTIVE_DESTINATION_MAPPERS = {}
+        app = mock_galaxy.App(job_conf=self._fixture_path("job_conf.yml"))
+        tool = mock_galaxy.Tool(tool_id)
+        job = mock_galaxy.Job()
+        user = mock_galaxy.User("gargravarr", "fairycake@vortex.org")
+        return gateway.map_tool_to_destination(
+            app, job, tool, user,
+            referrer=referrer,
+            tpv_config_files=[self._fixture_path("mapping-basic.yml")],
+            explain_collector=explain_collector,
+        )
+
+    def test_no_log_when_param_not_set(self):
+        """No warning is logged on failure when tpv_explain_on_failure is not configured."""
+        with self.assertNoLogs("tpv.rules.gateway", level="WARNING"):
+            with self.assertRaises(JobMappingException):
+                self._call_gateway("unschedulable_tool")
+
+    def test_logs_trace_on_mapping_failure(self):
+        """A WARNING containing the scheduling trace is logged when tpv_explain_on_failure=true and mapping fails."""
+        referrer = JobDestination(id="tpv_dispatcher", params={"tpv_explain_on_failure": True})
+        with self.assertLogs("tpv.rules.gateway", level="WARNING") as cm:
+            with self.assertRaises(JobMappingException):
+                self._call_gateway("unschedulable_tool", referrer=referrer)
+        self.assertEqual(len(cm.records), 1)
+        trace = cm.records[0].getMessage()
+        self.assertIn("TPV SCHEDULING DECISION TRACE", trace)
+        self.assertIn("REJECTED", trace)
+        self.assertIn("tag mismatch", trace)
+        self.assertIn("No destinations", trace)
+
+    def test_no_log_on_successful_mapping(self):
+        """No warning is logged when tpv_explain_on_failure=true but mapping succeeds."""
+        referrer = JobDestination(id="tpv_dispatcher", params={"tpv_explain_on_failure": True})
+        with self.assertNoLogs("tpv.rules.gateway", level="WARNING"):
+            destination = self._call_gateway("bwa", referrer=referrer)
+        self.assertIsNotNone(destination)
+
+    def test_explicit_collector_not_logged_by_gateway(self):
+        """When an explicit explain_collector is passed (dry-run path), the gateway does not log on failure."""
+        collector = ExplainCollector()
+        with self.assertNoLogs("tpv.rules.gateway", level="WARNING"):
+            with self.assertRaises(JobMappingException):
+                self._call_gateway("unschedulable_tool", explain_collector=collector)
+        self.assertTrue(len(collector.steps) > 0)

tpv/core/helpers.py

@@ -167,4 +167,5 @@ def get_dataset_attributes(
             "size": get_dataset_size(i.dataset.dataset),
         }
         for i in datasets or {}
+        if i.dataset
     }

tpv/rules/gateway.py

@@ -6,6 +6,7 @@
 
 from galaxy.app import UniverseApplication
 from galaxy.jobs import JobDestination, JobWrapper
+from galaxy.jobs.mapper import JobMappingException
 from galaxy.model import Job
 from galaxy.model import User as GalaxyUser
 from galaxy.tools import Tool as GalaxyTool
@@ -117,18 +118,27 @@ def map_tool_to_destination(
         raise ValueError("One of tpv_configs or tpv_config_files must be specified in execution environment.")
     referrer_id = referrer.id if referrer else None
     destination_mapper = lock_and_load_mapper(app, referrer_id or "tpv_dispatcher", resolved_tpv_configs)
-    if explain_collector:
+    explain_on_failure = bool(referrer.params.get("tpv_explain_on_failure", False)) if referrer else False
+    log_on_failure = explain_collector is None and explain_on_failure
+    collector = explain_collector or (ExplainCollector() if log_on_failure else None)
+    if collector:
         configs_list = listify(resolved_tpv_configs)
         for config_source in configs_list:
             source_name = config_source if isinstance(config_source, str) else "<inline config>"
-            explain_collector.add_step(ExplainPhase.CONFIG_LOADING, f"Loaded config: {source_name}")
-    return destination_mapper.map_to_destination(
-        app,
-        tool,
-        user,
-        job,
-        job_wrapper,
-        resource_params,
-        workflow_invocation_uuid,
-        explain_collector=explain_collector,
-    )
+            collector.add_step(ExplainPhase.CONFIG_LOADING, f"Loaded config: {source_name}")
+    try:
+        return destination_mapper.map_to_destination(
+            app,
+            tool,
+            user,
+            job,
+            job_wrapper,
+            resource_params,
+            workflow_invocation_uuid,
+            explain_collector=collector,
+        )
+    except JobMappingException:
+        if log_on_failure:
+            assert collector is not None
+            log.warning("Job mapping failed. TPV scheduling trace:\n%s", collector.render())
+        raise