From de184aacc7e101c74d5b23ba8ce788f0c5ede872 Mon Sep 17 00:00:00 2001 From: Paul Nilsson Date: Mon, 16 Dec 2024 10:05:01 +0100 Subject: [PATCH 1/7] Added missing f --- pilot/api/data.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pilot/api/data.py b/pilot/api/data.py index 61600cc2..7f734a66 100644 --- a/pilot/api/data.py +++ b/pilot/api/data.py @@ -1057,7 +1057,7 @@ def check_availablespace(self, files: list): else: if disk_space: available_space = convert_mb_to_b(disk_space) - self.logger.info("locally available space: {available_space} B") + self.logger.info(f"locally available space: {available_space} B") # are we within the limit? if totalsize > available_space: From aa6aabc039a1fcca36a3ce093e7aa433ac9a3fc9 Mon Sep 17 00:00:00 2001 From: Paul Nilsson Date: Mon, 16 Dec 2024 10:07:08 +0100 Subject: [PATCH 2/7] New version --- PILOTVERSION | 2 +- pilot/util/constants.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/PILOTVERSION b/PILOTVERSION index 1f3166a1..5ae185b5 100644 --- a/PILOTVERSION +++ b/PILOTVERSION @@ -1 +1 @@ -3.9.3.2 \ No newline at end of file +3.9.4.1 \ No newline at end of file diff --git a/pilot/util/constants.py b/pilot/util/constants.py index 30bab0af..4c792f0f 100644 --- a/pilot/util/constants.py +++ b/pilot/util/constants.py @@ -27,8 +27,8 @@ # Pilot version RELEASE = '3' # released number should be fixed at 3 for Pilot 3 VERSION = '9' # version number is '1' for first release, '0' until then, increased for bigger updates -REVISION = '3' # revision number should be reset to '0' for every new version release, increased for small updates -BUILD = '2' # build number should be reset to '1' for every new development cycle +REVISION = '4' # revision number should be reset to '0' for every new version release, increased for small updates +BUILD = '1' # build number should be reset to '1' for every new development cycle SUCCESS = 0 FAILURE = 1 From b95d9b9bd50c1be6401104f8c2e001376eac32e6 Mon Sep 17 00:00:00 2001 From: Paul Nilsson Date: Mon, 16 Dec 2024 11:31:56 +0100 Subject: [PATCH 3/7] Not allowing empty file names, possbly coming from \n in find command --- pilot/user/atlas/loopingjob_definitions.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pilot/user/atlas/loopingjob_definitions.py b/pilot/user/atlas/loopingjob_definitions.py index c78f46e1..0178639d 100644 --- a/pilot/user/atlas/loopingjob_definitions.py +++ b/pilot/user/atlas/loopingjob_definitions.py @@ -61,7 +61,8 @@ def remove_unwanted_files(workdir: str, files: list[str]) -> list[str]: "memory_" in _file or "mem." in _file or "docs/" in _file or - "DBRelease-" in _file): + "DBRelease-" in _file or + _file == ""): _files.append(_file) return _files From 1672aa73fd8526ada2059b51f26eea5d5fee63fa Mon Sep 17 00:00:00 2001 From: Paul Nilsson Date: Mon, 16 Dec 2024 22:36:27 +0100 Subject: [PATCH 4/7] Increased sleep to 1 s --- pilot/util/container.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pilot/util/container.py b/pilot/util/container.py index d4b1c627..7acfcf6f 100644 --- a/pilot/util/container.py +++ b/pilot/util/container.py @@ -113,7 +113,7 @@ def execute(executable: Any, **kwargs: dict) -> Any: # noqa: C901 def read_output(stream, queue): while True: - sleep(0.01) + sleep(1) try: line = stream.readline() if not line: From 13c79cce8adbaa922e37c8ba0974842026c6102e Mon Sep 17 00:00:00 2001 From: Paul Nilsson Date: Tue, 17 Dec 2024 12:37:54 +0100 Subject: [PATCH 5/7] Added new readout function --- PILOTVERSION | 2 +- pilot/util/constants.py | 2 +- pilot/util/container.py | 27 +++++++++++++++++++++++++-- 3 files changed, 27 insertions(+), 4 deletions(-) diff --git a/PILOTVERSION b/PILOTVERSION index 5ae185b5..6851997d 100644 --- a/PILOTVERSION +++ b/PILOTVERSION @@ -1 +1 @@ -3.9.4.1 \ No newline at end of file +3.9.4.9 \ No newline at end of file diff --git a/pilot/util/constants.py b/pilot/util/constants.py index 4c792f0f..f225b448 100644 --- a/pilot/util/constants.py +++ b/pilot/util/constants.py @@ -28,7 +28,7 @@ RELEASE = '3' # released number should be fixed at 3 for Pilot 3 VERSION = '9' # version number is '1' for first release, '0' until then, increased for bigger updates REVISION = '4' # revision number should be reset to '0' for every new version release, increased for small updates -BUILD = '1' # build number should be reset to '1' for every new development cycle +BUILD = '9' # build number should be reset to '1' for every new development cycle SUCCESS = 0 FAILURE = 1 diff --git a/pilot/util/container.py b/pilot/util/container.py index 7acfcf6f..f65058c4 100644 --- a/pilot/util/container.py +++ b/pilot/util/container.py @@ -27,6 +27,7 @@ import logging import queue import re +import select import shlex import signal import threading @@ -113,7 +114,28 @@ def execute(executable: Any, **kwargs: dict) -> Any: # noqa: C901 def read_output(stream, queue): while True: - sleep(1) + try: + # Use select to wait for the stream to be ready for reading + ready, _, _ = select.select([stream], [], [], 1.0) + if ready: + line = stream.readline() + if not line: + break + try: + queue.put_nowait(line) + except queue.Full: + pass # Handle the case where the queue is full + except (AttributeError, ValueError): + break + except OSError as e: + if e.errno == errno.EBADF: + break + else: + raise + + def read_output_old(stream, queue): + while True: + #sleep(1) try: line = stream.readline() if not line: @@ -130,7 +152,8 @@ def read_output(stream, queue): try: queue.put_nowait(line) except queue.Full: - sleep(0.01) # Sleep for a short interval to avoid busy waiting + pass + #sleep(0.01) # Sleep for a short interval to avoid busy waiting stdout_thread = threading.Thread(target=read_output, args=(process.stdout, stdout_queue)) stderr_thread = threading.Thread(target=read_output, args=(process.stderr, stderr_queue)) From f479fe90317c9ab04d13bd69ef102fe3f8bbe714 Mon Sep 17 00:00:00 2001 From: Paul Nilsson Date: Tue, 17 Dec 2024 12:38:12 +0100 Subject: [PATCH 6/7] Active CPU monitoring --- pilot/control/monitor.py | 31 ++++++++++++++----------------- pilot/util/psutils.py | 39 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 52 insertions(+), 18 deletions(-) diff --git a/pilot/control/monitor.py b/pilot/control/monitor.py index 89ff1f5f..399502fc 100644 --- a/pilot/control/monitor.py +++ b/pilot/control/monitor.py @@ -31,7 +31,7 @@ import re from collections import namedtuple -from os import environ, getuid +from os import environ, getuid, getpid from subprocess import ( Popen, PIPE @@ -53,6 +53,7 @@ get_local_oidc_token_info, update_local_oidc_token_info ) +from pilot.util.psutils import get_process_info from pilot.util.queuehandling import ( abort_jobs_in_queues, get_maxwalltime_from_job, @@ -84,8 +85,8 @@ def control(queues: namedtuple, traces: Any, args: object): # noqa: C901 last_token_check = t_0 # for CPU usage debugging - # cpuchecktime = int(config.Pilot.cpu_check) - # tcpu = t_0 + cpuchecktime = int(config.Pilot.cpu_check) + tcpu = t_0 last_minute_check = t_0 queuedata = get_queuedata_from_job(queues) @@ -158,15 +159,15 @@ def control(queues: namedtuple, traces: Any, args: object): # noqa: C901 time.sleep(1) # time to check the CPU usage? - # if is_pilot_check(check='cpu_usage'): - # if int(time.time() - tcpu) > cpuchecktime: - # processes = get_process_info('python3 pilot3/pilot.py', pid=getpid()) - # if processes: - # logger.info(f'PID={getpid()} has CPU usage={processes[0]}% CMD={processes[2]}') - # nproc = processes[3] - # if nproc > 1: - # logger.info(f'.. there are {nproc} such processes running') - # tcpu = time.time() + if is_pilot_check(check='cpu_usage'): + if int(time.time() - tcpu) > cpuchecktime: + processes = get_process_info('python3 pilot3/pilot.py', pid=getpid()) + if processes: + logger.info(f'PID={getpid()} has CPU usage={processes[0]}% CMD={processes[2]}') + nproc = processes[3] + if nproc > 1: + logger.info(f'.. there are {nproc} such processes running') + tcpu = time.time() # proceed with running the other checks run_checks(queues, args) @@ -283,11 +284,7 @@ def reached_maxtime_abort(args: Any): args.graceful_stop.set() -#def log_lifetime(sig, frame, traces): -# logger.info('lifetime: %i used, %i maximum', int(time.time() - traces.pilot['lifetime_start']), traces.pilot['lifetime_max']) - - -def get_process_info(cmd: str, user: str = "", args: str = 'aufx', pid: int = 0) -> list: +def get_process_info_old(cmd: str, user: str = "", args: str = 'aufx', pid: int = 0) -> list: """ Return process info for given command. diff --git a/pilot/util/psutils.py b/pilot/util/psutils.py index 90eb6394..95c28b88 100644 --- a/pilot/util/psutils.py +++ b/pilot/util/psutils.py @@ -17,7 +17,7 @@ # under the License. # # Authors: -# - Paul Nilsson, paul.nilsson@cern.ch, 2023 +# - Paul Nilsson, paul.nilsson@cern.ch, 2023-24 import logging import os @@ -373,3 +373,40 @@ def check_cpu_load(): else: logger.info("system load is normal") return False + + +def get_process_info(cmd: str, user: str = "", pid: int = 0) -> list: + """ + Return process info for given command. + + The function returns a list with format [cpu, mem, command, number of commands] for + a given command (e.g. python3 pilot3/pilot.py). + + :param cmd: command (str) + :param user: user (str) + :param pid: process id (int) + :return: list with process info (l[0]=cpu usage(%), l[1]=mem usage(%), l[2]=command(string)) (list). + """ + if not _is_psutil_available: + logger.warning('psutil not available, cannot check pilot CPU load') + return [] + + processes = [] + num = 0 + + for proc in psutil.process_iter(['pid', 'username', 'cpu_percent', 'memory_percent', 'cmdline']): + try: + if user and proc.info['username'] != user: + continue + cmdline = proc.info['cmdline'] + if cmdline and cmd in ' '.join(cmdline): + num += 1 + if proc.info['pid'] == pid: + processes = [proc.info['cpu_percent'], proc.info['memory_percent'], ' '.join(cmdline)] + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + continue + + if processes: + processes.append(num) + + return processes From c14a16b7c30977a05fb635084e112ed0bf288be4 Mon Sep 17 00:00:00 2001 From: Paul Nilsson Date: Wed, 18 Dec 2024 10:36:30 +0100 Subject: [PATCH 7/7] New version of execute() --- PILOTVERSION | 2 +- pilot/util/constants.py | 2 +- pilot/util/container.py | 105 +++++++++++++++++++++++++++++++--------- 3 files changed, 85 insertions(+), 24 deletions(-) diff --git a/PILOTVERSION b/PILOTVERSION index 6851997d..8d2739d3 100644 --- a/PILOTVERSION +++ b/PILOTVERSION @@ -1 +1 @@ -3.9.4.9 \ No newline at end of file +3.9.4.15 \ No newline at end of file diff --git a/pilot/util/constants.py b/pilot/util/constants.py index f225b448..b4d5ed62 100644 --- a/pilot/util/constants.py +++ b/pilot/util/constants.py @@ -28,7 +28,7 @@ RELEASE = '3' # released number should be fixed at 3 for Pilot 3 VERSION = '9' # version number is '1' for first release, '0' until then, increased for bigger updates REVISION = '4' # revision number should be reset to '0' for every new version release, increased for small updates -BUILD = '9' # build number should be reset to '1' for every new development cycle +BUILD = '15' # build number should be reset to '1' for every new development cycle SUCCESS = 0 FAILURE = 1 diff --git a/pilot/util/container.py b/pilot/util/container.py index f65058c4..6ba77ad0 100644 --- a/pilot/util/container.py +++ b/pilot/util/container.py @@ -52,6 +52,86 @@ def execute(executable: Any, **kwargs: dict) -> Any: # noqa: C901 """ Executes the command with its options in the provided executable list using subprocess time-out handler. + :param executable: command to be executed (str or list) + :param kwargs: kwargs (dict) + :return: exit code (int), stdout (str) and stderr (str) (or process if requested via returnproc argument). + """ + usecontainer = kwargs.get('usecontainer', False) + job = kwargs.get('job') + obscure = kwargs.get('obscure', '') # if this string is set, hide it in the log message + + # Convert executable to string if it is a list + if isinstance(executable, list): + executable = ' '.join(executable) + + if job and job.imagename != "" and "runcontainer" in executable: + usecontainer = False + job.usecontainer = usecontainer + + if usecontainer: + executable, diagnostics = containerise_executable(executable, **kwargs) + if not executable: + return None if kwargs.get('returnproc', False) else -1, "", diagnostics + + if not kwargs.get('mute', False): + print_executable(executable, obscure=obscure) + + timeout = get_timeout(kwargs.get('timeout', None)) + exe = ['/usr/bin/python'] + executable.split() if kwargs.get('mode', 'bash') == 'python' else ['/bin/bash', '-c', executable] + + process = None + try: + with execute_lock: + process = subprocess.Popen( + exe, + bufsize=-1, + stdout=kwargs.get('stdout', subprocess.PIPE), + stderr=kwargs.get('stderr', subprocess.PIPE), + cwd=kwargs.get('cwd', getcwd()), + start_new_session=True, + encoding='utf-8', + errors='replace' + ) + if kwargs.get('returnproc', False): + return process + + # Use communicate() to read stdout and stderr reliably + try: + logger.debug(f'subprocess.communicate() will use timeout {timeout} s') + stdout, stderr = process.communicate(timeout=timeout) + except subprocess.TimeoutExpired as exc: + # Timeout handling + stderr = f'subprocess communicate sent TimeoutExpired: {exc}' + logger.warning(stderr) + exit_code = errors.COMMANDTIMEDOUT + stderr = kill_all(process, stderr) + return exit_code, "", stderr + except Exception as exc: + logger.warning(f'exception caused when executing command: {executable}: {exc}') + exit_code = errors.UNKNOWNEXCEPTION + stderr = kill_all(process, str(exc)) + return exit_code, "", stderr + + exit_code = process.poll() + if stdout and stdout.endswith('\n'): + stdout = stdout[:-1] + + return exit_code, stdout, stderr + + finally: + # Ensure the process is cleaned up + if process and not kwargs.get('returnproc', False): + try: + process.wait(timeout=60) + process.stdout.close() + process.stderr.close() + except Exception: + pass + + +def execute_old3(executable: Any, **kwargs: dict) -> Any: # noqa: C901 + """ + Executes the command with its options in the provided executable list using subprocess time-out handler. The function also determines whether the command should be executed within a container. @@ -125,35 +205,15 @@ def read_output(stream, queue): queue.put_nowait(line) except queue.Full: pass # Handle the case where the queue is full - except (AttributeError, ValueError): - break - except OSError as e: - if e.errno == errno.EBADF: - break else: - raise - - def read_output_old(stream, queue): - while True: - #sleep(1) - try: - line = stream.readline() - if not line: - break + sleep(0.01) # Sleep for a short interval to avoid busy waiting except (AttributeError, ValueError): - # Handle the case where stream is None (AttributeError) or closed (ValueError) break except OSError as e: if e.errno == errno.EBADF: - # Handle the case where the file descriptor is bad break else: raise - try: - queue.put_nowait(line) - except queue.Full: - pass - #sleep(0.01) # Sleep for a short interval to avoid busy waiting stdout_thread = threading.Thread(target=read_output, args=(process.stdout, stdout_queue)) stderr_thread = threading.Thread(target=read_output, args=(process.stderr, stderr_queue)) @@ -178,7 +238,8 @@ def read_output_old(stream, queue): exit_code = errors.UNKNOWNEXCEPTION stderr = kill_all(process, str(exc)) else: - exit_code = process.poll() + #exit_code = process.poll() + exit_code = process.returncode # Wait for the threads to finish reading try: