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