pilot/util/workernode.py

0001 #!/usr/bin/env python
0002 # Licensed under the Apache License, Version 2.0 (the "License");
0003 # you may not use this file except in compliance with the License.
0004 # You may obtain a copy of the License at
0005 # http://www.apache.org/licenses/LICENSE-2.0
0006 #
0007 # Authors:
0008 # - Paul Nilsson, paul.nilsson@cern.ch, 2017-2021
0009
0010 import os
0011 import re
0012 import logging
0013
0014 from pilot.info import infosys
0015 from pilot.util.disk import disk_usage
0016
0017 logger = logging.getLogger(__name__)
0018
0019
0020 def get_local_disk_space(path):
0021     """
0022     Return remaning disk space for the disk in the given path.
0023     Unit is MB.
0024
0025     :param path: path to disk (string). Can be None, if call to collect_workernode_info() doesn't specify it.
0026     :return: disk space (float).
0027     """
0028
0029     if not path:
0030         return None
0031
0032     disk = 0.0
0033     # -mP = blocks of 1024*1024 (MB) and POSIX format
0034     diskpipe = os.popen("df -mP %s" % path)
0035     disks = diskpipe.read()
0036     if not diskpipe.close():
0037         try:
0038             disk = float(disks.splitlines()[1].split()[3])
0039         except ValueError as error:
0040             logger.warning('exception caught while trying to convert disk info: %s', error)
0041
0042     return disk
0043
0044
0045 def get_meminfo():
0046     """
0047     Return the total memory (in MB).
0048
0049     :return: memory (float).
0050     """
0051
0052     mem = 0.0
0053     with open("/proc/meminfo", "r") as fd:
0054         mems = fd.readline()
0055         while mems:
0056             if mems.upper().find("MEMTOTAL") != -1:
0057                 try:
0058                     mem = float(mems.split()[1]) / 1024  # value listed by command as kB, convert to MB
0059                 except ValueError as error:
0060                     logger.warning('exception caught while trying to convert meminfo: %s', error)
0061                 break
0062             mems = fd.readline()
0063
0064     return mem
0065
0066
0067 def get_cpuinfo():
0068     """
0069     Return the CPU frequency (in MHz).
0070
0071     :return: cpu (float).
0072     """
0073
0074     cpu = 0.0
0075     with open("/proc/cpuinfo", "r") as fd:
0076         lines = fd.readlines()
0077         for line in lines:
0078             if line.find("cpu MHz") != -1:  # Python 2/3
0079                 try:
0080                     cpu = float(line.split(":")[1])
0081                 except ValueError as error:
0082                     logger.warning('exception caught while trying to convert cpuinfo: %s', error)
0083                 break  # command info is the same for all cores, so break here
0084
0085     return cpu
0086
0087
0088 def collect_workernode_info(path=None):
0089     """
0090     Collect node information (cpu, memory and disk space).
0091     The disk space (in MB) is return for the disk in the given path.
0092
0093     :param path: path to disk (string).
0094     :return: memory (float), cpu (float), disk space (float).
0095     """
0096
0097     mem = get_meminfo()
0098     cpu = get_cpuinfo()
0099     disk = get_local_disk_space(path)
0100
0101     return mem, cpu, disk
0102
0103
0104 def get_disk_space(queuedata):
0105     """
0106     Get the disk space from the queuedata that should be available for running the job;
0107     either what is actually locally available or the allowed size determined by the site (value from queuedata). This
0108     value is only to be used internally by the job dispatcher.
0109
0110     :param queuedata: infosys object.
0111     :return: disk space that should be available for running the job (int).
0112     """
0113
0114     # --- non Job related queue data
0115     # jobinfo provider is required to consider overwriteAGIS data coming from Job
0116     _maxinputsize = infosys.queuedata.maxwdir
0117     logger.debug("resolved value from global infosys.queuedata instance: infosys.queuedata.maxwdir=%s B", _maxinputsize)
0118     _maxinputsize = queuedata.maxwdir
0119     logger.debug("resolved value: queuedata.maxwdir=%s B", _maxinputsize)
0120
0121     try:
0122         du = disk_usage(os.path.abspath("."))
0123         _diskspace = int(du[2] / (1024 * 1024))  # need to convert from B to MB
0124     except ValueError as error:
0125         logger.warning("failed to extract disk space: %s (will use schedconfig default)", error)
0126         _diskspace = _maxinputsize
0127     else:
0128         logger.info("available WN disk space: %d MB", _diskspace)
0129
0130     _diskspace = min(_diskspace, _maxinputsize)
0131     logger.info("sending disk space %d MB to dispatcher", _diskspace)
0132
0133     return _diskspace
0134
0135
0136 def get_node_name():
0137     """
0138     Return the local node name.
0139
0140     :return: node name (string)
0141     """
0142     if 'PANDA_HOSTNAME' in os.environ:
0143         host = os.environ.get('PANDA_HOSTNAME')
0144     elif hasattr(os, 'uname'):
0145         host = os.uname()[1]
0146     else:
0147         import socket
0148         host = socket.gethostname()
0149
0150     return get_condor_node_name(host)
0151
0152
0153 def get_condor_node_name(nodename):
0154     """
0155     On a condor system, add the SlotID to the nodename
0156
0157     :param nodename:
0158     :return:
0159     """
0160
0161     if "_CONDOR_SLOT" in os.environ:
0162         nodename = "%s@%s" % (os.environ.get("_CONDOR_SLOT"), nodename)
0163
0164     return nodename
0165
0166
0167 def get_cpu_model():
0168     """
0169     Get cpu model and cache size from /proc/cpuinfo.
0170
0171     Example.
0172       model name      : Intel(R) Xeon(TM) CPU 2.40GHz
0173       cache size      : 512 KB
0174
0175     gives the return string "Intel(R) Xeon(TM) CPU 2.40GHz 512 KB".
0176
0177     :return: cpu model (string).
0178     """
0179
0180     cpumodel = ""
0181     cpucache = ""
0182     modelstring = ""
0183
0184     re_model = re.compile(r'^model name\s+:\s+(\w.+)')  # Python 3 (added r)
0185     re_cache = re.compile(r'^cache size\s+:\s+(\d+ KB)')  # Python 3 (added r)
0186
0187     with open("/proc/cpuinfo", "r") as f:
0188
0189         # loop over all lines in cpuinfo
0190         for line in f.readlines():
0191             # try to grab cpumodel from current line
0192             model = re_model.search(line)
0193             if model:
0194                 # found cpu model
0195                 cpumodel = model.group(1)
0196
0197             # try to grab cache size from current line
0198             cache = re_cache.search(line)
0199             if cache:
0200                 # found cache size
0201                 cpucache = cache.group(1)
0202
0203             # stop after 1st pair found - can be multiple cpus
0204             if cpumodel and cpucache:
0205                 # create return string
0206                 modelstring = cpumodel + " " + cpucache
0207                 break
0208
0209     # default return string if no info was found
0210     if not modelstring:
0211         modelstring = "UNKNOWN"
0212
0213     return modelstring
0214
0215
0216 def check_hz():
0217     """
0218     Try to read the SC_CLK_TCK and write it to the log.
0219
0220     :return:
0221     """
0222
0223     try:
0224         _ = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
0225     except Exception:
0226         import traceback
0227         logger.fatal('failed to read SC_CLK_TCK - will not be able to perform CPU consumption calculation')
0228         logger.warning(traceback.format_exc())