Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

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))