Source code for galaxy.job_metrics.instrumenters.env

"""The module describes the ``env`` job metrics plugin."""
import logging
import re
from typing import (
    List,
    Optional,
)

from . import InstrumentPlugin
from ..formatting import JobMetricFormatter
from ..safety import Safety

log = logging.getLogger(__name__)


class EnvFormatter(JobMetricFormatter):
    pass


[docs]class EnvPlugin(InstrumentPlugin):
    """Instrumentation plugin capable of recording all or specific environment
    variables for a job at runtime.
    """

    plugin_type = "env"
    formatter = EnvFormatter()
    variables: Optional[List[str]]
    default_safety = Safety.UNSAFE

[docs]    def __init__(self, **kwargs):
        variables_str = kwargs.get("variables", None)
        if variables_str:
            self.variables = [v.strip() for v in variables_str.split(",")]
        else:
            self.variables = None

[docs]    def pre_execute_instrument(self, job_directory: str):
        """Use env to dump all environment variables to a file."""
        return f"env > '{self.__env_file(job_directory)}'"

[docs]    def post_execute_instrument(self, job_directory):
        return None

[docs]    def job_properties(self, job_id, job_directory):
        """Recover environment variables dumped out on compute server and filter
        out specific variables if needed.
        """
        variables = self.variables

        properties = {}
        env_string = "".join(open(self.__env_file(job_directory)).readlines())
        while env_string:
            # Check if the next lines contain a shell function.
            # We use '\n\}\n' as regex termination because shell
            # functions can be nested.
            # We use the non-greedy '.+?' because of re.DOTALL .
            m = re.match(r"([^=]+)=(\(\) \{.+?\n\})\n", env_string, re.DOTALL)
            if m is None:
                m = re.match("([^=]+)=(.*)\n", env_string)
            if m is None:
                # Some problem recording or reading back env output.
                message_template = "Problem parsing env metric output for job %s - properties will be incomplete"
                message = message_template % job_id
                log.debug(message)
                break
            (var, value) = m.groups()
            if not variables or var in variables:
                properties[var] = value
            env_string = env_string[m.end() :]

        return properties

    def __env_file(self, job_directory):
        return self._instrument_file_path(job_directory, "vars")


__all__ = ("EnvPlugin",)