Record container id and type in core job metrics

Closes #16588.
Not sure if we want to do this versus a separate but always loaded
plugin vs a regular relationship in the database.

Seems unlikely that we'd do anything beyond reporting, so maybe this is
fine for now ?

lib/galaxy/job_metrics/instrumenters/core.py

@@ -1,11 +1,14 @@
 """The module describes the ``core`` job metrics plugin."""
 
+import json
 import logging
+import os
 import time
 from typing import (
     Any,
     Dict,
     List,
+    Optional,
 )
 
 from . import InstrumentPlugin
@@ -23,10 +26,17 @@
 START_EPOCH_KEY = "start_epoch"
 END_EPOCH_KEY = "end_epoch"
 RUNTIME_SECONDS_KEY = "runtime_seconds"
+CONTAINER_FILE = "__container.json"
+CONTAINER_ID = "container_id"
+CONTAINER_TYPE = "container_type"
 
 
 class CorePluginFormatter(JobMetricFormatter):
     def format(self, key: str, value: Any) -> FormattedMetric:
+        if key == CONTAINER_ID:
+            return FormattedMetric("Container ID", value)
+        if key == CONTAINER_TYPE:
+            return FormattedMetric("Container Type", value)
         value = int(value)
         if key == GALAXY_SLOTS_KEY:
             return FormattedMetric("Cores Allocated", "%d" % value)
@@ -73,12 +83,20 @@ def job_properties(self, job_id, job_directory: str) -> Dict[str, Any]:
         properties[GALAXY_MEMORY_MB_KEY] = self.__read_integer(galaxy_memory_mb_file)
         start = self.__read_seconds_since_epoch(job_directory, "start")
         end = self.__read_seconds_since_epoch(job_directory, "end")
+        properties.update(self.__read_container_details(job_directory))
         if start is not None and end is not None:
             properties[START_EPOCH_KEY] = start
             properties[END_EPOCH_KEY] = end
             properties[RUNTIME_SECONDS_KEY] = end - start
         return properties
 
+    def __read_container_details(self, job_directory) -> Dict[str, str]:
+        try:
+            with open(os.path.join(job_directory, CONTAINER_FILE)) as fh:
+                return json.load(fh)
+        except FileNotFoundError:
+            return {}
+
     def __record_galaxy_slots_command(self, job_directory):
         galaxy_slots_file = self.__galaxy_slots_file(job_directory)
         return f"""echo "$GALAXY_SLOTS" > '{galaxy_slots_file}' """

lib/galaxy/jobs/command_factory.py

@@ -181,6 +181,10 @@ def __externalize_commands(
     source_command = ""
     if container:
         source_command = container.source_environment
+        with open(join(job_wrapper.working_directory, "__container.json"), "w") as container_file:
+            container_file.write(
+                json.dumps({"container_id": container.container_id, "container_type": container.container_type})
+            )
     script_contents = f"#!{shell}\n{integrity_injection}{set_e}{source_command}{tool_commands}"
     write_script(
         local_container_script,