import argparse import logging import logging.handlers import os import traceback from typing import Callable import psutil import shutil import subprocess import sys import json import threading import time import fcntl from types import MethodType from packaging.version import parse try: HBS_INST = os.readlink('/mnt/ext/opt/HybridBackup') except Exception: HBS_INST = '/mnt/ext/opt/HybridBackup' RR2_DIR = os.path.join(HBS_INST, 'rr2') BACKGROUND_TASK_CONF_DIR = os.path.join(RR2_DIR, 'config', 'background_tasks') VERSION_STR = '{"version": "v26.0.0"}' BGLOG = logging.getLogger('Bg_task') FICLONE = 0x40049409 try: import rru.util_nas2 import rru.util_sys import rru.common1 import rru.util_file2 import rru.util_log from rru import util_file from rru import rwlock except ImportError as ex: RR2_BIN_DIR = os.path.join(RR2_DIR, 'bin') if RR2_BIN_DIR not in sys.path: sys.path.append(RR2_BIN_DIR) # --> /rr2/bin for py3' import rru.util_nas2 import rru.util_sys import rru.common1 import rru.util_file2 import rru.util_log from rru import util_file from rru import rwlock def copy_permissions_and_attributes(src, dest, logger=logging.getLogger(__name__)) -> Callable: src_stat = os.stat(src) os.chmod(dest, src_stat.st_mode) os.chown(dest, uid=src_stat.st_uid, gid=src_stat.st_gid, follow_symlinks=False) try: for xattr in os.listxattr(src): os.setxattr(dest, xattr, os.getxattr(src, xattr)) except OSError as ex: logger.exception(ex) def mtime_callback(): # logger.debug(f"copy arrtibutes {src} -> {dest}") os.utime(dest, (src_stat.st_atime, src_stat.st_mtime), follow_symlinks=False) return mtime_callback # change mtime in the end def copy_hard_link(src, dest, status_storage:dict = None, **kwargs): if is_latest_to_version_directory(src, dest) and in_worm_shared_folder(src): return copy_fileclone_and_lock(src, dest, status_storage, **kwargs) logger: logging.Logger = kwargs.get('logger', BGLOG) logger.debug("params: src={0}, dest={1}, status_storage={2}".format(src, dest, status_storage)) try: # check dest existence if os.path.isfile(dest) or os.path.islink(dest): os.remove(dest) if os.path.isdir(dest): shutil.rmtree(dest) folder_mtime_stack = [] def __create_parent_folder(dst_dir_path, src_dir_path): dst_parent_dir, src_parent_dir = os.path.dirname(dst_dir_path), os.path.dirname(src_dir_path) dir_pairs_to_create = [] while not os.path.exists(dst_parent_dir): if not os.path.exists(os.path.dirname(dest)): # share folder missing issue raise FileNotFoundError(dest) dir_pairs_to_create.append({'dst': dst_parent_dir, 'src': src_parent_dir}) dst_parent_dir, src_parent_dir = os.path.dirname(dst_parent_dir), os.path.dirname(src_parent_dir) for dir_pair in dir_pairs_to_create[::-1]: os.mkdir(dir_pair['dst']) folder_mtime_stack.append(copy_permissions_and_attributes(dir_pair['src'], dir_pair['dst'], logger=logger)) _ = os.mkdir(dst_dir_path) if not os.path.exists(dst_dir_path) else None if os.path.isdir(src) and not os.path.islink(src): os.mkdir(dest) # shutil.copytree(src, dest, symlinks=False, copy_function=os.link) if isinstance(status_storage, dict): status_storage.update({'processed_file': 0}) for root, dirs, files in os.walk(src, followlinks=False, topdown=True): for dir in dirs: src_dir_path = os.path.join(root, dir) dst_dir_path = os.path.join(dest, os.path.relpath(src_dir_path, src)) if os.path.islink(src_dir_path): dirs.remove(dir) files.append(dir) continue if os.path.isfile(dst_dir_path) or os.path.islink(dst_dir_path): os.remove(dst_dir_path) if not os.path.exists(dst_dir_path): try: os.mkdir(dst_dir_path) except FileNotFoundError: # Sometimes python os.walk will skip some parent folder, so we need to create it logger.debug(f"dir {dst_dir_path} parent not exist") __create_parent_folder(dst_dir_path, src_dir_path) folder_mtime_stack.append(copy_permissions_and_attributes(src_dir_path, dst_dir_path, logger=logger)) for file in files: src_file_path = os.path.join(root, file) dst_file_path = os.path.join(dest, os.path.relpath(src_file_path, src)) if not os.path.exists(os.path.dirname(dst_file_path)): __create_parent_folder(os.path.dirname(dst_file_path), os.path.dirname(src_file_path)) os.link(src_file_path, dst_file_path) if isinstance(status_storage, dict): status_storage['processed_file'] += 1 s = time.time() for f in folder_mtime_stack[::-1]: f() logger.debug(f"changed folder mtime in {time.time()- s} second, total {len(folder_mtime_stack)} folders") else: if os.path.islink(dest) or os.path.exists(dest): os.remove(dest) os.link(src, dest) except Exception as ex: logger.error(f'Failed to copy hard link "{src}" to "{dest}". --- Exception:{ex}') raise ex logger.debug(f"finished link {src} -> {dest}") def copy_fileclone_and_lock(src, dest, status_storage:dict = None, **kwargs): dest_dir_name = os.path.basename(dest) datetime_str, s_type = get_backup_folder_type(dest_dir_name) if os.path.basename(src) != "latest" or s_type is None: raise Exception('expect latest -> version dir, got: {0} -> {1}'.format(src, dest)) dest = os.path.join(os.path.dirname(dest), '{0}.{1}'.format(datetime_str, "Q")) record_size = get_recordsize(src) fw_ver = util_file.get_value_from_conf('/etc/config/uLinux.conf', 'System', 'Version') fileclone = zfs_fileclone if parse(fw_ver) >= parse('5.2.0') else zfs_fileclone_old logger: logging.Logger = kwargs.get('logger', logging.getLogger(__name__)) try: # check dest existence if os.path.isfile(dest) or os.path.islink(dest): os.remove(dest) if os.path.isdir(dest): shutil.rmtree(dest) if isinstance(status_storage, dict): status_storage.update({'processed_file': 0}) # only use fileclone for hard link latest -> version directory if os.path.isdir(src) and not os.path.islink(src): os.mkdir(dest) # shutil.copytree(src, dest, symlinks=False, copy_function=os.link) folder_mtime_stack = [] for root, dirs, files in os.walk(src, followlinks=False): for dir in dirs: src_dir_path = os.path.join(root, dir) dst_dir_path = os.path.join(dest, os.path.relpath(src_dir_path, src)) if os.path.islink(src_dir_path): dirs.remove(dir) files.append(dir) continue if os.path.isfile(dst_dir_path) or os.path.islink(dst_dir_path): os.remove(dst_dir_path) if not os.path.exists(dst_dir_path): try: os.mkdir(dst_dir_path) except FileNotFoundError: # Sometimes python os.walk will skip some parent folder, so we need to create it logger.debug(f"dir {dst_dir_path} parent not exist") dst_parent_dir, src_parent_dir = os.path.dirname(dst_dir_path), os.path.dirname(src_dir_path) dir_pairs_to_create = [] while not os.path.exists(dst_parent_dir): dir_pairs_to_create.append({'dst': dst_parent_dir, 'src': src_parent_dir}) dst_parent_dir, src_parent_dir = os.path.dirname(dst_parent_dir), os.path.dirname(src_parent_dir) for dir_pair in dir_pairs_to_create[::-1]: os.mkdir(dir_pair['dst']) folder_mtime_stack.append(copy_permissions_and_attributes(dir_pair['src'], dir_pair['dst'], logger=logger)) _ = os.mkdir(dst_dir_path) if not os.path.exists(dst_dir_path) else None folder_mtime_stack.append(copy_permissions_and_attributes(src_dir_path, dst_dir_path, logger=logger)) for file in files: src_file_path = os.path.join(root, file) dst_file_path = os.path.join(dest, os.path.relpath(src_file_path, src)) os.makedirs(os.path.dirname(dst_file_path), exist_ok=True) fileclone(src_file_path, dst_file_path, record_size) copy_permissions_and_attributes(src_file_path, dst_file_path, logger=logger)() if isinstance(status_storage, dict): status_storage['processed_file'] += 1 s = time.time() for f in folder_mtime_stack[::-1]: f() logger.debug(f"changed folder mtime in {time.time()- s} second, total {len(folder_mtime_stack)} folders") else: if os.path.islink(dest) or os.path.exists(dest): os.remove(dest) os.link(src, dest) except Exception as ex: logger.error(f'Failed to copy hard link "{src}" to "{dest}". --- Exception:{ex}') if os.path.isdir(dest): shutil.rmtree(dest) raise ex lock_worm_folder(dest) logger.debug(f"finished link {src} -> {dest}") def in_worm_shared_folder(path): # only zfs has worm shared folder if not os.path.exists('/etc/QTS_ZFS'): return False try: out = subprocess.check_output(['zfs', 'get', '-H', '-o', 'value', 'wormtype', path]) return any(x in out.decode('utf-8') for x in ['enterprise', 'compliance']) except Exception: return False def get_recordsize(path): try: command = ["zfs", "get", "recordsize", "-H", "-o", "value", "-p", path] output = subprocess.check_output(command, text=True) record_size_bytes = int(output.strip()) except (subprocess.CalledProcessError, ValueError): # If there is an error or unexpected output, use the default value record_size_bytes = 128 * 1024 # 128K in bytes return record_size_bytes def zfs_fileclone(src, dest, record_size=None): # zfs fileclone does not support symlink if os.path.islink(src): target = os.readlink(src) os.symlink(target, dest) # zfs filecone not work for file size <= record_size, use copyfile instead if record_size is not None and os.path.getsize(src) <= record_size: shutil.copyfile(src, dest) return fd_in = os.open(src, os.O_RDONLY) if fd_in == -1: raise Exception('could not open src file: {0}'.format(src)) fd_out = os.open(dest, os.O_WRONLY | os.O_CREAT) if fd_out == -1: os.close(fd_in) raise Exception('could not open dest file: {0}'.format(src)) if fcntl.ioctl(fd_out, FICLONE, fd_in) == -1: os.close(fd_in) os.close(fd_out) raise Exception('fileclone: {0} to {1} failed'.format(src, dest)) else: os.close(fd_in) os.close(fd_out) return def zfs_fileclone_old(src, dest, record_size=None): # zfs fileclone does not support symlink if os.path.islink(src): target = os.readlink(src) os.symlink(target, dest) # zfs filecone not work for file size <= record_size, use copyfile instead if record_size is not None and os.path.getsize(src) <= record_size: shutil.copyfile(src, dest) return code = subprocess.run(['zfs', 'fileclone', src, dest]).returncode if code != 0: raise Exception('zfs fileclone return code: {0}'.format(code)) def lock_worm_folder(path): code = subprocess.run(['chattr', '+i', '-R', path]).returncode if code != 0: raise Exception('lock_worm_folder return code: {0}'.format(code)) def is_latest_to_version_directory(src, dest): src_dir = os.path.basename(src) dest_dir = os.path.basename(dest) _, s_type = get_backup_folder_type(dest_dir) if src_dir == "latest" and s_type is not None: return True else: return False def get_backup_folder_type(name): ''' check if name is started with time and in format '%Y%m%d%H%M', following with versioning char Return the time data and version type. ex: name=202312010630.H ; return 202312010630, H ''' if len(name) < 12 or len(name) > 14: return None, None date_time_str = name[0:12] if not date_time_str.isdigit() or name[12] != '.': return None, None s_type = name[13] date_time_str = name[0:12] return date_time_str, s_type try: from rru.util_file2 import write_file_with_lock_timeout except ImportError: def write_file_with_lock_timeout(store_file:str, data: str, timeout=60): """ Write string to file with file-lock, with timeout. Args: store_file: the file to store data data: the string to store timeout: delete the file lock and renew one if acquired failed after timeout Returns: None Raises: Exception with failed string """ try: with rwlock.FileLockRenew( store_file, timeout=timeout, delay=1, lock_file_contents=None, renew=True, blocking=True): with open(store_file, 'w') as writefile: writefile.write(data) except Exception: # pylint: disable=broad-except # --> remove .lock due to exception rwlock.FileLockRenew.purge(store_file) raise def init_logger(log_path: str = ''): if BGLOG.hasHandlers(): return if not log_path: log_path = "/tmp" if os.path.exists(RR2_DIR): log_path = RR2_DIR log_sub = ['logs', 'background_tasks'] for sub in log_sub: _curr_path = os.path.join(log_path, sub) if not os.path.exists(_curr_path): os.mkdir(_curr_path) log_path = _curr_path log_format = '%(asctime)s(%(levelname)s)::[%(module)s(%(lineno)s)]::PID(%(process)d): %(message)s' BGLOG.setLevel(logging.DEBUG) fh = logging.handlers.RotatingFileHandler( os.path.join(log_path, "bg_task.log"), maxBytes=2**20, backupCount=5) fh.setLevel(logging.DEBUG) fh.setFormatter(logging.Formatter(log_format)) BGLOG.addHandler(fh) def create_background_task_dir() -> str: """ create /mnt/ext/opt/HybridBackup/rr2/config/background_tasks folder if not exist. return "/mnt/ext/opt/HybridBackup/rr2/config/background_tasks" """ if not os.path.exists(RR2_DIR): raise FileNotFoundError(f"{RR2_DIR} not exist") base_path = RR2_DIR for sub_dir in ['config', 'background_tasks']: base_path = os.path.join(base_path, sub_dir) if not os.path.exists(base_path): os.mkdir(base_path) return base_path class Rr2PluginBackgroundHandler(): def __init__(self, task_id: str=None) -> None: self.bg_task_path = create_background_task_dir() self.task_id = task_id self.config_file = os.path.join(self.bg_task_path, "background_tasks.json") self.plugin_func_maps = { "run_cmd": rru.util_sys.run_cmdline_get_lines, "hardlink_cp": lambda src, dest, **kwargs: copy_hard_link(src, dest, logger=BGLOG, **kwargs), "remove_dir": shutil.rmtree, "test_bg_sleep": self._test_func } @staticmethod def _test_func(x: int=10): time.sleep(x) return x**2 def _clean_task_configs(self): BGLOG.debug("clean task config...") try: now = time.time() for file_name in os.listdir(self.bg_task_path): if not file_name.endswith('.json'): continue file_path = os.path.join(self.bg_task_path, file_name) if os.path.isfile(file_path): mtime = os.path.getmtime(file_path) if now - mtime > 2592000: # 30*24*60*60, clean file elder than 30 days BGLOG.debug(f'clean outdated task config:{file_name} ') os.remove(file_path) except Exception as ex: BGLOG.exception(f"clean task config error: {ex}") def _list_plugins(self): if self.plugin_func_maps: return self.plugin_func_maps.keys() raise ValueError("no support plugins") def _write_json_to_config(self, data: dict, conf_path: str = None): if conf_path is None: conf_path = self.config_file retry = 5 while retry: try: write_file_with_lock_timeout(conf_path, json.dumps(data, indent=4)) return except Exception as ex: retry -= 1 err_msg = f'write config error, config_path:"{conf_path}", data:{data}, ex:{ex}' BGLOG.error(err_msg) if retry <= 0: raise ex.__class__(f'write config error, config_path:"{conf_path}", data:{data}, ex:{ex}') time.sleep(5-retry) def __child_process(self, func: MethodType, *func_args, **func_kwargs): """ This function should run in child process """ # Update task info start_time = time.time() confs = rru.util_file2.read_json_file_with_lock_timeout(self.config_file) task_conf: dict = confs.get(self.task_id, None) if task_conf is None: BGLOG.warning(f"task_id:'{self.task_id}' not found in config file:'{self.config_file}', read conf:{confs}, write task info to config file") confs[self.task_id] = task_conf task_conf.update({'pid': os.getpid(), 'func_name': func.__name__, 'start_time': start_time, 'status': 'running'}) self._write_json_to_config(confs, self.config_file) ## run sub_task # threading for monitoring if task_conf['plugin_name'] == 'hardlink_cp': func_kwargs['status_storage'] = task_conf BGLOG.debug('__child_process, start run background task') self.task_result = None self.task_error = None def __thread_func_wrap(): try: self.task_result = func(*func_args, **func_kwargs) except Exception as ex: msg = f"{func.__name__} occur error, error:{ex}" BGLOG.exception(msg) self.task_error = {'msg': msg, 'ex': f"{ex}", 'traceback': f'{traceback.format_exc()}'} try: t = threading.Thread(name='backgound_task_thread', target=__thread_func_wrap) t.start() cnt = 0 while t.is_alive(): if cnt % 600 == 0: BGLOG.debug(f'background task still running, elapsed_time={cnt/10}s') if cnt % 20 == 0 and 'status_storage' in func_kwargs: try: self._write_json_to_config(confs, self.config_file) except Exception as e: BGLOG.exception(f"update json config error: {e}") cnt += 1 time.sleep(0.1) BGLOG.debug(f'background task finished, elapsed_time={cnt/10}s') if self.task_error: task_conf['status'] = 'failed' task_conf['error'] = self.task_error else: task_conf['status'] = 'finished' except Exception as ex: BGLOG.exception(f"background task occur error: {ex}") task_conf['status'] = 'failed' task_conf['error'] = {"exception": f'{ex}', "msg": "sub_task runtime error."} end_time = time.time() task_conf.update({'end_time': end_time, 'elapsed_time': end_time-start_time, 'result': self.task_result}) BGLOG.debug(f'finished background task, task_conf:{task_conf}') BGLOG.info(f"finished background task, task_id={self.task_id},"+ f"plugin_name:{task_conf.get('plugin_name', func.__name__)},"+ f"elapsed_time={end_time-start_time}") confs[self.task_id] = task_conf self._write_json_to_config(confs, self.config_file) def _bg_task_wrapper(self, func: MethodType, *func_args, **func_kwargs): BGLOG.debug(f"run task:{func.__name__}, args={func_args}, kwargs={func_kwargs}") child_pid = os.fork() if child_pid == 0: try: self.__child_process(func, *func_args, **func_kwargs) except Exception as ex: BGLOG.error(f"child_process occur error, highly possible is bg_task.py issue, not sub_task.\n{ex}") confs: dict = rru.util_file2.read_json_file_with_lock_timeout(self.config_file) task_conf: dict = confs.get(self.task_id, {}) task_conf.update({ "status": "failed", "error": { "msg": "chile_process occur error, highly possible is bg_task.py issue, not sub_task.", "exception": f'{ex}' }}) confs[self.task_id] = task_conf self._write_json_to_config(confs, self.config_file) BGLOG.debug("child process exit") self._clean_task_configs() sys.exit() else: BGLOG.info(f"fork child_process, child_pid:{child_pid}, run task:{func.__name__}, "+ f"task_args={func_args}, task_kwargs={func_kwargs}") def run_task_from_config(self, task_id: str, config_path: str=None): """ Run a task using configuration from a specified file. Args: task_id (str): The identifier for the task to run. config_path (str, optional): The path to the configuration file. If not provided, the current configuration file of the instance will be used. Raises: ValueError: If the task configuration is not found for the specified task ID. ValueError: If the plugin name is not found in the configuration file. ModuleNotFoundError: If the specified plugin name is not found in the list of available plugins. """ BGLOG.debug(f"run task from config file: '{config_path}', task_id: '{task_id}'") if config_path: self.config_file=config_path confs = rru.util_file2.read_json_file_with_lock_timeout(self.config_file) BGLOG.info(f"read config:{confs}") self.task_id = task_id task_conf: dict = confs.get(self.task_id, None) if task_conf is None: raise ValueError(f"task_id:\"{task_id}\" not found in {config_path}, read conf: {confs}") task_status = task_conf.get('status', 'waiting') if task_status != 'waiting': raise ValueError(f"task_id:'{task_id}', status in '{task_status}', not allow to run") plugin_name: str = task_conf.get("plugin_name", "") if plugin_name.strip() == "": raise ValueError(f"plugin_name not found in config file[{config_path}] section[{task_id}]") plugin_func = self.plugin_func_maps.get(plugin_name, None) if plugin_func is None: raise ModuleNotFoundError(f"plugin_name: '{plugin_name}' not found in list, available plugins: {list(self._list_plugins())}") task_conf['status'] = 'running' plugin_kwargs=task_conf.get("plugin_args", {}) self._bg_task_wrapper(plugin_func, **plugin_kwargs) return task_conf def get_task_info(self, task_id: str, config_path: str) -> dict: # If process status is not found, and task status is running, return with error "pid not found" confs: dict = rru.util_file2.read_json_file_with_lock_timeout(config_path) task_info: dict = confs.get(task_id, None) if task_info is None: raise KeyError(f"task_id:'{task_id}' not found in config file {config_path}, read config: {confs}") pid = task_info.get('pid', None) status = task_info.get('status', None) if pid: try: process = psutil.Process(pid) p_status = process.status() task_info['process_status'] = p_status except psutil.NoSuchProcess as ex: if status == "running": task_info['error'] = {'exception': f'{ex}', 'msg': f"task might dead, status running but process not found by pid({pid})"} task_info['process_status'] = None except psutil.AccessDenied as ex: if status != 'finished': task_info['error'] = {'exception': f'{ex}', 'msg': f"check process({pid}) status occur AccessDenied error"} task_info['process_status'] = None # self._write_json_to_config(confs, config_path) # Check config should not write the result back to config, prevent race with task run. return task_info # def run_task_from_args(self, plugin_name: str, *plugin_args, **plugin_kwargs): # """ # generate task_id and return task_id # """ # plugin_func = self.plugin_func_maps.get(plugin_name, None) # if plugin_func is None: # raise ModuleNotFoundError(f"plugin_name: \"{plugin_name}\" not found in list, available plugins: {self._list_plugins()}") # self.task_id = f"{str(random.randint(0, 65536)).zfill(5)}-{time.time_ns()}" # self._bg_task_wrapper(plugin_func, *plugin_args, **plugin_kwargs) # return self.task_id def check_task(config_file: str, task_id: str): try: handler = Rr2PluginBackgroundHandler() task_info = handler.get_task_info(task_id=task_id, config_path=config_file) return task_info except Exception as ex: BGLOG.exception(f"check_task with error: {ex}") return {"error": { "exception": f"{ex}", "msg": f"get_task_info with job_id=({task_id}), config_file=({config_file}) error" }} def cli_check_task(args): config_file = args.config task_id = args.task_id task_info = check_task(config_file, task_id) try: print(json.dumps(task_info, indent=4)) except Exception as ex: msg = f"dump task_info to json error, task_info={task_info}, error:{ex}" BGLOG.exception(msg) print(json.dumps({'error': {'ex': f'{ex}', 'msg': msg}}, indent=4)) def run_task(config_file: str, task_id: str, plugin_name: str = None, plugin_args: dict = None): init_logger() try: handler = Rr2PluginBackgroundHandler() task_info = handler.run_task_from_config(task_id=task_id, config_path=config_file) return task_info except Exception as ex: BGLOG.exception(f"run_task with error: {ex}") return {'error': {'ex': f'{ex}', 'msg': f"cli run_task with error: {ex}"}} def cli_run_task(args): config_file = args.config task_id = args.task_id plugin_name = args.plugin_name plugin_args = args.plugin_args task_info = run_task(config_file, task_id, plugin_name, plugin_args) try: print(json.dumps(task_info, indent=4)) except Exception as ex: msg = f"dump task_info to json error, task_info={task_info}, error:{ex}" BGLOG.exception(msg) print(json.dumps({'error': {'ex': f'{ex}', 'msg': msg}}, indent=4)) def version(args): print(VERSION_STR) def main(): create_background_task_dir() parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", help="Available commands") parser.add_argument("-c", "--config", type=str, help="config json file for run/check task, should contain plugin_name and plugin_args") parser.add_argument("-i", "--task_id", help="task_id to run or check from config file, if config file not given, create in default config file") run_task_parser = subparsers.add_parser('run') run_task_parser.add_argument("-n", "--plugin_name", help="plugin name to run, if config file is given, this arg will be ignored", type=str) run_task_parser.add_argument("plugin_args", help="plugin_args for run plugin, if config file is given, this arg will be ignored", nargs="*") run_task_parser.set_defaults(func=cli_run_task) check_task_parser = subparsers.add_parser('check') check_task_parser.set_defaults(func=cli_check_task) version_task_parser = subparsers.add_parser('version') version_task_parser.set_defaults(func=version) args = parser.parse_args() if getattr(args, 'func', None) is None: parser.print_help() print(f"Example:\n\tpython {__file__} -c 'path to config.json' -i 'task_id' run") print(f"\tpython {__file__} -c 'path to config.json' -i 'task_id' check") return 1 init_logger() args.func(args) if __name__ == "__main__": main()