Let instrument plugins read the job, without importing it

Metrics plugins have only ever seen a job id and a directory, so anything wanting a
property of the job itself has had to reach for the app -- and galaxy-job-metrics is one
of the five packages Pulsar installs on compute nodes, so it cannot import galaxy.model
or SQLAlchemy to do that.

Declare the slice a plugin is allowed to see as a Protocol in the package itself, the way
galaxy.objectstore does with UserObjectStoreResolver and galaxy.files with
UserDefinedFileSources, and have the caller pass the Job it already holds. Job satisfies
it structurally; no import crosses the boundary and no dependency is added.

The framework now calls a new collect() hook whose default delegates to job_properties,
so in-tree and out-of-tree plugins that only need the directory are untouched. Pulsar
only ever calls pre_execute_commands/post_execute_commands, so widening the internal
collect_properties signature does not reach it.

lib/galaxy/job_metrics/__init__.py

@@ -35,7 +35,10 @@ from .safety import (
 )
 
 if TYPE_CHECKING:
-    from galaxy.job_metrics.instrumenters import InstrumentPlugin
+    from galaxy.job_metrics.instrumenters import (
+        InstrumentPlugin,
+        ProvidesJobMetricsContext,
+    )
     from galaxy.util import Element
 
 log = logging.getLogger(__name__)
@@ -143,8 +146,8 @@ class JobMetrics:
             job_instrumenter = NULL_JOB_INSTRUMENTER
         self.job_instrumenters[destination_id] = job_instrumenter
 
-    def collect_properties(self, destination_id, job_id, job_directory):
-        return self.job_instrumenters[destination_id].collect_properties(job_id, job_directory)
+    def collect_properties(self, destination_id, job: "ProvidesJobMetricsContext", job_directory):
+        return self.job_instrumenters[destination_id].collect_properties(job, job_directory)
 
     def __plugins_dict(self):
         import galaxy.job_metrics.instrumenters
@@ -162,7 +165,7 @@ class JobInstrumenterI(metaclass=ABCMeta):
         return None
 
     @abstractmethod
-    def collect_properties(self, job_id, job_directory: str) -> dict[str, Any]:
+    def collect_properties(self, job: "ProvidesJobMetricsContext", job_directory: str) -> dict[str, Any]:
         return {}
 
     @abstractmethod
@@ -177,7 +180,7 @@ class NullJobInstrumenter(JobInstrumenterI):
     def post_execute_commands(self, job_directory):
         return None
 
-    def collect_properties(self, job_id, job_directory):
+    def collect_properties(self, job, job_directory):
         return {}
 
     def get_configured_plugin(self, plugin_type: str):
@@ -221,11 +224,11 @@ class JobInstrumenter(JobInstrumenterI):
                 log.exception("Failed to generate post-execute commands for plugin %s", plugin)
         return "\n".join(c for c in commands if c)
 
-    def collect_properties(self, job_id, job_directory):
+    def collect_properties(self, job, job_directory):
         per_plugin_properties = {}
         for plugin in self.plugins:
             try:
-                properties = plugin.job_properties(job_id, job_directory)
+                properties = plugin.collect(job, job_directory)
                 if properties:
                     per_plugin_properties[plugin.plugin_type] = properties
             except FileNotFoundError as e:

lib/galaxy/job_metrics/instrumenters/__init__.py

@@ -10,6 +10,7 @@ from abc import (
 )
 from typing import (
     Any,
+    Protocol,
 )
 
 from .. import formatting
@@ -22,6 +23,22 @@ INSTRUMENT_FILE_PREFIX = "__instrument"
 InstrumentableT = str | list[str] | None
 
 
+class ProvidesJobMetricsContext(Protocol):
+    """The slice of a Galaxy job that plugins may read when metrics are collected.
+
+    Declared structurally instead of importing ``galaxy.model``: this package ships to
+    Pulsar compute nodes (see ``packages/packages_for_pulsar_by_dep_dag.txt``) and so
+    depends on ``galaxy-util`` alone. The app passes something Job-shaped in; nothing
+    here needs to know it is a Job.
+    """
+
+    @property
+    def id(self) -> int: ...
+
+    @property
+    def resubmission_count(self) -> int: ...
+
+
 class InstrumentPlugin(metaclass=ABCMeta):
     """Describes how to instrument job scripts and retrieve collected metrics."""
 
@@ -55,6 +72,15 @@ class InstrumentPlugin(metaclass=ABCMeta):
         post_execute_instrument are available.
         """
 
+    def collect(self, job: ProvidesJobMetricsContext, job_directory: str) -> dict[str, Any]:
+        """Collect properties for this plugin, given the job they belong to.
+
+        The framework calls this rather than job_properties directly. Override it to read
+        the job itself; the default keeps plugins that only need the directory working
+        unchanged.
+        """
+        return self.job_properties(job.id, job_directory)
+
     def safety(self, metric_name: str) -> Safety:
         """Return safety level of metric."""
         # None of the plugins override this to dispatch on metric_name but on next

lib/galaxy/jobs/__init__.py

@@ -2464,9 +2464,7 @@ class MinimalJobWrapper(HasResourceParameters):
             except Exception:
                 log.exception("Could not recover job metrics")
                 return
-        per_plugin_properties = self.app.job_metrics.collect_properties(
-            job.destination_id, self.job_id, job_metrics_directory
-        )
+        per_plugin_properties = self.app.job_metrics.collect_properties(job.destination_id, job, job_metrics_directory)
         if per_plugin_properties:
             log.info(
                 f"Collecting metrics for {type(has_metrics).__name__} {getattr(has_metrics, 'id', None)} in {job_metrics_directory}"

test/unit/job_metrics/test_instrument_plugin_context.py

@@ -0,0 +1,63 @@
+"""The job context the framework hands to instrument plugins at collection time."""
+
+from typing import Any
+
+from galaxy.job_metrics import JobInstrumenter
+from galaxy.job_metrics.instrumenters import InstrumentPlugin
+from galaxy.util.plugin_config import PluginConfigSource
+
+
+class FakeJob:
+    """Something Job-shaped, satisfying ProvidesJobMetricsContext structurally."""
+
+    def __init__(self, id: int, resubmission_count: int = 0) -> None:
+        self.id = id
+        self.resubmission_count = resubmission_count
+
+
+class DirectoryOnlyPlugin(InstrumentPlugin):
+    """A plugin of the pre-existing kind: it knows a job id and a directory, nothing more."""
+
+    plugin_type = "directory_only"
+
+    def job_properties(self, job_id, job_directory: str) -> dict[str, Any]:
+        return {"job_id": job_id, "job_directory": job_directory}
+
+
+class JobReadingPlugin(InstrumentPlugin):
+    """A plugin that overrides the new hook to read the job itself."""
+
+    plugin_type = "job_reading"
+
+    def job_properties(self, job_id, job_directory: str) -> dict[str, Any]:
+        return {}
+
+    def collect(self, job, job_directory: str) -> dict[str, Any]:
+        return {"resubmission_count": job.resubmission_count}
+
+
+def _instrumenter_for(*plugins) -> JobInstrumenter:
+    instrumenter = JobInstrumenter({}, PluginConfigSource("dict", []))
+    instrumenter.plugins = list(plugins)
+    return instrumenter
+
+
+def test_collect_defaults_to_job_properties_with_the_job_id():
+    properties = DirectoryOnlyPlugin().collect(FakeJob(42), "/job/directory")
+
+    assert properties == {"job_id": 42, "job_directory": "/job/directory"}
+
+
+def test_collect_can_be_overridden_to_read_the_job():
+    properties = JobReadingPlugin().collect(FakeJob(42, resubmission_count=3), "/job/directory")
+
+    assert properties == {"resubmission_count": 3}
+
+
+def test_instrumenter_routes_collection_through_collect():
+    instrumenter = _instrumenter_for(DirectoryOnlyPlugin(), JobReadingPlugin())
+
+    per_plugin = instrumenter.collect_properties(FakeJob(42, resubmission_count=1), "/job/directory")
+
+    assert per_plugin["directory_only"] == {"job_id": 42, "job_directory": "/job/directory"}
+    assert per_plugin["job_reading"] == {"resubmission_count": 1}