Source code for galaxy.job_metrics

"""This module defines the job metrics collection framework for Galaxy jobs.

The framework consists of two parts - the :class:`JobMetrics` class and
individual :class:`JobInstrumenter` plugins.

A :class:`JobMetrics` object reads any number of plugins from a configuration
source such as an XML file, a YAML file, or a dictionary.

Each :class:`JobInstrumenter` plugin object describes how to inject a bits
of shell code into a job scripts (before and after tool commands run) and then
collect the output of these from a job directory.
"""

import collections
import logging
import os
from abc import (
    ABCMeta,
    abstractmethod,
)
from typing import (
    Any,
    cast,
    NamedTuple,
    TYPE_CHECKING,
    Union,
)

from galaxy import util
from galaxy.util import plugin_config
from . import formatting
from .safety import (
    DEFAULT_SAFETY,
    Safety,
)

if TYPE_CHECKING:
    from galaxy.job_metrics.instrumenters import (
        InstrumentPlugin,
        ProvidesJobMetricsContext,
    )
    from galaxy.util import Element

log = logging.getLogger(__name__)


DEFAULT_FORMATTER = formatting.JobMetricFormatter()
DEFAULT_CONFIG = [{"type": "core"}]


class DictifiableMetric(NamedTuple):
    """The full context of a metric that is ready to be exposed to an external client."""

    title: str
    value: str
    raw_value: str
    name: str
    plugin: str
    safety: Safety = Safety.POTENTIALLY_SENSITVE

    def dict(self) -> dict[str, str]:
        return dict(
            title=self.title,
            value=self.value,
            plugin=self.plugin,
            name=self.name,
            raw_value=self.raw_value,
        )


class RawMetric(NamedTuple):
    metric_name: str
    metric_value: Any
    metric_plugin: str


class JobMetrics:
    """Load and store a collection of :class:`JobInstrumenter` objects."""

    def __init__(self, conf_file=None, conf_dict=None, **kwargs):
        """Load :class:`JobInstrumenter` objects from specified configuration file."""
        self.plugin_classes = cast(dict[str, "InstrumentPlugin"], self.__plugins_dict())
        if conf_file and os.path.exists(conf_file):
            self.default_job_instrumenter = JobInstrumenter.from_file(self.plugin_classes, conf_file, **kwargs)
        elif conf_dict or conf_dict is None:
            if conf_dict is None:
                conf_dict = DEFAULT_CONFIG
            self.default_job_instrumenter = JobInstrumenter.from_dict(self.plugin_classes, conf_dict, **kwargs)
        else:
            # allows for setting non-None falsey values to get no metrics config whatsoever
            self.default_job_instrumenter = NULL_JOB_INSTRUMENTER
        self.job_instrumenters = collections.defaultdict(lambda: self.default_job_instrumenter)

    def format(self, plugin: str, key: str, value: Any) -> formatting.FormattedMetric | None:
        """Find :class:`formatting.JobMetricFormatter` corresponding to instrumented plugin value.

        None means the plugin recorded this metric but does not want it displayed.

        Asks the configured plugin first, the way the safety lookup below does, so that display
        options an admin set travel from the metrics configuration to the rendered metric. The
        default instrumenter is the one consulted: rendering happens without knowing which
        destination the job ran on.
        """
        formatter = None
        configured_plugin = self.default_job_instrumenter.get_configured_plugin(plugin)
        if configured_plugin is not None:
            formatter = configured_plugin.formatter
        if formatter is None and plugin in self.plugin_classes:
            formatter = self.plugin_classes[plugin].formatter
        if formatter is None:
            formatter = DEFAULT_FORMATTER
        return formatter.format(key, value)

    def dictifiable_metrics(self, raw_metrics: list[RawMetric], allowed_safety: Safety) -> list[DictifiableMetric]:
        def raw_to_dictifiable(raw_metric: RawMetric) -> DictifiableMetric | None:
            metric_name, metric_value, metric_plugin = raw_metric
            formatted = self.format(metric_plugin, metric_name, metric_value)
            if formatted is None:
                return None
            title, value = formatted
            configured_plugin = self.default_job_instrumenter.get_configured_plugin(metric_plugin)
            if configured_plugin is not None:
                safety = configured_plugin.safety(metric_name)
            elif metric_plugin in self.plugin_classes:
                plugin_class = self.plugin_classes[metric_plugin]
                safety = plugin_class.default_safety
            else:
                safety = DEFAULT_SAFETY
            return DictifiableMetric(
                title,
                value,
                str(metric_value),
                metric_name,
                metric_plugin,
                safety,
            )

        metrics = (m for m in map(raw_to_dictifiable, raw_metrics) if m is not None)
        return [m for m in metrics if m.safety.value >= allowed_safety.value]

    def set_destination_conf_file(self, destination_id: str, conf_file: str) -> None:
        instrumenter = JobInstrumenter.from_file(self.plugin_classes, conf_file)
        self.set_destination_instrumenter(destination_id, instrumenter)

    def set_destination_conf_element(self, destination_id: str, element: "Element") -> None:
        plugin_source = plugin_config.PluginConfigSource("xml", element)
        instrumenter = JobInstrumenter(self.plugin_classes, plugin_source)
        self.set_destination_instrumenter(destination_id, instrumenter)

    def set_destination_conf_dicts(self, destination_id: str, conf_dicts: list[dict[str, Any]]) -> None:
        plugin_source = plugin_config.PluginConfigSource("dict", conf_dicts)
        instrumenter = JobInstrumenter(self.plugin_classes, plugin_source)
        self.set_destination_instrumenter(destination_id, instrumenter)

    def set_destination_instrumenter(
        self, destination_id: str, job_instrumenter: Union["JobInstrumenterI", None] = None
    ) -> None:
        if job_instrumenter is None:
            job_instrumenter = NULL_JOB_INSTRUMENTER
        self.job_instrumenters[destination_id] = job_instrumenter

    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

        return plugin_config.plugins_dict(galaxy.job_metrics.instrumenters, "plugin_type")


class JobInstrumenterI(metaclass=ABCMeta):
    @abstractmethod
    def pre_execute_commands(self, job_directory: str) -> str | None:
        return None

    @abstractmethod
    def post_execute_commands(self, job_directory: str) -> str | None:
        return None

    @abstractmethod
    def collect_properties(self, job: "ProvidesJobMetricsContext", job_directory: str) -> dict[str, Any]:
        return {}

    @abstractmethod
    def get_configured_plugin(self, plugin_type: str):
        return None


class NullJobInstrumenter(JobInstrumenterI):
    def pre_execute_commands(self, job_directory):
        return None

    def post_execute_commands(self, job_directory):
        return None

    def collect_properties(self, job, job_directory):
        return {}

    def get_configured_plugin(self, plugin_type: str):
        return None


NULL_JOB_INSTRUMENTER = NullJobInstrumenter()



[docs]
class JobInstrumenter(JobInstrumenterI):

[docs]
    def __init__(self, plugin_classes, plugins_source, **kwargs):
        self.extra_kwargs = kwargs
        self.plugin_classes = plugin_classes
        self.plugins = self.__plugins_from_source(plugins_source)



[docs]
    def get_configured_plugin(self, plugin_type: str):
        for plugin in self.plugins:
            if plugin.plugin_type == plugin_type:
                return plugin
        return None



[docs]
    def pre_execute_commands(self, job_directory):
        commands = []
        for plugin in self.plugins:
            try:
                plugin_commands = plugin.pre_execute_instrument(job_directory)
                if plugin_commands:
                    commands.extend(util.listify(plugin_commands))
            except Exception:
                log.exception("Failed to generate pre-execute commands for plugin %s", plugin)
        return "\n".join(c for c in commands if c)



[docs]
    def post_execute_commands(self, job_directory):
        commands = []
        for plugin in self.plugins:
            try:
                plugin_commands = plugin.post_execute_instrument(job_directory)
                if plugin_commands:
                    commands.extend(util.listify(plugin_commands))
            except Exception:
                log.exception("Failed to generate post-execute commands for plugin %s", plugin)
        return "\n".join(c for c in commands if c)



[docs]
    def collect_properties(self, job, job_directory):
        per_plugin_properties = {}
        for plugin in self.plugins:
            try:
                properties = plugin.collect(job, job_directory)
                if properties:
                    per_plugin_properties[plugin.plugin_type] = properties
            except FileNotFoundError as e:
                log.warning("Failed to collect job properties for plugin %s: %s", plugin, e)
            except Exception:
                log.exception("Failed to collect job properties for plugin %s", plugin)
        return per_plugin_properties


    def __plugins_from_source(self, plugins_source):
        return plugin_config.load_plugins(self.plugin_classes, plugins_source, self.extra_kwargs)


[docs]
    @staticmethod
    def from_file(plugin_classes, conf_file, **kwargs) -> "JobInstrumenterI":
        if not conf_file or not os.path.exists(conf_file):
            return NULL_JOB_INSTRUMENTER
        plugins_source = plugin_config.plugin_source_from_path(conf_file)
        return JobInstrumenter(plugin_classes, plugins_source, **kwargs)



[docs]
    @staticmethod
    def from_dict(plugin_classes, conf_dict, **kwargs) -> "JobInstrumenterI":
        plugin_source = plugin_config.plugin_source_from_dict(conf_dict)
        return JobInstrumenter(plugin_classes, plugin_source, **kwargs)




__all__ = (
    "JobInstrumenter",
    "Safety",
)