Source code for galaxy.job_metrics.instrumenters.meminfo

"""The module describes the ``meminfo`` job metrics plugin."""

import re
from typing import Any

from galaxy import util
from . import InstrumentPlugin
from .. import formatting
from ..safety import Safety

MEMINFO_LINE = re.compile(r"(\w+)\s*\:\s*(\d+) kB")

# Important (non-verbose) meminfo property titles.
MEMINFO_TITLES = {"memtotal": "Total System Memory", "swaptotal": "Total System Swap"}


class MemInfoFormatter(formatting.JobMetricFormatter):
    def format(self, key: str, value: Any) -> formatting.FormattedMetric:
        title = MEMINFO_TITLES.get(key, key)
        return formatting.FormattedMetric(title, util.nice_size(value * 1000))  # kB = *1000, KB = *1024 - wikipedia


[docs]class MemInfoPlugin(InstrumentPlugin):
    """Gather information about processor configuration from /proc/cpuinfo.
    Linux only.
    """

    plugin_type = "meminfo"
    formatter = MemInfoFormatter()
    default_safety = Safety.SAFE

[docs]    def __init__(self, **kwargs):
        self.verbose = util.asbool(kwargs.get("verbose", False))

[docs]    def pre_execute_instrument(self, job_directory):
        return f"cat /proc/meminfo > '{self.__instrument_meminfo_path(job_directory)}'"

[docs]    def job_properties(self, job_id, job_directory):
        properties = {}
        with open(self.__instrument_meminfo_path(job_directory)) as f:
            for line in f:
                line = line.strip()
                if not line:  # Skip empty lines
                    continue
                line_match = MEMINFO_LINE.match(line)
                if not line_match:
                    continue
                key = line_match.group(1).lower()
                # By default just grab important meminfo properties with titles
                # defined for formatter. Grab everything in verbose mode for
                # an arbitrary snapshot of memory at beginning of run.
                if key in MEMINFO_TITLES or self.verbose:
                    value = int(line_match.group(2))
                    properties[key] = value
        return properties

    def __instrument_meminfo_path(self, job_directory):
        return self._instrument_file_path(job_directory, "meminfo")


__all__ = ("MemInfoPlugin",)