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.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 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",)