Source code for galaxy.job_metrics.instrumenters

"""This module describes the abstract interface for :class:`InstrumentPlugin`.

These are responsible for collecting and formatting a coherent set of metrics.
"""

import os.path
from abc import (
    ABCMeta,
    abstractmethod,
)
from typing import (
    Any,
    Dict,
    List,
    Optional,
    Union,
)

from .. import formatting
from ..safety import (
    DEFAULT_SAFETY,
    Safety,
)

INSTRUMENT_FILE_PREFIX = "__instrument"
InstrumentableT = Optional[Union[str, List[str]]]


[docs]class InstrumentPlugin(metaclass=ABCMeta): """Describes how to instrument job scripts and retrieve collected metrics.""" formatter = formatting.JobMetricFormatter() default_safety = DEFAULT_SAFETY @property @abstractmethod def plugin_type(self): """Short string providing labelling this plugin"""
[docs] def pre_execute_instrument(self, job_directory: str) -> InstrumentableT: """Optionally return one or more commands to instrument job. These commands will be executed on the compute server prior to the job running. """ return None
[docs] def post_execute_instrument(self, job_directory: str) -> InstrumentableT: """Optionally return one or more commands to instrument job. These commands will be executed on the compute server after the tool defined command is ran. """ return None
[docs] @abstractmethod def job_properties(self, job_id, job_directory: str) -> Dict[str, Any]: """Collect properties for this plugin from specified job directory. This method will run on the Galaxy server and can assume files created in job_directory with pre_execute_instrument and post_execute_instrument are available. """
[docs] 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 # iteration it would make sense to allow admins to expose particular env vars # or to have cgroup keys we know are about runtime or memeory to be exposed # at a safer level. return self.default_safety
def _instrument_file_name(self, name: str) -> str: """Provide a common pattern for naming files used by instrumentation plugins - to ease their staging out of remote job directories. """ return f"{INSTRUMENT_FILE_PREFIX}_{self.plugin_type}_{name}" def _instrument_file_path(self, job_directory: str, name: str) -> str: return os.path.join(job_directory, self._instrument_file_name(name))