This document is for an in-development version 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.cpuinfo

"""The module describes the ``cpuinfo`` job metrics plugin."""
import logging
import re
from typing import Any

from galaxy import util
from . import InstrumentPlugin
from ..formatting import (

log = logging.getLogger(__name__)

PROCESSOR_LINE = re.compile(r"processor\s*\:\s*(\d+)")

class CpuInfoFormatter(JobMetricFormatter):
    def format(self, key: str, value: Any) -> FormattedMetric:
        if key == "processor_count":
            return FormattedMetric("Processor Count", f"{int(value)}")
            return super().format(key, value)

[docs]class CpuInfoPlugin(InstrumentPlugin): """Gather information about processor configuration from /proc/cpuinfo. Linux only. """ plugin_type = "cpuinfo" formatter = CpuInfoFormatter()
[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/cpuinfo > '{self.__instrument_cpuinfo_path(job_directory)}'"
[docs] def job_properties(self, job_id, job_directory): properties = {} processor_count = 0 with open(self.__instrument_cpuinfo_path(job_directory)) as f: current_processor = None for line in f: line = line.strip().lower() if not line: # Skip empty lines continue processor_line_match = PROCESSOR_LINE.match(line) if processor_line_match: processor_count += 1 current_processor = processor_line_match.group(1) elif current_processor and self.verbose: # If verbose, dump information about each processor # into database... key, value = line.split(":", 1) key = f"processor_{current_processor}_{key.strip()}" value = value properties["processor_count"] = processor_count return properties
def __instrument_cpuinfo_path(self, job_directory): return self._instrument_file_path(job_directory, "cpuinfo")
__all__ = ("CpuInfoPlugin",)