1.3.2

Author: vinifmor

CHANGELOG.md

@@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## [1.3.2] 2023-01-30
+### Improvements
+- replaced some subprocess calls executed in behalf of non-root users by async calls (Python's asyncio native approach)
+
+### Fixes
+- the optimizer service (as root) not able to execute some commands in behalf of non-root users (started with Python 3.10.9)
+
 ## [1.3.1] 2022-10-22
 ### Improvements
 - Optimizing the children of the target process

LICENSE

@@ -1,4 +1,4 @@
-                  Copyright (c) 2021-2023 Vinícius Moreira
+                  Copyright (c) 2021 Vinícius Moreira
 
                           zlib/libpng license
 

guapow/__init__.py

@@ -2,4 +2,4 @@
 
 ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
 __app_name__ = 'guapow'
-__version__ = '1.3.1'
+__version__ = '1.3.2'

guapow/common/scripts.py

@@ -1,14 +1,9 @@
-import asyncio
-import datetime
 import os
-import subprocess
 from logging import Logger
-from multiprocessing import Process
 from typing import Optional, List, Dict, Set
 
-from guapow.common import system
 from guapow.common.model import ScriptSettings
-from guapow.common.system import run_user_command
+from guapow.common.system import run_async_process, ProcessTimedOutError
 from guapow.common.users import is_root_user
 
 
@@ -19,31 +14,6 @@ def __init__(self, name: str, root_allowed: bool, logger: Logger):
         self._log = logger
         self.root_allowed = root_allowed
 
-    async def _execute_user_scripts(self, settings: ScriptSettings, user_id: int, user_env: Optional[Dict[str, str]]) -> Set[int]:
-        pids = set()
-
-        for cmd in settings.scripts:
-            res = system.new_user_process_response()
-
-            self._log.info(f"Running {self._name} script '{cmd}' (user={user_id})")
-            p = Process(daemon=True, target=run_user_command, args=(cmd, user_id, settings.wait_execution, settings.timeout, user_env, res))
-            p.start()
-
-            if settings.wait_execution or (settings.timeout is not None and settings.timeout > 0):
-                self._log.info(f"Waiting {self._name} script '{cmd}' to finish (user={user_id})")
-
-                while p.is_alive():  # to allow other async tasks to execute
-                    await asyncio.sleep(0.0005)
-
-                self._log.info(f"{self._name.capitalize()} script '{cmd}' finished (user={user_id})")
-
-            p.join()
-
-            if res['pid'] is not None:
-                pids.add(res['pid'])
-
-        return pids
-
     @staticmethod
     def get_environ(env: Optional[Dict[str, str]] = None) -> Dict[str, str]:
         final_env = env if env else dict(os.environ)
@@ -53,42 +23,40 @@ def get_environ(env: Optional[Dict[str, str]] = None) -> Dict[str, str]:
 
         return final_env
 
-    async def _execute_scripts(self, settings: ScriptSettings, user_env: Optional[Dict[str, str]]) -> Set[int]:
+    async def _execute_scripts(self, settings: ScriptSettings, user_id: Optional[int] = None,
+                               user_env: Optional[Dict[str, str]] = None) -> Set[int]:
         pids = set()
         env = self.get_environ(user_env)
 
+        valid_timeout = settings.has_valid_timeout()
+
+        if not valid_timeout and settings.timeout is not None:
+            self._log.warning(f"Invalid {self._name} scripts timeout defined: {settings.timeout}. "
+                              f"No script will be awaited")
+
+        should_wait = settings.wait_execution or valid_timeout
+
         for cmd in settings.scripts:
-            p = await asyncio.create_subprocess_shell(cmd=cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
-                                                      stderr=subprocess.DEVNULL, env=env)
-            pids.add(p.pid)
-            self._log.info(f"Started {self._name} script: {cmd} (pid={p.pid})")
-
-            if settings.timeout is not None:
-                if not settings.has_valid_timeout():
-                    self._log.warning(f"Invalid {self._name} script timeout defined: {settings.timeout}. It will not be awaited")
-                    continue
-
-                self._log.info(f"Waiting {self._name} script '{cmd}' to finish (pid={p.pid})")
-                timeout = datetime.datetime.now() + datetime.timedelta(seconds=settings.timeout)
-
-                timed_out = True
-                while timeout > datetime.datetime.now():
-                    if p.returncode is None:
-                        await asyncio.sleep(0.001)
-                    else:
-                        timed_out = False
-                        break
-
-                if timed_out:
-                    self._log.warning(f"{self._name.capitalize()} script '{cmd}' timed out (pid={p.pid})")
-
-            elif settings.wait_execution:
-                self._log.info(f"Waiting {self._name} script '{cmd}' to finish (pid={p.pid})")
-                await p.wait()
+            self._log.info(f"{'Waiting' if should_wait else 'Starting'} {self._name} script: {cmd}")
+            try:
+                pid, _, output = await run_async_process(cmd=cmd, user_id=user_id, custom_env=env,
+                                                         wait=settings.wait_execution, timeout=settings.timeout,
+                                                         output=False)
+
+                if pid is not None:
+                    pids.add(pid)
+                    if should_wait:
+                        self._log.info(f"{self._name.capitalize()} script finished: {cmd} (pid={pid})")
+                else:
+                    err_output = f": {output}" if output else ""
+                    self._log.error(f"Unexpected error when running {self._name} script '{cmd}'{err_output}")
+            except ProcessTimedOutError as e:
+                self._log.warning(f"{self._name.capitalize()} script '{cmd}' timed out (pid={e.pid})")
 
         return pids
 
-    async def run(self, scripts: List[ScriptSettings], user_id: Optional[int], user_env: Optional[Dict[str, str]]) -> Set[int]:
+    async def run(self, scripts: List[ScriptSettings], user_id: Optional[int],
+                  user_env: Optional[Dict[str, str]]) -> Set[int]:
         current_user_id = os.getuid()
         root_user = is_root_user(current_user_id)
 
@@ -97,17 +65,18 @@ async def run(self, scripts: List[ScriptSettings], user_id: Optional[int], user_
         for settings in scripts:
             if root_user:
                 if not settings.run_as_root and user_id is not None and not is_root_user(user_id):
-                    pids.update(await self._execute_user_scripts(settings, user_id, user_env))
+                    pids.update(await self._execute_scripts(settings, user_id, user_env))
                 elif self.root_allowed:
-                    pids.update(await self._execute_scripts(settings, None))
+                    pids.update(await self._execute_scripts(settings))
                 else:
-                    self._log.warning(f"{self._name.capitalize()} scripts {settings.scripts} are not allowed to run at the root level")
+                    self._log.warning(f"{self._name.capitalize()} scripts {settings.scripts} are not allowed "
+                                      f"to run at the root level")
             elif settings.run_as_root:
                 self._log.warning(f"Cannot execute {self._name} scripts {settings.scripts} as root user")
             elif user_id is None:
-                pids.update(await self._execute_scripts(settings, None))
+                pids.update(await self._execute_scripts(settings))
             elif current_user_id == user_id:
-                pids.update(await self._execute_scripts(settings, user_env))
+                pids.update(await self._execute_scripts(settings, user_env=user_env))
             else:
                 self._log.warning(f"Cannot execute {self._name} scripts {settings.scripts} as user {user_id}")
 

guapow/common/system.py

@@ -2,16 +2,21 @@
 import os
 import subprocess
 import traceback
+from datetime import datetime, timedelta
 from io import StringIO
-from multiprocessing import Manager, Process
-from multiprocessing.managers import DictProxy
 from re import Pattern
 from typing import Optional, Tuple, Dict, List, Set, Callable, TypeVar, Collection, Generator
 
 BAD_USER_ENV_VARS = {'LD_PRELOAD'}
 T = TypeVar('T')
 
 
+class ProcessTimedOutError(Exception):
+    def __init__(self, pid: int):
+        super(ProcessTimedOutError, self).__init__()
+        self.pid = pid
+
+
 def syscall(cmd: str, shell: bool = True, cwd: Optional[str] = None, custom_env: Optional[dict] = None, stdin: bool = False, return_output: bool = True) -> Tuple[int, Optional[str]]:
     params = {
         'args': cmd.split(' ') if not shell else [cmd],
@@ -276,72 +281,70 @@ async def find_children(ppids: Set[int], ppid_map: Optional[Dict[int, Set[int]]]
     return children_list
 
 
-def run_user_command(cmd: str, user_id: int, wait: bool, timeout: Optional[float] = None,
-                     env: Optional[dict] = None, response: Optional[DictProxy] = None, forbidden_env_vars: Optional[Set[str]] = BAD_USER_ENV_VARS):
-    args = {"args": cmd, "shell": True, "stdin": subprocess.DEVNULL,
-            "stdout": subprocess.PIPE if wait else subprocess.DEVNULL,
-            "stderr": subprocess.STDOUT if wait else subprocess.DEVNULL}
-
-    if env:
+async def run_async_process(cmd: str, user_id: Optional[int] = None, custom_env: Optional[dict] = None,
+                            forbidden_env_vars: Optional[Set[str]] = BAD_USER_ENV_VARS,
+                            wait: bool = True, timeout: Optional[float] = None, output: bool = True,
+                            exception_output: bool = True) \
+        -> Tuple[Optional[int], Optional[int], Optional[str]]:
+    """
+    Runs a process using the async API
+    Args:
+        cmd:
+        user_id: if the process should be executed in behalf of a different user
+        custom_env: custom environment variables available for the process to be executed
+        forbidden_env_vars: environment variables that should not be passed to the process
+        wait: if the process should be waited
+        timeout: in seconds
+        output: if the process output should be read and returned
+        exception_output: if the traceback of an unexpected raised exception should be returned as the output
+    Returns: a tuple containing the process id, exitcode and output as a String
+
+    """
+    args = {"cmd": cmd, "stdin": subprocess.DEVNULL,
+            "stdout": subprocess.PIPE if output else subprocess.DEVNULL,
+            "stderr": subprocess.STDOUT if output else subprocess.DEVNULL}
+
+    if user_id is not None:
+        args["preexec_fn"] = lambda: os.setuid(user_id)
+
+    if custom_env:
         if forbidden_env_vars:
-            args['env'] = {k: v for k, v in env.items() if k not in forbidden_env_vars}
+            args['env'] = {k: v for k, v in custom_env.items() if k not in forbidden_env_vars}
         else:
-            args['env'] = env
+            args['env'] = custom_env
 
     try:
-        os.setpriority(os.PRIO_PROCESS, os.getpid(), 0)  # always launch a command with nice 0
-        os.setuid(user_id)
-
-        p = subprocess.Popen(**args)
-
-        if response is not None:
-            response['pid'] = p.pid
-
-        if timeout is not None and timeout > 0:
-            p.wait(timeout)
-        elif wait:
-            p.wait()
-
-        if response is not None:
-            response['exitcode'] = p.returncode
+        p = await asyncio.create_subprocess_shell(**args)
+    except Exception:
+        return None, 1, (traceback.format_exc().replace('\n', ' ') if exception_output else None)
 
-            string = StringIO()
-            for output in p.stdout:
-                decoded = output.decode()
-                string.write(decoded)
-
-            string.seek(0)
-            response['output'] = string.read()
-
-    except Exception as e:
-        if response is not None:
-            response['exitcode'] = 1
-            response['output'] = traceback.format_exc()
-
-
-async def run_async_user_process(cmd: str, user_id: int, user_env: Optional[dict], forbidden_env_vars: Optional[Set[str]] = BAD_USER_ENV_VARS) -> Tuple[int, Optional[str]]:
-    res = new_user_process_response()
+    if user_id is not None:  # set default niceness in case the process is executed in behalf of another user
+        try:
+            os.setpriority(os.PRIO_PROCESS, p.pid, 0)  # always launch a command with nice 0
+        except Exception:
+            pass  # do nothing in case the priority could not be changed
 
+    should_wait = wait or (timeout and timeout > 0)
     try:
-        p = Process(target=run_user_command, kwargs={'cmd': cmd, 'user_id': user_id, 'wait': True, 'timeout': None, 'env': user_env,
-                                                     'response': res, 'forbidden_env_vars': forbidden_env_vars})
-        p.start()
+        if should_wait:
+            if timeout is None or timeout < 0:
+                return p.pid, await p.wait(), ((await p.stdout.read()).decode() if output else None)
+            elif timeout and timeout > 0:
+                timeout_at = datetime.now() + timedelta(seconds=timeout)
 
-        while p.is_alive():
-            await asyncio.sleep(0.001)
+                while datetime.now() < timeout_at:
+                    if p.returncode is not None:
+                        return p.pid, p.returncode, ((await p.stdout.read()).decode() if output else None)
 
-        return res['exitcode'], res['output']
-    except:
-        error_msg = traceback.format_exc().replace('\n', ' ')
-        return 1, error_msg
+                    await asyncio.sleep(0.0005)
 
+                raise ProcessTimedOutError(p.pid)
 
-def new_user_process_response() -> DictProxy:
-    proxy_res = Manager().dict()
-    proxy_res['exitcode'] = None
-    proxy_res['output'] = None
-    proxy_res['pid'] = None
-    return proxy_res
+        return p.pid, p.returncode, None
+    except ProcessTimedOutError:
+        raise
+    except Exception:
+        return p.pid, 1, (traceback.format_exc().replace('\n', ' ') if exception_output else None)
 
 
 async def map_processes_by_parent() -> Dict[int, Set[Tuple[int, str]]]:

guapow/service/optimizer/post_process/task.py

@@ -1,13 +1,12 @@
 import asyncio
 import re
-import traceback
 from abc import ABC, abstractmethod
-from multiprocessing import Process
 from re import Pattern
 from typing import List, Optional, Dict, Set, Tuple, Type, Awaitable
 
 from guapow.common import system
 from guapow.common.scripts import RunScripts
+from guapow.common.system import run_async_process
 from guapow.common.users import is_root_user
 from guapow.service.optimizer.gpu import GPUState, GPUDriver
 from guapow.service.optimizer.post_process.context import PostProcessContext
@@ -273,28 +272,21 @@ def get_python_cmd_pattern(self) -> Pattern:
     def should_run(self, context: PostProcessContext) -> bool:
         return bool(context.stopped_processes and context.user_id is not None)
 
-    async def _run_command(self, name: str, cmd: str):
-        try:
-            await system.async_syscall(cmd, return_output=False, wait=False)
-            self._log.info(f"Process '{name}' ({cmd}) relaunched")
-        except:
-            stack_log = traceback.format_exc().replace('\n', ' ')
-            self._log.warning(f"An exception happened when relaunching process '{name}' ({cmd}): {stack_log}")
-
-    def _run_user_command(self, name: str, cmd: str, user_id: int, user_env: Optional[Dict[str, str]] = None):
-        try:
-            Process(daemon=True, target=system.run_user_command, kwargs={'cmd': cmd, 'user_id': user_id, 'env': user_env, 'wait': False}).start()
-            self._log.info(f"Process '{name}' ({cmd}) relaunched (user={user_id})")
-        except:
-            stack_log = traceback.format_exc().replace('\n', ' ')
-            self._log.warning(f"An exception happened when relaunching process '{name}' ({cmd}) [user={user_id}]: {stack_log}")
+    async def _run_command(self, name: str, cmd: str, user_id: Optional[int] = None,
+                           user_env: Optional[Dict[str, str]] = None):
+        await run_async_process(cmd=cmd, user_id=user_id, custom_env=user_env, wait=False, output=False,
+                                exception_output=False)
+
+        user_log = f"(user={user_id})" if user_id is not None else ""
+        self._log.info(f"Process '{name}' ({cmd}) relaunched {user_log}")
 
     async def run(self, context: PostProcessContext):
         self_is_root = is_root_user()
         root_request = is_root_user(context.user_id)
 
         if not self_is_root and root_request:
-            self._log.warning(f"It will not be possible to launch the following root processes: {', '.join((c[0] for c in context.stopped_processes))}")
+            self._log.warning(f"It will not be possible to launch the following root processes: "
+                              f"{', '.join((c[0] for c in context.stopped_processes))}")
             return
 
         running_cmds = await system.find_processes_by_command({p[1] for p in context.stopped_processes})
@@ -309,11 +301,8 @@ async def run(self, context: PostProcessContext):
 
             real_cmd = python_cmd[0] if python_cmd else cmd
 
-            if self_is_root:
-                if root_request:
-                    await self._run_command(name, real_cmd)
-                else:
-                    self._run_user_command(name, real_cmd, context.user_id, context.user_env)
+            if self_is_root and not root_request:
+                await self._run_command(name, real_cmd, context.user_id, context.user_env)
             else:
                 await self._run_command(name, real_cmd)