#$@%#$%!

Ngiler SH3LL 360
: / opt / cloudlinux / venv / lib / python3.11 / site-packages /

Information Server
MySQL : OFF Perl : OFF CURL : ON WGET : OFF PKEXEC : OFF

Directive	Local Value
IP Address	162.0.209.101
System	Linux premium169.web-hosting.com 4.18.0-553.50.1.lve.el8.x86_64 #1 SMP Thu Apr 17 19:10:24 UTC 2025 x86_64
User	luxeelil
PHP Version	8.0.30
Software	LiteSpeed
Doc root	Writable

Edit File : lvectllib.py \| Size : 102.55 KB
# coding=utf-8 # Liblve functions lib # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import contextlib import copy import errno import json import math import mmap import os import pwd import re import subprocess import sys import syslog import warnings import xml.dom.minidom as xml from builtins import range from functools import partial from mmap import PAGESIZE from typing import Dict, List, Optional, Text, Tuple, TypedDict # NOQA import unshare import clcommon import cldetectlib import lveapi from clcommon.const import Feature from clcommon.cpapi import admins, get_main_username_by_uid, is_panel_feature_supported, reseller_users from clcommon.cpapi.cpapiexceptions import EncodingError from clcommon.lock import acquire_lock from clcontrollib import detect_panelclass from clevents import reseller_limits_disabled_post, reseller_limits_enabled_post from cllimits.lib import exec_utility from cllvectl.log import get_subprocess_logger from clveconfig.ve_config import BadVeConfigException, get_xml_config, save_xml from clveconfig.ve_lock import LockFailedException, setup_global_lock from lveapi import LVP_XML_TAG_NAME, Lve, NameMap, PyLve, PyLveError from secureio import create_dir_secure, write_file_via_tempfile GET_CP_PACKAGE_SCRIPT = '/usr/bin/getcontrolpaneluserspackages' CPUINFO = '/proc/cpuinfo' CORE_WEIGHT = 10000 DEFAULT_PACKAGE = "VE_DEFAULT" NOIOPS = False UMOUNT = '/bin/umount' EXCLUDE_MOUNTS_CONF = '/etc/container/exclude_mounts.conf' MULTI_FORMAT = 'multi' SINGLE_FORMAT = 'single' IS_DEBUG = int(os.environ.get('PYLVE_DEBUG', 0)) XML_PLESK_ID = 'plesk_id' # XML attribute name to bind package name in the ve.cfg with plesk DB id if not is_panel_feature_supported(Feature.LVE): pylve = None lve = None else: pylve = PyLve(debug=IS_DEBUG) lve = Lve(py=pylve) class LiblveSettings(TypedDict): ls_cpu: int \| str ls_cpus: int ls_io: int ls_enters: int ls_memory_phy: int ls_nproc: int ls_iops: int def create_liblve_settings(*kwargs) -> LiblveSettings: defaults: LiblveSettings = { 'ls_cpu': 0, 'ls_cpus': 0, 'ls_io': 0, 'ls_enters': 0, 'ls_memory_phy': 0, 'ls_nproc': 0, 'ls_iops': 0 } defaults.update(kwargs) return defaults def lvp_list(): """Helper function for easy mocking in unittests""" if lve.reseller_limit_supported(): return lve.proc.lvp_id_list() return [] def get_active_resellers(): """ Get list of resellers with activated reseller limits :return: list of pairs (name, uid) """ name_map = NameMap() name_map.link_xml_node() return name_map.load_from_node() def is_active_reseller_limits(reseller_name): """ Check whether giver reseller has activated reseller limits or not :return: bool """ return reseller_name in (name for name, uid in get_active_resellers()) # TODO: py3 move it to cllib/long script def raise_cpanel_encoding_error(e: EncodingError): """ Since cPanel user can corrupt config file for some user with wrong encodings, we want to notify him that he should fix encoding problems with the link to documentation. Print error message and exit with code 1 or raise given exception if it isn't cPanel. :return: None """ if not cldetectlib.is_cpanel(): raise e if JSON: json_format('multi', ['ERROR', str(e)]) else: print(e) sys.exit(1) def get_global_lock(write=False): """ ~~~~~~~~~~~~~~~~~~ !!! DEPRECATED !!! ~~~~~~~~~~~~~~~~~~ Please, use setup_global_lock instead if possible Wrapper over setup_global_lock. If lock cannot be set, it will write message and close app The only reason why it is here is legacy function check_result_and_exit that we use in TWO places :type write: bool :return: Nothing """ try: setup_global_lock(write) except LockFailedException: check_result_and_exit(1, 'can`t get lock') def check_result_and_exit(result, message): # on cl5 some func is unimplemented; so ENOSYS is not error; if result not in (0, -errno.ENOSYS): if JSON: json_format(MULTI_FORMAT, ['ERROR', f'lvectl: {message}']) else: print(f'lvectl: Error: {message}') sys.exit(result) # Default parameters for lve LVE_DEFAULT = { 'cpu': 25, 'ncpu': 1, 'io': 25, 'ep': 20, 'mem': 0, 'pmem': 262144, 'nproc': 0, 'iops': 1024 } MEM_DEFAULT_CL5 = 262144 # Default parameters for lvp LVP_DEFAULT = { 'cpu': 100, 'ncpu': 1, 'io': 0, 'ep': 0, 'mem': 0, 'pmem': 0, 'nproc': 0, 'iops': 0 } LIMITS_LIST_NAME = ['ncpu', 'cpu', 'io', 'mem', 'pmem', 'nproc', 'iops', 'ep'] LVE_VERSION = 4 JSON = False BYTES_FLAG = False # defined structures for liblve and turples for functions lve_settings = '' setup_data = '' # type: dict # dict with user-packages relations # keys = int UID or str package name # data = string package name packages_users = {} # defined ve.cfg variables ve_cfg = '' ve_lveconfig = '' ve_default = '' ve_lve = '' ve_lvp = '' # for resellers limits ve_defaults = '' # type: dict ve_package = '' ubc = 'false' # TODO: looks like not used anymore, check and remove it ve_enter_by_name = '' ve_binary = '' ve_cfg_version = '' # Set JSON if json output required def set_json(json_flag): global JSON JSON = json_flag def set_bytes(bytes_flag): global BYTES_FLAG BYTES_FLAG = bytes_flag def get_fields(): if NOIOPS and LVE_VERSION == 8: version = 'noiops_8' elif LVE_VERSION == 8: version = '8' elif LVE_VERSION == 6: version = '6' else: # LVE_VERSION == 4 version = '4' fields = { 'noiops_8': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO'], '8': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO','IOPS'], '6': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO'], '4': ['ID','SPEED','VMEM','EP','IO'] }[version] if JSON: speed_idx = fields.index('SPEED') + 1 return (fields[:speed_idx] + ["CPU"] + fields[speed_idx:]) return fields # Create structure def init(lve_ver=None): global LVE_VERSION if lve_ver is None: lve_ver = clcommon.get_lve_version() if lve_ver[0] is None: raise RuntimeError('get_lve_version failed') LVE_VERSION = lve_ver[0] else: LVE_VERSION = lve_ver global lve_settings lve_status = pylve.initialize() if not lve_status: raise RuntimeError('init_lve() failed.') lve_settings = pylve.liblve_settings() # we use /proc/cpuinfo to get cpu speed, but unfortunately # it returns current CPU MHZ, which is different in lve environment # and we cannot get right speed value there # FIXME: LU-947 def _get_cpu_data_from_env(): """Get cpu information from environment veriable""" packed_cpu_data = os.environ.get('CPU_DATA') if packed_cpu_data is None: return None try: return json.loads(packed_cpu_data) except (TypeError, ValueError) as e: print('Invalid environment variable \'CPU_DATA\' format', str(e)) sys.exit(1) def get_cpu_data(): """ Parse /proc/cpuinfo return [NumProc, frequency in MHZ] """ cpuinfo = {} procinfo = {} nprocs = 0 try: # f = open(CPUINFO, 'r') with open(CPUINFO, 'r', encoding='utf-8') as f: for line in f: if not line.strip(): # end of one processor cpuinfo[f'proc{nprocs}'] = procinfo nprocs = nprocs + 1 # Reset procinfo = {} else: if len(line.split(':')) == 2: procinfo[line.split(':')[0].strip()] = line.split(':')[1].strip() else: procinfo[line.split(':')[0].strip()] = '' except IOError: print(f'lvectl: Error: Can`t open {CPUINFO}.') sys.exit(1) return [nprocs, cpuinfo['proc0']['cpu MHz']] # It's extremely rare case when CPU changes at runtime so we will use cached CPUINFO_DATA = _get_cpu_data_from_env() or get_cpu_data() def convert_from_old_cpu(data, lncpu=0): """ Try converting to kernel format from old CPU format (percentage of whole cpu) and optionally the NCPU format. Return whichever is less. :param data: string presumably in old CPU format :param lncpu: integer number of cores limit """ data = str(data) lncpu = lncpu or 0 cpu_data = CPUINFO_DATA ncpu = int(cpu_data[0]) cpu_percent = re.match(r'\d{1,2}0?$', data) # 0-100 if cpu_percent is not None: data = int(data) if 0 < data <= 100: from_cpu_limit = int(round(CORE_WEIGHT // 100 ncpu * data)) if lncpu == 0: return from_cpu_limit return min(lncpu * CORE_WEIGHT, from_cpu_limit) return None def convert_from_speed_percent(data): """ Try converting cpu limit from SPEED in percentage of one CORE format to kernel format. """ data = str(data) cpu_data = CPUINFO_DATA ncpu = int(cpu_data[0]) percent = re.match(r'\d+(?:\.\d+)?%$', data) # % if percent is not None: percent = float(data.replace('%', '')) if percent > ncpu 100: percent = ncpu * 100 if percent > 0: return int(round(CORE_WEIGHT // 100 * percent)) return None return None def convert_from_speed_hz(data): """ Try converting cpu limit from SPEED in mhz/gzh format to kernel format. """ data = str(data) cpu_data = CPUINFO_DATA ncpu = int(cpu_data[0]) cpu_freq = float(cpu_data[1]) pattern = re.compile(r'(?P<freq>\d+(?:\.\d+)?)(?P<suffix>mhz\|ghz)+$', re.IGNORECASE) match = pattern.match(data) # mhz\ghz if match is not None: suffix = match.group('suffix') freq = float(match.group('freq')) if suffix.upper() == 'GHZ': freq = freq 1000 if freq > cpu_freq * ncpu: freq = cpu_freq * ncpu if freq > 0: return int(round(freq * CORE_WEIGHT / cpu_freq)) return None def convert_from_speed(data): """ Try converting cpu limit value from either SPEED limit format (percentage of CORE or mhz/ghz) to kernel format. """ return ( convert_from_speed_percent(data) or convert_from_speed_hz(data) ) def convert_to_kernel_format(data, lncpu=0): """ Convert different variants of cpu limit to kmod ver 8 variant :param data: Value in old CPU format or SPEED with % or mhz/ghz. :param lncpu: Limit in old NCPU format. :return: CPU limit in kmod ver 8+ format or None for bad format """ from_cpu = convert_from_old_cpu(data, lncpu) if from_cpu is not None: return from_cpu from_speed_percent = convert_from_speed_percent(data) if from_speed_percent is not None: return from_speed_percent from_speed_hz = convert_from_speed_hz(data) if from_speed_hz is not None: return from_speed_hz return None def speed_to_old_cpu(speed): """ convert speed to old cpu format args: cpu limit in speed value return: old cpu limit format """ cpu_data = CPUINFO_DATA nproc = int(cpu_data[0]) speed = str(speed) if '' in speed: return '' + str(int(round(int(speed.lstrip('')) // nproc))) return str(int(round(int(speed) // nproc))) @contextlib.contextmanager def temporary_lve(settings): # type: (pylve.liblve_settings) -> contextlib.GeneratorContextManager """ Run subprocess in lve with pseudo-random id and given limits """ pylve.initialize() lve_id = pylve.get_available_lve_id() try: pylve.lve_setup(lve_id, settings) except PyLveError: syslog.syslog(syslog.LOG_ALERT, f"Unable to setup lve with id {lve_id}, " "something is wrong, check dmesg for details") raise try: yield lve_id finally: pylve.lve_destroy(lve_id) def make_liblve_settings(settings: LiblveSettings): # type: (LiblveSettings) -> pylve.liblve_settings """ Just a nice user-friendly constructor of liblve_settings object You can pass the following ls_cpu and ls_cpus values: - in percents of one core (just ls_cpu='75%', ls_cpus will be ignored) - in old 'CPU' format (two arguments, ls_cpu and ls_cpus required, both int) """ s = pylve.liblve_settings() s.ls_cpu = convert_to_kernel_format(settings['ls_cpu'], lncpu=settings['ls_cpus']) s.ls_io = settings['ls_io'] s.ls_enters = settings['ls_enters'] s.ls_nproc = settings['ls_nproc'] s.ls_iops = settings['ls_iops'] # convert memory from bytes to mempages s.ls_memory_phy = int(math.ceil(1. settings['ls_memory_phy'] / PAGESIZE)) return s def get_ve_lve_user_uid(ve_lve_element): user_uid = str(ve_lve_element.getAttribute('id')) if not user_uid: user_name = ve_lve_element.getAttribute('user') user_uid = pwd.getpwnam(user_name).pw_uid return int(user_uid) def json_format(error_type, data, extensions=None): """ Print output in json as: {"status": "ERROR/OK", "msg": "Some Message", "ext1": "foo", "ext2": "bar"} where "status" and "msg" field are mandatory :param str error_type: Either MULTI_ERROR or SINGLE_ERROR :param list data: List with a status string and a message string :param dict extensions: Some additional fields for the final json object :return: None """ result = {'status': str(data[0])} if error_type == MULTI_FORMAT: result['msg'] = str(data[1]) if extensions is not None: result.update(extensions) print(json.dumps(result)) def check_def_value(xml, ve_defaults, ve_cfg, val, default): try: ve_defaults[val] = int(ve_default.getElementsByTagName(val)[0].getAttribute('limit')) except (ValueError, IndexError, TypeError): ve_defaults[val] = default[val] node = ve_cfg.createElement(val) node.setAttribute('limit',str(default[val])) try: xml.appendChild(node) except Exception: pass def xml_filter_tag(node, tag): return [_ for _ in node.childNodes if isinstance(_, xml.Element) and _.tagName == tag] def xml_filter_first(node, tag, attr=None, attr_val=None): for child_node in xml_filter_tag(node, tag): if attr is not None and not child_node.hasAttribute(attr): continue if attr_val is not None and child_node.getAttribute(attr) != attr_val: continue return child_node def get_child_tag_atrr(node, tag, attr): filtered_child_node = xml_filter_first(node=node, tag=tag, attr=attr) if filtered_child_node is None: raise IndexError() return filtered_child_node.getAttribute(attr) def set_child_tag_atrr(node, tag, attr, val): """ Find in children nodes node with tag and setup attribute insted el.getElementsByTagName not search recursiveli in tree """ first_child_node = xml_filter_tag(node, tag)[0] first_child_node.setAttribute(attr, str(val)) def _load_config_wrapper(): """Load config from ve.cfg""" global ve_cfg global ve_lveconfig try: ve_cfg, ve_lveconfig = get_xml_config() except BadVeConfigException as e: if JSON: json_format(MULTI_FORMAT, ['ERROR', str(e)]) else: print(str(e)) sys.exit(1) def _load_default_limits(lvp_id: int, lvp_defaults: bool): """Load default limits :param int lvp_id: lvp id :param bool lvp_defaults: load reseller's default limits instead of global :return: dict with default limits """ global ve_default global ubc ubc = 'true' try: if not lvp_id: ve_default = xml_filter_tag(ve_lveconfig, 'defaults')[0] return LVE_DEFAULT if lvp_defaults: ve_default = ve_cfg.createElement('defaults') return LVP_DEFAULT defaults_root_node = xml_filter_first(ve_lveconfig, LVP_XML_TAG_NAME, 'id', str(lvp_id)) if defaults_root_node: # if no such reseller with lvp in config ve_default = defaults_root_node.getElementsByTagName('defaults')[0] else: ve_default = ve_default.cloneNode(ve_default) return LVE_DEFAULT except IndexError: if JSON: json_format('multi', ['WARNING', 'default section error in ve.cfg']) sys.exit(1) else: print('warning: default section error in ve.cfg') return LVE_DEFAULT def _all_config_elements_loaded(): return all(x != '' for x in (ve_lve, ve_lvp, ve_package, ve_binary, ve_enter_by_name, ve_cfg_version)) def _load_config_elements(): """Load all config elements from ve.cfg""" global ve_lve global ve_lvp global ve_package global ve_binary global ve_enter_by_name global ve_cfg_version ve_lve = ve_lveconfig.getElementsByTagName("lve") ve_package = ve_lveconfig.getElementsByTagName("package") ve_lvp = ve_lveconfig.getElementsByTagName(LVP_XML_TAG_NAME) lve.map.name_map.link_xml_node(ve_lveconfig) enter_by_name_elems = ve_lveconfig.getElementsByTagName('enter-by-name') if len(enter_by_name_elems) > 0: ve_enter_by_name = enter_by_name_elems[0] else: ve_enter_by_name = ve_cfg.createElement('enter-by-name') ve_lveconfig.appendChild(ve_enter_by_name) ve_binary = ve_enter_by_name.getElementsByTagName('binary') # expected version tag in next format # # <lveconfig> # <version>2</version> # <system> # .... # </system> # .... # </lveconfig> cfg_version_elems = ve_lveconfig.getElementsByTagName('version') ve_cfg_version = int(cfg_version_elems[0].firstChild.nodeValue) if len(cfg_version_elems) > 0 else 1 def _load_ve_defaults(default_limits: Dict[str, int]): """Create ve_defaults dict with default values for all limits :param dict default_limits: default limits for lve or lvp """ global ve_cfg global ve_defaults global ve_default ve_defaults = {} check_def_val = partial( check_def_value, xml=ve_default, ve_defaults=ve_defaults, ve_cfg=ve_cfg, default=default_limits, ) check_def_val(val='ncpu') try: speed = ve_default.getElementsByTagName('cpu')[0].getAttribute('limit') ve_defaults['cpu'] = convert_to_kernel_format(speed, lncpu=ve_defaults['ncpu']) except (ValueError, IndexError, TypeError): ve_defaults['cpu'] = convert_to_kernel_format(default_limits['cpu'], lncpu=ve_defaults['ncpu']) cpu = ve_cfg.createElement('cpu') cpu.setAttribute('limit', str(default_limits['cpu'])) try: ve_default.appendChild(cpu) except Exception: pass try: ve_defaults['ep'] = int(ve_default.getElementsByTagName('other')[0].getAttribute('maxentryprocs')) except (ValueError, IndexError, TypeError): ve_defaults['ep'] = default_limits['ep'] ep = ve_cfg.createElement('other') ep.setAttribute('maxentryprocs', str(default_limits['ep'])) try: ve_default.appendChild(ep) except Exception: pass check_def_val(val='io') if LVE_VERSION > 5: check_def_val(val='mem') else: check_def_val(val='mem', default={'mem': MEM_DEFAULT_CL5}) # pylint: disable=redundant-keyword-arg check_def_val(val='pmem') check_def_val(val='nproc') check_def_val(val='iops') _check_defaults_for_nones() def _check_defaults_for_nones(): """Check that all default values are not None""" global ve_defaults for key, value in ve_defaults.items(): if value is not None: continue err_msg = f'ERROR: Incorrect {key} default value' if JSON: json_format('multi', ['ERROR', err_msg]) else: sys.stderr.write(f'{err_msg}\n') sys.exit(1) def get_XML_cfg(lvp_id=0, lvp_defaults=False, load_config_elements=True): """ :param bool lvp_defaults: load reseller's default limits instead of global :param int lvp_id: lvp id to load customise defaults """ _load_config_wrapper() default_limits = _load_default_limits(lvp_id, lvp_defaults) # load config elements if load_config_elements=True or they were not loaded before if load_config_elements or not _all_config_elements_loaded(): _load_config_elements() _load_ve_defaults(default_limits) def check_value(val, el, ve_defaults, setup_data): try: value = int(get_child_tag_atrr(el, tag=val, attr='limit')) setup_data[val] = value return value except (ValueError, IndexError, TypeError): return int(ve_defaults[val]) def _load_resellers_xml_data(reseller, xml_config_load_elements=True): """ This function is a pure workaround for our ugly globals-based API which should be fixed partially with LU-496, because there is no clean way to retrieve reseller's data from ve.cfg without touching globals :param reseller: reseller name :return: Nothing. It just updates some globals """ # TODO after LU-496 we should read and cache all reseller's settings name_map = lveapi.NameMap() name_map.link_xml_node() reseller_id = name_map.get_id(reseller) get_XML_cfg(reseller_id, load_config_elements=xml_config_load_elements) def prepare_setup_data(plan_id=None, reseller=None, lve_id=None): # type: (Optional[Text], Optional[Text]) -> None """ Put limit values that will be applied later in a global variable `setup_ve`. :param plan_id: package :param reseller: If reseller is None we only inherit from admin packages. In that case we ignore all tags in ve.cfg with a "reseller" attribute. """ global setup_data setup_data = copy.copy(ve_defaults) if plan_id is not None: res_pkg_dict = get_reseller_packages_map() # LU-510: Investigate the problem with reseller's list, part 2. # Fix applying limit for reseller user with admin package if reseller is not None and reseller in res_pkg_dict and plan_id in res_pkg_dict[reseller]: def is_needed_plan(el): return el.getAttribute('id') == plan_id and el.getAttribute('reseller') == reseller else: if cldetectlib.is_da() and lve_id is not None: try: user_pwd = pwd.getpwuid(lve_id) filename = f'/usr/local/directadmin/data/users/{user_pwd.pw_name}/user.conf' with open(filename, encoding='utf-8') as f: text = f.read() except Exception: text = '' # LU-1663 --> LU-3410: # it is normal situation for DA to set DA(not LVE) user package to "custom" # in this case the package is still correct and this warning should not be logged if 'package=custom' not in text: syslog.syslog( syslog.LOG_ALERT, f"Package for user with id {lve_id} is incorrect, please recover it using Note from " "https://docs.cloudlinux.com/cloudlinux_os_components/#installation-enabling-and-disabling", ) # Ignore all tags in ve.cfg with a "reseller" attribute. def is_needed_plan(el): return el.getAttribute('id') == plan_id and not el.getAttribute('reseller') # Example command when `ve_package` is not empty: # cloudlinux-packages set --json --for-reseller root --package kekage --pmem 994 --nproc 77 # `ve_package` is set by reading ve.cfg in `get_XML_cfg`. # <package> tag is added to ve.cfg in `package_set_ext`. for el in ve_package: if is_needed_plan(el): lncpu = check_value('ncpu', el, ve_defaults, setup_data) try: cpu = int(convert_to_kernel_format(get_child_tag_atrr(el, tag='cpu', attr='limit'), lncpu=lncpu)) setup_data['cpu'] = cpu except (ValueError, IndexError, TypeError): pass check_value('io', el, ve_defaults, setup_data) if (ubc == 'true'): check_value('mem', el, ve_defaults, setup_data) else: setup_data['mem'] = 0 try: ep = int(get_child_tag_atrr(el, tag='other', attr='maxentryprocs')) setup_data['ep'] = ep except (ValueError, IndexError, TypeError): pass check_value('nproc', el, ve_defaults, setup_data) check_value('pmem', el, ve_defaults, setup_data) check_value('iops', el, ve_defaults, setup_data) def umount_dir(path): try: # run the "umount" command and suppress it's output, return True when child exit code is not zero with subprocess.Popen( [UMOUNT, "-l", path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) as proc: proc.communicate() return proc.returncode != 0 except OSError: check_result_and_exit(-1, f'failed to run "{UMOUNT} -l {path}"') def prepare_mounts(): """ Unmount all paths from /proc/mounts that match regular expressions from /etc/container/exclude_mounts.conf file """ if not os.path.isfile(EXCLUDE_MOUNTS_CONF): return reg_exp_list = [] try: with open(EXCLUDE_MOUNTS_CONF, 'r', encoding='utf-8') as conf: for r in conf: pattern = r.strip() if pattern: reg_exp_list.append(re.compile(pattern)) except IOError: check_result_and_exit(-1, f'failed to read {EXCLUDE_MOUNTS_CONF}') if not reg_exp_list: return unshare.unshare(unshare.CLONE_NEWNS) try: with open('/proc/mounts', 'r', encoding='utf-8') as f: mounts = [m.split()[1] for m in f.readlines()] except (IndexError, IOError): check_result_and_exit(-1, 'failed to parse /proc/mounts') ATTEMPTS = 10 for _ in range(ATTEMPTS): error = False for mount in mounts: for reg_exp in reg_exp_list: m = reg_exp.search(mount) if m: error = umount_dir(mount) or error break if not error: break def lve_start(): """ Start LVE engine and initialize default mount namespace for LVE """ MOUNT_CMD = '/bin/mount --make-rprivate / >/dev/null 2>&1' try: subprocess.call(MOUNT_CMD, shell=True, executable='/bin/bash') except OSError: print('Error: failed to execute', MOUNT_CMD) prepare_mounts() pylve.lve_start(err_msg='Can`t init lve default settings') def lve_create(lve_id, ignore_error=False): """ Create LVE container for given ID :type lve_id: int :type ignore_error: bool :return: Nothing """ pylve.lve_create(lve_id, err_msg=f'lvectl: Can`t create lve with id {lve_id}; error code {{code}}', ignore_error=ignore_error) def lvp_create(lvp_id, ignore_error=False): """ Create LVP container for given ID :type lvp_id: int :type ignore_error: bool :return: Nothing """ pylve.lve_lvp_create(lvp_id, err_msg=f'lvectl: Can`t create lvp with id {lvp_id}; error code {{code}}', ignore_error=ignore_error) def destroy_lvp_all(): logger = get_subprocess_logger('lvectllib') destroyed_list = [] for lvp_id in lvp_list(): logger.debug('destroy_lvp_all: destroying LVP with id %s', lvp_id) pylve.lve_lvp_destroy(lvp_id, err_msg=f'lvectl: Can`t destroy lvp with id {lvp_id}; error code {{code}}') destroyed_list.append(lvp_id) return destroyed_list def lvp_destroy(lvp_id): logger = get_subprocess_logger('lvectllib') if lvp_id == 'all': destroy_lvp_all() return logger.debug('lvp_destroy: destroying LVP with id %s', lvp_id) pylve.lve_lvp_destroy(lvp_id, err_msg=f'lvectl: Can`t destroy lvp with id {lvp_id}; error code {{code}}') # Destroy LVE container for ID def lve_destroy(lve_id): logger = get_subprocess_logger('lvectllib') cant_remove_msg = f'Can\'t remove lve {lve_id} from kernel - error code -3' cant_destroy_msg = f'Can`t destroy lve with id {lve_id}; error code {{code}}' destroyed = False if lve_id == 'all': if lve.reseller_limit_supported(): destroyed = bool(destroy_lvp_all()) # destroy all top level containers if len(list(lve.proc.lve_id_list())) > 0: for id_ in lve.proc.lve_id_list(): logger.debug('lve_destroy all: destroying LVE with id %s', id_) lve.lve_destroy(id_, err_msg=cant_destroy_msg) destroyed = True else: destroyed = True # empty list - all lve already destroyed else: if lve.proc.check_inside_list(lve_id) or \ (lve.proc.resellers_supported() and lve.proc.detect_inside_lvp(lve_id) is not None): logger.debug('lve_destroy: destroying LVE with id %s', lve_id) lve.lve_destroy(lve_id, err_msg=cant_destroy_msg) destroyed = True elif not lve.proc.check_inside_list(lve_id): destroyed = True # lve_id doesn`t exist so it`s already destroy! if destroyed: if JSON: json_format(SINGLE_FORMAT, ['OK']) else: if JSON: json_format(MULTI_FORMAT, ['WARN', cant_remove_msg]) else: print(f'warning: {cant_remove_msg}') # Setup LVE for ID def lve_setup(lve_id, lvp_id=0): if lvp_id and not lve.proc.exist_lvp(lvp_id): # create lve top container if not exist lvp_create(lvp_id) with warnings.catch_warnings(): # convert all warning to exceptions warnings.filterwarnings('error') try: lve_settings.ls_io = int(setup_data['io']) lve_settings.ls_cpu = int(setup_data['cpu']) lve_settings.ls_cpus = int(setup_data['ncpu']) lve_settings.ls_memory = int(setup_data['mem']) lve_settings.ls_enters = int(setup_data['ep']) if LVE_VERSION > 5: lve_settings.ls_memory_phy = int(setup_data['pmem']) lve_settings.ls_nproc = int(setup_data['nproc']) if LVE_VERSION > 6: lve_settings.ls_iops = int(setup_data['iops']) if lvp_id: if lve_id == 0: pylve.lve_set_default( lvp_id, lve_settings, err_msg=f'Can`t setup default settings for LVP {lvp_id}') else: pylve.lve_lvp_setup( lvp_id, lve_settings, err_msg=f'Can`t setup lvp with id {lvp_id}; error code {{code}}') elif lve_id == 0: pylve.lve_set_default(lve_settings, err_msg='Can`t setup default settings') else: pylve.lve_setup( lve_id, lve_settings, err_msg=f'Can`t setup lve with id {lve_id}; error code {{code}}') except RuntimeWarning as rw: # exit if caught a warning - can`t set limits to lve check_result_and_exit(1,'Can`t setup lve ' + str(lve_id) + '. RuntimeWarning excepted: ' + str(rw)) def get_package_and_reseller_by_lve_id(lve_id): """ Get pair of package, reseller for lve_id :param lve_id: lve_id, UID with package, reseller :return: tuple of (package, reseller); Both can be None """ global packages_users package = None reseller = None # hate this! backup global ref to packages_users old_packages_users = packages_users GetControlPanelUsers('list-users') # now packages_users should be dict of dicts: # {lve_id : {'package' : , 'reseller':}} temp_package = packages_users # restore global ref to packages_users packages_users = old_packages_users if isinstance(temp_package, dict): try: reseller = temp_package[lve_id]['reseller'] # if reseller is empty then reseller is root/admin. return to None if not reseller: reseller = None except KeyError: reseller = None try: package = temp_package[lve_id]['package'] # if package is empty - return to undefined state if not package: package = None except KeyError: package = None return package, reseller # (rprilipskii): Honestly, relying on global variables for storage like this leaves a bad # taste in my mouth, but I'd have to rewrite a more lvectl code than I'd prefer if I # wanted to do it in a more pythonic way. Technical debt is a problem. CACHED_EFFECTIVE_LIMITS = {} EFFECTIVE_CACHE_FILE = "/var/run/cloudlinux/effective-normal-limits" BURSTABLE_LIMITS_FLAG_FILE = "/opt/cloudlinux/flags/enabled-flags.d/burstable-limits.flag" def write_effective_cache(reset=False): """ Save calculated effective normal limits for an LVE to a cache file. The cache file is stored in /var/run (tmpfs) for faster operations. It disappears on reboot, but lvectl apply all runs in the lvectl systemd service anyway, so the file will always reappear if the service functions normally. If additional speed is desired, consider replacing the standard JSON lib with a faster one like rapidjson or orjson. :param reset: If True, recreate the cache file from scratch with data available in the dict, instead of updating it, defaults to False :type reset: bool, optional """ # NOTE: When running `lvectl apply all`, this write will happen at the very end # of the operation. Applying the limits, however, happens before this - right after # they're calculated. # This means a substantial time gap and a potential race condition if someone calls # `lvectl set` while `apply all` is not yet finished (e.g. cache and kernel having different limits). # Could be solved by makin limit application happen after all effective limits are # done calculating, and wrapping limit application and cache writing in the same filelock. # Do nothing if the corresponding feature flag is not set if not os.path.exists(BURSTABLE_LIMITS_FLAG_FILE): return # If there's no /var/run/cloudlinux directory, create it. try: effective_dir = os.path.dirname(EFFECTIVE_CACHE_FILE) if not os.path.isdir(effective_dir): var_run_dir = os.path.dirname(effective_dir) create_dir_secure(effective_dir, 0o600, 0, 0, var_run_dir) except OSError as e: print(f"Error: failed to create the folder {effective_dir}: {e}") raise # Load already cached effective limits and merge them with # the limits calculated during the current lvectl run. # Unless we want to discard them and start over, that is. effective_cache_lock = acquire_lock(f"{EFFECTIVE_CACHE_FILE}.lock") with effective_cache_lock: if reset: effective_limits = {} else: try: if os.path.isfile(EFFECTIVE_CACHE_FILE): with open(EFFECTIVE_CACHE_FILE, "r", encoding="utf8") as readfile: effective_limits = json.load(readfile) else: effective_limits = {} except json.JSONDecodeError as e: print(f"Error: failed to parse the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}") raise except OSError as e: print(f"Error: failed to read the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}") raise effective_limits.update(CACHED_EFFECTIVE_LIMITS) try: write_file_via_tempfile(json.dumps(effective_limits), EFFECTIVE_CACHE_FILE, 0o600) except OSError as e: print(f"Error: failed to save the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}") def cache_effective_limits(lve_id): """ Cache the calculated effective normal limits for an LVE to a dictionary. setup_data is a global dict that contains the LVE configuration last applied. Holds only one entry - during operations like lvectl apply all, we enter this function repeatedly with setup_data containing info for one processed LVE each time. These limits can be used by the Burstable Limits components to set/reset burst or normal limits without having to go through effective limit calculation or invoking lvectl as a middleman. Comparing the cached limits to current active limits (in kernel) also shows whether or not the LVE has burst limits active at the moment. This function gets called in lve_apply, which means it also runs within: * lvectl apply all * lvectl apply-many * lvectl set * and others, that also call lve_apply :param lve_id: LVE ID for which the limits are being saved. :type lve_id: int """ global CACHED_EFFECTIVE_LIMITS CACHED_EFFECTIVE_LIMITS[str(lve_id)] = setup_data # pylint: disable-msg=too-many-arguments # Apply user settings def lve_apply(lve_id, plan_id=None, result=False, reseller=None, out_node=None, lvp_id=0): """ Aplly limits to LVE lve_id :param lve_id: lve id :type lve_id: int :param plan_id: package for user with lve_id. deprecated :type plan_id: string :param result: if True = don't apply limits. only create setup_data with actual limits :type result: boolean :param reseller: if True = plan_id is resellers plan. deprecated :type reseller: boolean :param out_node: node with limits for lve_id :type out_node: xml_node :param lvp_id: reseller container id; host container if 0 """ global setup_data global packages_users old_packages_users = packages_users if lve_id != 0: GetControlPanelUsers('userid', lve_id) new_packages_users, packages_users = packages_users, old_packages_users if ve_cfg == '': get_XML_cfg(lvp_id=lvp_id) el = None if out_node is not None: el = out_node else: node_list = ve_lvp if lvp_id else ve_lve el = next( filter(lambda node: get_ve_lve_user_uid(ve_lve_element=node) == lve_id, node_list), None ) # reseller (lvp_id!=0) should not use package and default limits if lvp_id == 0: try: plan_id = new_packages_users[lve_id]['package'] except (NameError, KeyError): plan_id = None try: reseller = new_packages_users[lve_id]['reseller'] except (NameError, KeyError): reseller = None if el is not None: # get limits from package prepare_setup_data(plan_id, reseller=reseller) # prepare custom limits for lve lncpu = check_value('ncpu', el, ve_defaults, setup_data) try: setup_data['cpu'] = convert_to_kernel_format(get_child_tag_atrr(el, tag='cpu', attr='limit'), lncpu=lncpu) except (ValueError, IndexError, TypeError): pass if setup_data['cpu'] is None: setup_data['cpu'] = ve_defaults['cpu'] check_value('io', el, ve_defaults, setup_data) if ubc == 'true': check_value('mem', el, ve_defaults, setup_data) else: setup_data['mem'] = 0 try: setup_data['ep'] = int(get_child_tag_atrr(el, tag='other', attr='maxentryprocs')) except (ValueError, IndexError, TypeError): pass check_value('nproc', el, ve_defaults, setup_data) check_value('pmem', el, ve_defaults, setup_data) check_value('iops', el, ve_defaults, setup_data) else: # apply default limits prepare_setup_data(plan_id, reseller=reseller, lve_id=lve_id) if ubc == 'false': setup_data['mem'] = 0 if not result: cache_effective_limits(lve_id) # apply limits lve_setup(lve_id, lvp_id=lvp_id) def _pprint(fields): """ Print data with the last column 30 symbols wide. Useful for printing data that contains package names. """ formatted_string = _format_fields(fields, wide_indices=[len(fields) - 1]) print(formatted_string) def _pprint_f(fields): """ Print data with the two last columns 30 symbols wide. Useful for printing full data of every user with package name and reseller name. """ formatted_string = _format_fields( fields, wide_indices=[len(fields) - 2, len(fields) - 1], ) print(formatted_string) def _pprint_p(fields): """ Print data with the first column 30 symbols wide. Useful for printing packages data. """ formatted_string = _format_fields(fields, wide_indices=[0]) print(formatted_string) def _pprint_r(fields): """ Print data with the first and last columns 30 symbols wide. Useful for printing data with user names and package names. """ formatted_string = _format_fields(fields, wide_indices=[0, len(fields) - 1]) print(formatted_string) def _format_fields(fields: tuple, wide_indices: list, width: int = 30) -> str: """ Helper function to format fields based on specified indices for wide columns. Args: fields: The fields to format. wide_indices: List of indices in the fields that should be wide. width: The width of the wide columns. Returns: A formatted string with specified fields widened. """ formatted_fields = [] for index, field in enumerate(fields): # Convert field to string to ensure compatibility with formatting field_str = str(field) if index in wide_indices: formatted_fields.append(f"{field_str:>{width}}") else: formatted_fields.append(f"{field_str:>8}") return ''.join(formatted_fields) def _pmem_vmem_to_bytes_value(value): """ Convert pmem or vmem limits to bytes value :param value: pmem or vmem limits in kbytes value :return: bytes value of limit """ # if value was changed we remove asterisk from value for counted this value = str(value) was_changed = isinstance(value, str) and value.startswith('') value = value.replace('', '') if value: value = int(value) value = 4096 else: value = 0 # return asterisk in value if this was changed value = f'{value}' if was_changed else value return value def _mb_mem(value): """ Convert amount of RAM to M format :param string value: amount of memory in KB :rtype: string :return: amount of memory in MB like "1234M" """ result = '' was_changed = False if isinstance(value, str) and value.startswith(''): value = value[1:] was_changed = True try: v = int(value) except ValueError: return "" if was_changed: result = '' value = v * 4 // 1024 if value > 0: result = f'{result}{value}M' else: result = f'{result}{v * 4}K' return result def _formatter(printer, default_id="0", default_package="default", more_fields=None): """ Generate header and default package data either as print to stdout or as json string """ defaults = ve_defaults.copy() # convert from kernel format for output defaults['cpu'] = defaults['cpu'] // 100 def get_data(key): return defaults.get(key, '') _cpu = speed_to_old_cpu(get_data("cpu")) if get_data("cpu") != '' else '' def convert_mem_limits(value): return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value) fields_map = { 'ID': default_id, 'SPEED': str(get_data('cpu')), 'CPU': str(_cpu), 'NCPU': str(get_data('ncpu')), 'PMEM': str(convert_mem_limits(get_data('pmem'))), 'VMEM': str(convert_mem_limits(get_data('mem'))), 'EP': str(get_data('ep')), 'NPROC': str(get_data('nproc')), 'IO': str(get_data('io')), 'IOPS': str(get_data('iops')), 'PACKAGE': default_package } res = [] fields = get_fields() if more_fields is not None: fields += more_fields if JSON: line = ','.join(f'"{f}":"{fields_map.get(f, "")}"' for f in fields) res = [f'{{{line}}}'] else: printer(fields) printer([fields_map.get(f, "") for f in fields]) return res def _user_formatter(fields, printer=_pprint): """ Generate inner function with closured fields names and printer function :param list fields: List of strings that represent names of fields in final output :param callable printer: Function to format and print data for every entry :rtype: callable :return: function to format data for every user """ def wrapper(user): """ :param string user: Find and format data for this User ID :rtype: list :return: List of given user's statistics data line or empty list """ data = '' package = packages_users[user]["package"] reseller = packages_users[user]["reseller"] if reseller == '': reseller = None if ve_cfg_version <= 1: # Reseller's default limits will not be inherited by its end-users. # Backward compatibility - show some reseller packages in # paneluserslimits. prepare_setup_data(package, reseller=None) else: if reseller is not None: # We can set xml_config_load_elements=False # because get_XML_cfg was called with True before _user_formatter called _load_resellers_xml_data(reseller, xml_config_load_elements=False) else: # It's important to re-read admin's limits here or we will use # limits from previous reseller for next users in list # We can set xml_config_load_elements=False # because get_XML_cfg was called with True before _user_formatter called get_XML_cfg(load_config_elements=False) prepare_setup_data(package, reseller=reseller) data = copy.copy(setup_data) # only after reading reseller's xml lve_apply(user, plan_id=package, reseller=reseller, result=True) def check_changed(key): return '' + str(setup_data[key]) if str(data[key]) != str(setup_data[key]) else str(data[key]) def convert_mem_limits(value): return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value) data['id'] = str(user) data['cpu'] = str(check_changed('cpu')) data['ncpu'] = str(check_changed('ncpu')) data['pmem'] = str(convert_mem_limits(check_changed('pmem'))) data['vmem'] = str(convert_mem_limits(check_changed('mem'))) data['ep'] = str(check_changed('ep')) data['io'] = str(check_changed('io')) data['nproc'] = str(check_changed('nproc')) data['iops'] = str(check_changed('iops')) if JSON: data['package'] = _normalize_str(package) else: data['package'] = package if reseller is None: data['reseller'] = 'N/A' if JSON else '' else: data['reseller'] = reseller if '' in data['cpu']: data['cpu'] = '' + str(int(data['cpu'].lstrip('')) // 100) else: data['cpu'] = int(data['cpu']) // 100 data['speed'] = data['cpu'] data['cpu'] = str(speed_to_old_cpu(data['speed'])) res = [] if JSON: line = ','.join(f'"{f}":"{data[f.lower()]}"' for f in fields) res = [f'{{{line}}}'] else: printer([data[f.lower()] for f in fields]) return res return wrapper # Show current user's limits for control panel # 'lvectl paneluserslimits' or 'lvectl paneluserlimits lve_id' def paneluserslimits(userid=None, reseller=None): get_XML_cfg() try: # create cache for userid_calls GetControlPanelUsers('list-users') # use explicit compare, because userid may be zero! # if userid == 0, then show only default limits # LU-374 if userid is not None and userid: GetControlPanelUsers('userid', userid) # LU-530 elif reseller is not None: GetControlPanelUsers('list-reseller-users', reseller=reseller) else: GetControlPanelUsers('list-users') except Exception: pass more_fields = ["PACKAGE"] result = _formatter(_pprint, more_fields=more_fields) fields = get_fields() + more_fields formatter = _user_formatter(fields) for user in packages_users: result += formatter(user) if JSON: print('{"data":[' + ','.join(result) + ']}') def paneluserslist(): # type: () -> List[Tuple[int, str, str]] """Get list of tuples[lve_id, reseller, package] from control panel""" GetControlPanelUsers('list-users') result = [] for str_uid, payload in packages_users.items(): result.append((int(str_uid), payload['reseller'], payload['package'])) return result def panelpackagesdict(): # type: () -> Dict[str, List[str]] """Get dict of pairs[provider, list[package_name]] from control panel""" from clveconfig import DEFAULT_PROVIDER # NOQA packages = {} GetControlPanelUsers('list-packages') # admin's packages are already in bytes... packages[DEFAULT_PROVIDER] = list(packages_users.keys()) GetControlPanelUsers('list-resellers-packages') # ..but we must convert reseller's package to bytes, because cl-summary # expects bytes and print warnings about unicode comparison packages.update(packages_users) return packages # lvectl all-user-list def all_users_limits(): """ Implements lvectl all-user-list command :return: None, prints result to stdout """ get_XML_cfg() GetControlPanelUsers('list-users') result = _formatter(_pprint_f, more_fields=["PACKAGE", "RESELLER"]) fields = get_fields() + ["PACKAGE", "RESELLER"] formatter = _user_formatter(fields, printer=_pprint_f) for user in packages_users: result += formatter(user) if JSON: print('{"data":[' + ','.join(result) + ']}') def _filtering_da_admins(ve_dict): """ Filtering DirectAdmin's admins for `lvectl apply all` command :param ve_dict: dict with LVE :return: filtering dict """ if cldetectlib.getCPName() == 'DirectAdmin': # get list of uids DirectAdmin's admins uids_da_admins = [pwd.getpwnam(user).pw_uid for user in admins()] ve_dict = {key: value for key, value in ve_dict.items() if key not in uids_da_admins} return ve_dict def prepare_apply_data(lvp_id=0): try: # update packages_users global dict GetControlPanelUsers() packages_users_ = dict(packages_users) if lvp_id: # filter for apply lve top containers cfg_lvp_id_list = lve.map.name_map.id_list() packages_users_ = {k: v for k, v in packages_users.items() if k in cfg_lvp_id_list} except Exception: packages_users_ = {} if lvp_id is True: node_list = ve_lvp id_list = lvp_list() else: node_list = ve_lve id_list = lve.proc.lve_id_list(lvp_id=lvp_id) # get xml node for each lve_id # ve_dict is a local dict with lve_id and node with limit for lve_id # keys - int lve_id # data - xml node or None ve_dict = {} for node in node_list: ve_dict[get_ve_lve_user_uid(ve_lve_element=node)] = {'node' : node, 'reseller' : None} for lve_id in id_list: if lve_id not in ve_dict: # add lve_id for LVE that are not in ve.cfg ve_dict[lve_id] = {'node': None, 'reseller': None} if (len(packages_users_) != 0): # filtering addon admins DA. # package_users contain only users, not addon admins DA ve_dict = _filtering_da_admins(ve_dict) for uid in packages_users_: if uid not in ve_dict: node = None else: node = ve_dict[uid]['node'] # add lve_id for users that have package assigned pkg = packages_users_[uid] resellers = guess_reseller_by_package(pkg) if len(resellers) > 0: ve_dict[uid] = {'node' : node, 'reseller' : resellers[0]} else: ve_dict[uid] = {'node' : node, 'reseller' : None} return ve_dict def lve_destroy_and_recreate_all(): lve_ve_dict, lve_lvp_map = _get_lve_ve_dict_and_lvp_map() remaning_alive_lves = set(lve.proc.lve_id_list()) # NOTE: First we handle all LVEs not belonging to any LVP. for lve_id in lve_ve_dict.keys(): if lve_lvp_map.get(lve_id, 0) != 0: # This code path will never be triggered if reseller limits are disabled remaning_alive_lves.discard(lve_id) continue if lve_id in remaning_alive_lves: lve.lve_destroy(lve_id) remaning_alive_lves.discard(lve_id) # TODO(vlebedev): Is this context manager really necessary here? with lve.py.context_ignore_error(ignore_error=True): lve_apply( lve_id, out_node=lve_ve_dict[lve_id]['node'], reseller=lve_ve_dict[lve_id]['reseller'], ) # NOTE: Reset the default LVE settings. lve_apply(lve_id=0) if lve.reseller_limit_supported(): lvp_ve_dict = prepare_apply_data(True) remaining_alive_lvps = set(lve.proc.lvp_id_list()) # NOTE: Now we handle LVPs and LVEs inside those LVPs. for lvp_id in lvp_ve_dict.keys(): if lvp_id in remaining_alive_lvps: pylve.lve_lvp_destroy(lvp_id) remaining_alive_lvps.discard(lvp_id) _create_if_necessary_and_configure_lvp(lvp_ve_dict, lvp_id) # apply reseller's users reseller_name = lve.map.get_reseller_name(lvp_id) kernel_mapping = lve.proc.map() for user in clcommon.cpapi.reseller_users(reseller_name): lve_id = pwd.getpwnam(user).pw_uid if lve_id in remaning_alive_lves: lve.lve_destroy(lve_id) remaning_alive_lves.discard(lve_id) if kernel_mapping.get(lve_id, 0) != lvp_id: lve.py.lve_lvp_move( lvp_id, lve_id, err_msg=f'Can`t move lve_id={lve_id} to lvp_id={lvp_id}; error code {{code}}' ) lve_apply(lve_id=lve_id, reseller=reseller_name) # NOTE: Destroy any remaining live LVP that was not handled above. for lvp_id in remaining_alive_lvps: pylve.lve_lvp_destroy(lvp_id) # TODO(vlebedev): Not sure if this cleanup (and the one for LVPs above) is # really necesary but at least it faithfully replicates the # behaviour of `lvectl destroy all` part. # NOTE: Destroy any remaining live LVE that was not handled above. for lve_id in remaning_alive_lves: lve.lve_destroy(lve_id) # Apply all users and resellers settings def lve_apply_all(): ve_dict, lve_lvp_map = _get_lve_ve_dict_and_lvp_map() with lve.py.context_ignore_error(ignore_error=True): for lve_id in ve_dict.keys(): if lve_lvp_map.get(lve_id, 0) == 0: lve_apply(lve_id, out_node=ve_dict[lve_id]['node'], reseller=ve_dict[lve_id]['reseller']) lve_apply(lve_id=0) # apply limits for all LVP and LVEs inside LVP if not lve.reseller_limit_supported(): return ve_dict = prepare_apply_data(True) kernel_mapping = lve.proc.map() for lvp_id_ in ve_dict.keys(): _create_if_necessary_and_configure_lvp(ve_dict, lvp_id_) # apply reseller's users reseller_name = lve.map.get_reseller_name(lvp_id_) for user in clcommon.cpapi.reseller_users(reseller_name): lve_id_ = pwd.getpwnam(user).pw_uid # LU-511: create mapping before lve if needed if kernel_mapping.get(lve_id_, 0) != lvp_id_: lve.py.lve_lvp_move( lvp_id_, lve_id_, err_msg=f'Can`t move lve_id={lve_id_} to lvp_id={lvp_id_}; error code {{code}}' ) lve_apply(lve_id=lve_id_, reseller=reseller_name) def _get_lve_ve_dict_and_lvp_map() -> tuple[dict, dict]: get_XML_cfg() GetControlPanelUsers('list-users') lve_ve_dict = prepare_apply_data() lve_lvp_map = dict(lve.lve_id_lvp_id_pairs()) if lve.reseller_limit_supported() else {} return lve_ve_dict, lve_lvp_map def _create_if_necessary_and_configure_lvp(ve_dict, lvp_id) -> None: # apply lvp limits using defaults for lvp # load reseller defaults instead of gloabal get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True, load_config_elements=False) lve_apply(lve_id=lvp_id, out_node=ve_dict[lvp_id]['node'], reseller=ve_dict[lvp_id]['reseller'], lvp_id=lvp_id) get_XML_cfg(lvp_id=lvp_id, load_config_elements=False) lve_apply(lve_id=0, lvp_id=lvp_id) def _remove_reseller(lvp_id): """Remove reseller from ve.cfg and from procfs.""" get_global_lock(True) get_XML_cfg(lvp_id=lvp_id) for el in ve_lvp: if get_ve_lve_user_uid(ve_lve_element=el) == lvp_id: users = lve.proc.map_lve_id_list(lvp_id) # move containers to host for lve_id in users[:]: pylve.lve_lvp_move(0, lve_id) try: pwd.getpwuid(lve_id) except KeyError: if lve.py.lve_exists(lve_id): lve.lve_destroy(lve_id) users.remove(lve_id) lvp_destroy(lvp_id) # destroy container el.parentNode.removeChild(el) # remove record save_xml(ve_cfg) # load defaults host settings for end users get_XML_cfg(lvp_id=0) # apply limits from config (including package limits) for lve_id in users: lve_apply(lve_id) get_XML_cfg(lvp_id=lvp_id) return True return False def disable_reseller_limits(reseller_name, lvp_id): """Disable reseller limits and call hooks""" if _remove_reseller(lvp_id): reseller_limits_disabled_post.throw_event(reseller=reseller_name) else: if JSON: json_format('multi', ['WARNING', f'no configuration found for LVP {lvp_id}']) sys.exit(-1) else: print(f'warning: no configuration found for LVP {lvp_id}') # Delete User from ve.cfg and set default lve settings def lve_delete(lve_id): get_global_lock(True) get_XML_cfg() Deleted = False for el in ve_lve: if get_ve_lve_user_uid(ve_lve_element=el) == lve_id: Deleted = True lve_destroy(lve_id) lve_create(lve_id) el.parentNode.removeChild(el) save_xml(ve_cfg) get_XML_cfg() lve_apply(lve_id) if not Deleted: if JSON: json_format('multi', ['WARNING', f'no configuration found for VE {lve_id}']) sys.exit(-1) else: print(f'warning: no configuration found for VE {lve_id}') def lve_enter_check(): if not os.path.exists('/proc/lve/enter'): if JSON: json_format('multi', ['WARNING', 'enter by name not supported']) else: print('warning: enter by name not supported') sys.exit(-1) def enter_apply(sign, binary): lve_enter_check() try: msg = sign + binary.strip() with open('/proc/lve/enter', 'w', encoding='utf-8') as f: f.write(msg) except Exception: pass def list_binaries(): get_XML_cfg() if JSON: result = '{"data":[' first = True for el in ve_binary: path = el.getAttribute('path') if first: result += '"' + path + '"' first = False else: result += ',"' + path + '"' result += ']}' print(result) else: print("Binaries") for el in ve_binary: print(el.getAttribute('path')) def load_binaries(): get_XML_cfg() for el in ve_binary: enter_apply('+', el.getAttribute('path')) def reload_binaries(): lve_enter_check() with open('/proc/lve/enter', 'r', encoding='utf-8') as f: for line in f: enter_apply('-', line) load_binaries() def del_binary(binary): global ve_binary get_global_lock(True) lve_enter_check() get_XML_cfg() deleted = False for el in ve_binary: if el.getAttribute('path') == binary: deleted = True enter_apply('-', binary) el.parentNode.removeChild(el) save_xml(ve_cfg) get_XML_cfg() if not deleted: if JSON: json_format('multi', ['WARNING', f'no configuration found for {binary}']) else: print(f'warning: no configuration found for {binary}') sys.exit(-1) def set_binary(binary): global ve_binary global ve_enter_by_name get_global_lock(True) get_XML_cfg() for el in ve_binary: if el.getAttribute('path') == binary: return # nothing to do, it is already there enter_apply('+', binary) bin_xml = ve_cfg.createElement('binary') bin_xml.setAttribute('path', binary) ve_enter_by_name.appendChild(bin_xml) save_xml(ve_cfg) get_XML_cfg() def lve_set_default(set_data, package_flag, is_needed, lvp_id=0): """ Set given lve or package to default values for given parameters :param dict set_data: Arguments of lvectl call :param bool package_flag: Should we delete package or lve with given id :param callable is_needed: Function that takes xml element and set_data dict and returns whether current xml element contains info about needed ID from set_data """ try: if package_flag: data = ve_package elif lvp_id: data = ve_lvp else: data = ve_lve el = [e for e in data if is_needed(e, set_data)][0] except IndexError: return if lvp_id: # for lvectl set-reseller {id} --default=A,B,C; remove limit record in ve.cfg for tag_ in set_data['set-default']: if tag_ == 'ep': n = xml_filter_first(el, 'other', 'maxentryprocs') else: n = xml_filter_first(el, tag_, 'limit') if n: n.parentNode.removeChild(n) return to_keep = set(LIMITS_LIST_NAME) - set_data['set-default'] for limit in to_keep: if limit == 'ep' and len(el.getElementsByTagName('other')) > 0: # dict.setdefault isn't lazy evaluated if limit not in set_data: set_data[limit] = el.getElementsByTagName('other')[0].getAttribute('maxentryprocs') elif len(el.getElementsByTagName(limit)) > 0: # dict.setdefault isn't lazy evaluated if limit not in set_data: set_data[limit] = el.getElementsByTagName(limit)[0].getAttribute('limit') # delete this lve or package if package_flag: plan_delete(set_data['ve_id']) else: lve_delete(set_data['ve_id']) def _check_reseller_user_pair(uid, reseller_name): """ Checks is uid owned by reseller :param uid: uid for check :param reseller_name: Reseller name, None treats as root :return: True - valid reseller/user pair, False - else Special case: if reseller_name is None (root) - always valid """ if reseller_name in (None, 'root'): return True # reseller is not root # determine username username = get_main_username_by_uid(uid) if username in ('root', 'N/A'): # user is root or no such user -- error return False # determine users of supplied reseller's container. try: # Get reseller's users list reseller_users_list = reseller_users(reseller_name) except Exception: # any error - ignore, reseller is root reseller_users_list = [] if username in reseller_users_list: return True return False # Set limits for user # TODO: split this method into several independent: # - enable_reseller_limits # - set_reseller_default_limits # - set_reseller_limits # - set_lve_limits def lve_set(set_data, lvp_id=0): # set_data example: # {'iops': 2222, 'reseller_name': 'res', 'save': False, 've_id': 1023, 'pmem': 524288} # 524288 4096 = 2G -- pmem=2G if lvp_id == 0: # Set limits for user's LVE, check reseller/user match reseller_name = set_data.get('reseller_name', None) lve_id = set_data['ve_id'] if not _check_reseller_user_pair(lve_id, reseller_name): return False global setup_data get_global_lock(True) if lvp_id and lvp_id == set_data['ve_id']: # reseller's container limits if lve.proc.exist_lvp(lvp_id): # reseller's container exists... load info about his container get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True) else: # reseller's container does not exists... create new one with default limits get_XML_cfg(lvp_id=lvp_id) elif lvp_id == 0 and lve.reseller_limit_supported(): # user's limits (reseller & not) get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id'])) else: # default limits and limits for user when reseller's does not supported get_XML_cfg(lvp_id=lvp_id) try: GetControlPanelUsers() except Exception: pass try: # LU-366. Fix reset user's limits to unlimited if user in reseller package # and reseller is not root/admin package = packages_users[set_data['ve_id']] resellers = guess_reseller_by_package(package) reseller = resellers[0] if resellers else '' prepare_setup_data(package, reseller=reseller) except Exception: setup_data = ve_defaults if set_data['ve_id'] != 0: has_ve = False # set default def is_needed_user(el, set_data): return get_ve_lve_user_uid(ve_lve_element=el) == set_data['ve_id'] if 'set-default' in set_data: lve_set_default(set_data, package_flag=False, is_needed=is_needed_user, lvp_id=lvp_id) if 'ncpu' in set_data: lncpu = int(set_data['ncpu']) else: lncpu = ve_defaults['ncpu'] # check that cpu value in any format (cpu, speed=% or speed=[m\|g]hz) is equal or not cpu_is_different = True if 'cpu' in set_data: setted_cpu = convert_to_kernel_format(set_data['cpu'], lncpu = lncpu) if setted_cpu == setup_data['cpu']: cpu_is_different = False if lvp_id: el_list = ve_lvp else: el_list = ve_lve # choose top level container for modifications for el in el_list: if is_needed_user(el, set_data): for key in LIMITS_LIST_NAME: if key in set_data: try: if key == 'ep': set_child_tag_atrr(el, 'other', 'maxentryprocs', set_data[key]) else: set_child_tag_atrr(el, key, 'limit', set_data[key]) except (ValueError, IndexError, TypeError): # we already checked cpu value, so use cpu_is_different result if key == "cpu": is_different = cpu_is_different # otherwise compare with default in usual way else: is_different = setup_data[key] != set_data[key] if is_different or set_data['save']: if key == 'ep': node = ve_cfg.createElement('other') node.setAttribute('maxentryprocs',str(set_data[key])) else: node = ve_cfg.createElement(key) node.setAttribute('limit',str(set_data[key])) el.appendChild(node) if not set_data.get('skip-update-cfg', False): save_xml(ve_cfg) has_ve = True if lvp_id and lvp_id == set_data['ve_id']: # reseller's container limits if lve.proc.exist_lvp(lvp_id): # reseller's container does not exists... create new one with default limits get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True) else: # reseller's container exists... load info about his container get_XML_cfg(lvp_id=lvp_id) elif lvp_id == 0 and lve.reseller_limit_supported(): # user's limits (reseller & not) get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id'])) else: # default limits and limits for user when reseller's does not supported get_XML_cfg(lvp_id=lvp_id) lve_apply(set_data['ve_id'], lvp_id=lvp_id) else: pass if not has_ve and set_data['ve_id']: el_name = LVP_XML_TAG_NAME if lvp_id else 'lve' el = ve_cfg.createElement(el_name) if lvp_id: # for create resellers limit config # set reseller_name reseller_id map el.setAttribute('id', str(lvp_id)) el.setAttribute('user', set_data['user']) el.appendChild(ve_default) # copy default limits to reseller else: if set_data.get('save-username'): el.setAttribute('user', pwd.getpwuid(set_data['ve_id']).pw_name) else: el.setAttribute('id', str(set_data['ve_id'])) for key in LIMITS_LIST_NAME: if key in set_data: # we already checked cpu value, so use cpu_is_different result if key == "cpu": is_different = cpu_is_different # otherwise compare with default in usual way else: is_different = setup_data[key] != set_data[key] if is_different or set_data['save']: if key == 'ep': node = ve_cfg.createElement('other') node.setAttribute('maxentryprocs',str(set_data[key])) else: node = ve_cfg.createElement(key) node.setAttribute('limit',str(set_data[key])) el.appendChild(node) added = False for el2 in ve_package: el2.parentNode.insertBefore(el,el2) added = True break if not added: ve_cfg.lastChild.appendChild(el) if not set_data.get('skip-update-cfg', False): save_xml(ve_cfg) if lvp_id: enables_reseller_limits = not lve.proc.exist_lvp(lvp_id) # load lvp defaults and lvp tag, set limits for reseller get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True) lve_apply(set_data['ve_id'], lvp_id=lvp_id) # copy default limits from host container pylve.lve_set_default(lvp_id, pylve.lve_info(0)) # load lvp tag and reseller's end user defaults get_XML_cfg(lvp_id=lvp_id) reseller_name = lve.map.get_reseller_name(lvp_id) for lve_id_ in lve.map.lvp_lve_id_list(lvp_id=lvp_id): lve.py.lve_lvp_move(lvp_id, lve_id_) lve_apply(lve_id_, reseller=reseller_name) # call hook if we enabled reseller limits if enables_reseller_limits: reseller_limits_enabled_post.throw_event(reseller=reseller_name) else: if lve.reseller_limit_supported(): get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id'])) else: get_XML_cfg(lvp_id=lvp_id) lve_apply(set_data['ve_id']) else: for key in LIMITS_LIST_NAME: if key in set_data: if key == 'ep': ve_default.getElementsByTagName('other')[0].setAttribute('maxentryprocs',str(set_data[key])) else: ve_default.getElementsByTagName(key)[0].setAttribute('limit',str(set_data[key])) if not set_data.get('skip-update-cfg', False): save_xml(ve_cfg) get_XML_cfg(lvp_id=lvp_id) lve_apply(set_data['ve_id'], lvp_id=lvp_id) return True def package_set(set_data, is_reseller=False): """ Set package with some heuristic algorithm to simulate old package set behavior """ get_global_lock(True) get_XML_cfg() # Removed in LU-351 # set_data['ve_id'] = unicode(set_data['ve_id'].decode('utf-8')) reseller_list = guess_reseller_by_package(set_data['ve_id']) if len(reseller_list) == 0: reseller = None elif len(reseller_list) >= 1: # if dublicated packages found - use first as a reseller reseller = reseller_list[0] #if ve.cfg has tag <version>2</version> - trying to guess reseller name by package #if reseller is undef or ve.cfg has not tag version - work with ver 1 if reseller is not None and ve_cfg_version > 1: set_data['reseller_name'] = reseller package_set_ext(set_data, is_reseller=True) else: package_set_ext(set_data, is_reseller=False) # Set new package or modify exist # package-set-ext def package_set_ext(set_data, is_reseller=False): get_global_lock(True) get_XML_cfg() has_package = False if is_reseller: def is_needed_plan(el, set_data): return ( el.getAttribute('id') == set_data['ve_id'] and el.getAttribute('reseller') == set_data['reseller_name'] ) else: def is_needed_plan(el, set_data): return el.getAttribute('id') == set_data['ve_id'] and not el.getAttribute('reseller') if 'set-default' in set_data: lve_set_default(set_data, package_flag=True, is_needed=is_needed_plan) for el in ve_package: if is_needed_plan(el, set_data): for key in LIMITS_LIST_NAME: if key in set_data: try: if key == 'ep': el.getElementsByTagName('other')[0].setAttribute('maxentryprocs',str(set_data[key])) else: el.getElementsByTagName(key)[0].setAttribute('limit',str(set_data[key])) except (ValueError, IndexError, TypeError): if key == 'ep': node = ve_cfg.createElement('other') node.setAttribute('maxentryprocs',str(set_data[key])) else: node = ve_cfg.createElement(key) node.setAttribute('limit',str(set_data[key])) el.appendChild(node) if cldetectlib.is_plesk(): plesk_id = _plesk_get_package_id(set_data.get('reseller_name', ''), set_data['ve_id']) el.setAttribute(XML_PLESK_ID, str(plesk_id)) has_package = True if not has_package: package_reseller = '' el = ve_cfg.createElement('package') el.setAttribute('id', set_data['ve_id']) if is_reseller: el.setAttribute('reseller', set_data['reseller_name']) package_reseller = set_data['reseller_name'] if cldetectlib.is_plesk(): plesk_id = _plesk_get_package_id(package_reseller, set_data['ve_id']) el.setAttribute(XML_PLESK_ID, str(plesk_id)) for key in LIMITS_LIST_NAME: if key in set_data: if key == 'ep': node = ve_cfg.createElement('other') node.setAttribute('maxentryprocs',str(set_data[key])) else: node = ve_cfg.createElement(key) node.setAttribute('limit',str(set_data[key])) el.appendChild(node) ve_cfg.lastChild.appendChild(el) save_xml(ve_cfg) get_XML_cfg() copy_package_settings_to_cpanel(set_data) if 'ncpu' in set_data: lncpu = int(set_data['ncpu']) else: lncpu = ve_defaults['ncpu'] if 'cpu' in set_data: set_data['cpu'] = convert_to_kernel_format(set_data['cpu'], lncpu=lncpu) reseller = set_data['reseller_name'] if is_reseller else None plan_apply(set_data['ve_id'], reseller=reseller) def _plesk_get_package_id(reseller: str, package: str) -> Optional[int]: """ Find the right package id from plesk DB query """ panel = detect_panelclass() packages = panel.list_domain_packages_with_id() try: pack = next(filter( lambda x: x[0] in {reseller, 'root'} and x[1] == package, # no reseller == reseller is root (admin) packages )) return pack[2] except StopIteration: return None def get_reseller_packages_map(): """ Retrives resellers to packages map from panel using /usr/bin/getcontrolpaneluserspackages :return: Dictionary: { 'reseller1' -> ['pack1', 'pack2'], 'reseller2' -> ['pack'] } """ global packages_users packages_users_copy = packages_users.copy() GetControlPanelUsers('list-resellers-packages') reseller_packages_map = packages_users packages_users = packages_users_copy return reseller_packages_map def reseller_package_set(set_data): """ Set reseller package limits :param set_data: input data dictionary :return: True - limits was set succesfully False - supplied provider has no supplied package """ # set limits to package that belongs to given reseller reseller_name = set_data['reseller_name'] package_name = set_data['ve_id'] # Retrive resellers packages from panel reseller_packages_map = get_reseller_packages_map() # If reseller has supplied package -- set limit if reseller_name in reseller_packages_map and package_name in reseller_packages_map[reseller_name]: # Reseller/package pair valid - set limit package_set_ext(set_data, is_reseller=True) return True # ERROR: Supplied reseller has no supplied package return False def copy_package_settings_to_cpanel(set_data): """ Copy package limits from ve.cfg to cpanel packages data """ package = set_data['ve_id'] if not cldetectlib.is_cpanel(): return # skip func if panel not cPanel package_path = f'/var/cpanel/packages/{package}' if not os.path.isfile(package_path): return # skip func if no cPanel packages found with open(package_path, 'r', encoding='utf-8') as f: cpanel_package_data = f.readlines() new_cpanel_package_data = cpanel_package_data[:] old_cpanel_data = {} # proces old_cpanel_package_data - get old limits and remove stings from it # result of processing - cpanel_package_data_modify and old_cpanel_data for line in cpanel_package_data: if line.startswith('lve_'): line_parts = line.strip().split('=') limit_name = line_parts[0].replace('lve_', '').strip() if line_parts[1] != 'DEFAULT': old_cpanel_data[limit_name] = line_parts[1] # get old_limits if limit_name in LIMITS_LIST_NAME: new_cpanel_package_data.remove(line) if line.startswith('_PACKAGE_EXTENSIONS') and 'lve' not in line: return # skip func if no lve extention install to the package for limit_name in ('pmem', 'mem', 'vmem'): if limit_name in old_cpanel_data: memory_page_value = clcommon.memory_to_page(old_cpanel_data[limit_name]) old_cpanel_data[limit_name] = memory_page_value or ve_defaults[limit_name] if is_limits_equals(old_cpanel_data, set_data): return # skip writeting to file - limits are equals # create and add to new cpanel_data_file limits lines like: # lve_ + limit_name + = + limit_value cpanel_data = create_cpanel_limits(package, ve_package) for limit_name in LIMITS_LIST_NAME: limit_value = cpanel_data[limit_name] limit_line = f'lve_{limit_name}={limit_value}\n' new_cpanel_package_data.append(limit_line) write_file_via_tempfile(''.join(new_cpanel_package_data), package_path, 0o644) def is_limits_equals(old_limits, new_limits): """ check if new set of limits for package are equals to used """ for key in new_limits.keys(): if key in ('ve_id', 'save'): continue # ve_id == package name. skip this key try: if old_limits[key] != new_limits[key]: return False except KeyError: return False return True def create_cpanel_limits(package_id, xml_packages): """ create limits for cpanel package file use data from ve.cfg: limit = limit if found in ve.cfg or DEFAULT return dict """ result_data = {} for el in xml_packages: if el.getAttribute('id') == package_id: for limit in LIMITS_LIST_NAME: try: if limit == 'ep': result_data[limit] = str( el.getElementsByTagName('other')[0].getAttribute('maxentryprocs') ).strip() elif limit in ("mem", "vmem", "pmem"): result_data[limit] = str( clcommon.page_to_memory( int(el.getElementsByTagName(limit)[0].getAttribute("limit")) ) ).strip() else: result_data[limit] = str(el.getElementsByTagName(limit)[0].getAttribute('limit')).strip() except (ValueError, IndexError, TypeError): result_data[limit] = 'DEFAULT' return result_data # Delete plan from ve.cfg def plan_delete(plan_id, reseller_name=None): get_global_lock(True) get_XML_cfg() Deleted = False if reseller_name is None: def is_needed_package(el): return el.getAttribute('id') == plan_id and not el.getAttribute('reseller') else: def is_needed_package(el): return el.getAttribute('id') == plan_id and el.getAttribute('reseller') == reseller_name for el in ve_package: if is_needed_package(el): Deleted = True el.parentNode.removeChild(el) save_xml(ve_cfg) break if not Deleted: # try to guess reseller name only if no reseller name if reseller_name is None: resellers_list = guess_reseller_by_package(plan_id) if len(resellers_list) == 0: reseller = None else: # if some resellers found - use first reseller = resellers_list[0] # try to delete package only if we guess reseller if reseller is not None: plan_delete(plan_id, reseller) return if JSON: json_format( 'multi', ['WARNING', f'no configuration found for plan {plan_id}'] ) sys.exit(-1) else: print(f'warning: no configuration found for plan {plan_id}') lve_apply_all() def reseller_plan_delete(plan_id, reseller_name): plan_delete(plan_id, reseller_name=reseller_name) def get_xml_limit(el, key): try: return str(el.getElementsByTagName(key)[0].getAttribute('limit')) except (ValueError, IndexError): # convert from kernel format to output return ve_defaults[key] if key != 'cpu' else f'{ve_defaults[key] // 100}%' def _normalize_str(data_str): """ Normalize string for JSON output. Example: - Input string: -_&[{}]'"`te\\s/t\a - Output string: -_&[{}]'\"`te\\\\s/t\\a :param data_str: String for normalize :return: Normalied string """ def _get_char_index(input_string, char_to_search, ordinal): """ Get the index of the specified occurrence of character in string :param input_string: String :param char_to_search: Character to search :param ordinal: Required occurence number :return: Char index """ count = 0 for idx, ch in enumerate(input_string): if ch == char_to_search: # Char found count += 1 if count == ordinal: return idx # Char not found return -1 if data_str is None: return None json_str = json.dumps({'str': data_str}) # json_str example: {"str": "-_&[{}]'\"`te\\\\s/t\\a"}, # get '-_&[{}]'\"`te\\\\s/t\\a' from it # Get third " index trd_idx = _get_char_index(json_str, '"', 3) # Get Last " index last_idx = json_str.rfind('"') return json_str[trd_idx + 1:last_idx] def _package_formatter(fields, is_reseller=False, printer=None): """ Generate inner function with closured fields names, is_reseller flag and printer function :param list fields: List of strings that represent names of fields in final output :param boolean is_reseller: Format output with info about reseller or not :param callable printer: Function to format and print data for every entry :rtype: callable :return: function to format data for every user """ printer = printer if printer is not None else _pprint_r if is_reseller else _pprint_p def wrapper(package_name, reseller_name=None): """ :param string package_name: Find and format data for this package name :param string reseller_name: reseller name, owner of supplied package :rtype: list :return: List of giver package's statistics data line or empty list """ # in order to avoid unicode warnings here if is_reseller: # Reseller package _load_resellers_xml_data(reseller_name) data = copy.copy(ve_defaults) # only after reading reseller's xml data['reseller'] = reseller_name def is_needed_package(el): return package_name == el.getAttribute('id') and el.getAttribute('reseller') == reseller_name else: # Admin's package def is_needed_package(el): return package_name == el.getAttribute('id') and not el.getAttribute('reseller') data = copy.copy(ve_defaults) data['id'] = _normalize_str(package_name) if JSON else package_name def convert_mem_limits(value): return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value) data['vmem'] = convert_mem_limits(data['mem']) if LVE_VERSION > 4: data['pmem'] = convert_mem_limits(data['pmem']) for el in ve_package: if is_needed_package(el): lncpu = get_xml_limit(el, 'ncpu') data['ncpu'] = lncpu if lncpu != '' else str(ve_defaults['ncpu']) data['speed'] = str(convert_to_kernel_format( get_xml_limit(el, 'cpu'), lncpu=int(data['ncpu'])) // 100 ) try: data['ep'] = str(int( el.getElementsByTagName('other')[0].getAttribute('maxentryprocs') )) except (IndexError, ValueError): pass data['pmem'] = convert_mem_limits(get_xml_limit(el, 'pmem')) data['vmem'] = convert_mem_limits(get_xml_limit(el, 'mem')) data['io'] = get_xml_limit(el, 'io') data['nproc'] = get_xml_limit(el, 'nproc') data['iops'] = get_xml_limit(el, 'iops') if data.get('speed') is None: # convert from kernel format for output data['speed'] = data['cpu'] // 100 data['cpu'] = str(speed_to_old_cpu(data['speed'])) res = [] if JSON: line = ','.join(f'"{f}":"{data.get(f.lower(), "N/A")}"' for f in fields) res = [f'{{{line}}}'] else: printer([data.get(f.lower(), '') for f in fields]) return res return wrapper # lvectl package-list def get_packages_list(): get_XML_cfg() GetControlPanelUsers('list-packages') packages = packages_users.copy() GetControlPanelUsers('list-resellers-packages') reseller_packages = packages_users.copy() result = _formatter(_pprint_p, default_id=DEFAULT_PACKAGE) formatter = _package_formatter(get_fields(), is_reseller=False, printer=_pprint_p) for package in packages: result += formatter(package) if ve_cfg_version > 1: formatter = _package_formatter(get_fields(), is_reseller=True, printer=_pprint_p) # reseller_packages: {'reseller_name': ['pack1', 'pack2']} for reseller_name, packages_list in reseller_packages.items(): # On DA skip all non-admin's packages if cldetectlib.is_da() and reseller_name != 'admin': continue for reseller_package in packages_list: if ve_cfg_version > 1: result += formatter(reseller_package, reseller_name) else: result += formatter(reseller_package) if JSON: print('{"data":[' + ','.join(result) + ']}') # lvectl reseller-package-list def get_resellers_packages_list(): get_XML_cfg() GetControlPanelUsers('list-resellers-packages') more_fields = ["RESELLER"] fields = get_fields() + more_fields result = _formatter(_pprint_p, default_id=DEFAULT_PACKAGE) formatter = _package_formatter(fields, is_reseller=True) # packages_users: {'reseller_name': ['pack1', 'pack2']} for reseller_name, packages_list in packages_users.items(): for reseller_package in packages_list: result += formatter(reseller_package, reseller_name) if JSON: print('{"data":[' + ','.join(result) + ']}') # lvectl all-package-list def get_all_packages_list(): get_XML_cfg() GetControlPanelUsers('list-packages') packages = packages_users.copy() GetControlPanelUsers('list-resellers-packages') reseller_packages = packages_users.copy() more_fields = ["RESELLER"] fields = get_fields() + more_fields # make header with default package result = _formatter(_pprint_r, default_id=DEFAULT_PACKAGE, more_fields=more_fields) formatter = _package_formatter(fields, is_reseller=False, printer=_pprint_r) for package in packages: result += formatter(package) # Print resellers packages formatter = _package_formatter(fields, is_reseller=True) # reseller_packages: {'reseller_name': ['pack1', 'pack2']} for reseller_name, packages_list in reseller_packages.items(): for reseller_package in packages_list: result += formatter(reseller_package, reseller_name) if JSON: print('{"data": [' + ','.join(result) + ']}') cached_resellers_packages = None cached_list_packages = None cached_users = None cached_reseller_users = None cached_default = None def _convert_packages_list(package_list): """ Converts package list to internal format :param package_list: Package list. Example: ['BusinessPackage', 'Package2'] :return: Package list as dictionary. Example: {'BusinessPackage': 'BusinessPackage', 'Package2': 'Package2'} """ packages_users_dict = {} for package in package_list: packages_users_dict[package] = package return packages_users_dict # Get users from control panel with plans def GetControlPanelUsers(option='list-all', lve_package_id='', reseller=None): """ Parse output from GET_CP_PACKAGE_SCRIPT and get package and lve relations :param option: option for GET_CP_PACKAGE_SCRIPT. Option is one from the following possible values: 'userid', 'package', 'list-packages', 'list-resellers-packages' :type option: string :param lve_package_id: lve_id or package_name :type lve_package_id: string or int :param reseller: :type reseller: string """ global cached_list_packages global cached_resellers_packages global cached_users global cached_reseller_users global cached_default global packages_users # Check arguments if option not in ('list-all', 'userid', 'package', 'list-packages', 'list-resellers-packages', 'list-users', 'list-reseller-users'): return False if option in ('userid', 'package') and lve_package_id == '': return False from clcommon.cpapi import ( # pylint: disable=import-outside-toplevel admin_packages, get_reseller_users, get_uids_list_by_package, list_all, list_users, reseller_package_by_uid, resellers_packages, ) try: if option == 'userid': if cached_users is not None: try: packages_users = { lve_package_id: { 'package': cached_users[lve_package_id]['package'], 'reseller': cached_users[lve_package_id]['reseller'] } } except KeyError: packages_users = {lve_package_id: {'package': '', 'reseller': ''}} else: try: reseller_name, package = reseller_package_by_uid(lve_package_id) except ValueError: # this is possible on vm without control panel reseller_name = package = '' packages_users = {lve_package_id: {'package': package, 'reseller': reseller_name}} return True elif option == 'package': # reseller - optional argument packages_users = {lve_package_id: get_uids_list_by_package(lve_package_id, reseller)} return True elif option == 'list-packages': # Result format: # {'BusinessPackage': 'BusinessPackage', 'Package2': 'Package2'} if cached_list_packages is None: package_list = admin_packages() # Convert to output format packages_users = _convert_packages_list(package_list) cached_list_packages = packages_users else: # list-packages data already present packages_users = cached_list_packages return True elif option == 'list-resellers-packages': # Result format: # {'res2 SimplePackage': 'res2 Package', # 'res1 BusinessPackage': 'res1 UltraPackage'} if cached_resellers_packages is None: # in order to produce same results as code that works with Popen packages_users = resellers_packages() cached_resellers_packages = packages_users else: # list-resellers-packages data already present packages_users = cached_resellers_packages return True elif option == 'list-users': # Result format: # {1000: {'reseller': '', 'package': 'Package1'}, # 1001: {'reseller': '', 'package': 'BusinessPackage'}, # } if cached_users is None: cached_users = list_users() packages_users = cached_users return True elif option == 'list-reseller-users': # {1001: {'reseller': 'res1', 'package': 'BusinessPackage'}, # 1004: {'reseller': 'res1', 'package': 'BusinessPackage'}} if cached_reseller_users is None: reseller_users_dict = get_reseller_users(reseller) # for uid, user_data in reseller_users_dict.iteritems(): # packages_users[uid] = user_data cached_reseller_users = reseller_users_dict packages_users = reseller_users_dict else: # list-reseller-users data already present packages_users = cached_reseller_users return True elif option == 'list-all': # deprecated. TODO: Remove this option # Result format: # {1000: 'Package1', 1001: 'BusinessPackage'} if cached_default is None: cached_default = list_all() packages_users = cached_default return True except EncodingError as e: raise_cpanel_encoding_error(e) except OSError: pass return False def get_panel_users_count(): """ Retrieves panel users count :return: """ GetControlPanelUsers() return len(packages_users) # Apply plan settings for users def plan_apply(plan_id, reseller=None): # fill the cache to speedup `lvectl package-set-ext` with many users in one package GetControlPanelUsers("list-users") if GetControlPanelUsers("package", plan_id, reseller=reseller): for uid in packages_users[plan_id]: lve_apply(int(uid), plan_id, reseller=reseller) # Destroy many LVEs from stdin def destroy_many(users_list): for line in users_list: line = line.replace('\n','') users = line.strip().split() for user in users: if (len(user) != 0): try: user = int(user) lve_destroy(user) except Exception: pass # Apply many LVEs from stdin def apply_many(users_list): get_XML_cfg() try: GetControlPanelUsers() except Exception: pass for line in users_list: line = line.replace('\n','') users = line.strip().split() for user in users: if (len(user) != 0): try: user = int(user) lve_apply(user) except Exception: pass # Put pid into LVE def limit_pid(lve_id, pid, flags): pylve.lve_enter_pid_flags( int(lve_id), int(pid), flags, err_msg=f'Can`t put proccess with pid {pid} in lve {lve_id}; error code {{code}}' ) # Get pid from LVE def release_pid(pid): pylve.lve_leave_pid(int(pid), err_msg=f'Can`t release process with pid {pid}') def get_globals(): global ve_cfg global ve_lveconfig global ve_default global ve_lve global ve_defaults global ve_package global ve_binary global ve_enter_by_name global ubc return {'ve_cfg': ve_cfg, 've_lveconfig': ve_lveconfig, 've_default': ve_default, 've_lve': ve_lve, 've_defaults': ve_defaults, 've_package': ve_package, 'ubc': ubc, 've_enter_by_name': ve_enter_by_name} def guess_reseller_by_package(package): reseller = [] global packages_users pkg_users_old = packages_users.copy() GetControlPanelUsers('list-resellers-packages') reseller_packages = packages_users.copy() packages_users = pkg_users_old.copy() # reseller_packages: {'reseller_name': ['pack1', 'pack2']} for reseller_name, packages_list in reseller_packages.items(): for package_name_in_key in packages_list: if package == package_name_in_key: reseller.extend([reseller_name]) return reseller # LU-400 def call_endurance_custom_script(args): """ Call Endurance's custom script :param args: list of arguments for pass to Endurance's custom script :return: None """ endurance_custom_script = cldetectlib.get_param_from_file(cldetectlib.CL_CONFIG_FILE, 'ENDURANCE_CUSTOM_SCRIPT', separator='=') if endurance_custom_script and os.path.isfile(endurance_custom_script): ret_code, std_out = exec_utility(endurance_custom_script, args) if ret_code != 0: message = f'Error while executing Endurance\'s custom script\n{std_out}' if JSON: json_format('multi', ['ERROR', message]) else: err_message = f"error: {message}" sys.stderr.write(f"{err_message}\n") sys.exit(ret_code) def _page_to_memory_or_bytes(value): """ Convert page value to human-readable value or bytes, depending on BYTES_FLAG; E.g. >>> _page_to_memory_or_bytes(1233254) # BYTES_FLAG=False '100M' >>> _page_to_memory_or_bytes(1233254) # BYTES_FLAG=True 654321 :type value: int :rtype: str \| int """ if BYTES_FLAG: return int(round(value mmap.PAGESIZE)) return _mb_mem(value) def remove_absent_resellers(): """ Remove from LVE all resellers, which are absent from panel :return: None """ # Build resellers ids list for removal reseller_id_list_for_delete = [] # Get resellers list from cpapi # Create a list from a generator for repeated membership testing cpapi_resellers_list = list(lve.map.resellers()) get_XML_cfg() # for loading reseller_name<=>reseller_id map form ve.cfg # If reseller present in LVE, but absent in panel - add it to list for removal for lve_reseller_id in lvp_list(): try: lve_reseller_name = lve.map.get_reseller_name(lve_reseller_id) if lve_reseller_name not in cpapi_resellers_list: # Reseller does not exist in panel, remove it from LVE reseller_id_list_for_delete.append(lve_reseller_id) except (KeyError, OSError, IOError): # No such user, remove it from LVE reseller_id_list_for_delete.append(lve_reseller_id) # Remove all selected resellers ignoring all errors (for example no such reseller) for reseller_id_for_delete in reseller_id_list_for_delete: _remove_reseller(reseller_id_for_delete) def remove_absent_users(): """ Remove from LVE all users, which absent in system :return: None """ for lve_id in lve.proc.map(): try: pwd.getpwuid(lve_id) # Check the existence of the user except KeyError: try: lve.lve_destroy(lve_id) # Destroy lve, if user not exist except PyLveError: # If lve not exist try: lve.py.lve_create(lve_id) # we create lve lve.lve_destroy(lve_id) # and destroy them again # After destroing lve, mapping will be cleansed of absent lve_id except PyLveError: pass Back

Edit File : lvectllib.py | Size : 102.55 KB

# coding=utf-8
# Liblve functions lib

#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import contextlib
import copy
import errno
import json
import math
import mmap
import os
import pwd
import re
import subprocess
import sys
import syslog
import warnings
import xml.dom.minidom as xml
from builtins import range
from functools import partial
from mmap import PAGESIZE
from typing import Dict, List, Optional, Text, Tuple, TypedDict  # NOQA

import unshare

import clcommon
import cldetectlib
import lveapi
from clcommon.const import Feature
from clcommon.cpapi import admins, get_main_username_by_uid, is_panel_feature_supported, reseller_users
from clcommon.cpapi.cpapiexceptions import EncodingError
from clcommon.lock import acquire_lock
from clcontrollib import detect_panelclass
from clevents import reseller_limits_disabled_post, reseller_limits_enabled_post
from cllimits.lib import exec_utility
from cllvectl.log import get_subprocess_logger
from clveconfig.ve_config import BadVeConfigException, get_xml_config, save_xml
from clveconfig.ve_lock import LockFailedException, setup_global_lock
from lveapi import LVP_XML_TAG_NAME, Lve, NameMap, PyLve, PyLveError
from secureio import create_dir_secure, write_file_via_tempfile

GET_CP_PACKAGE_SCRIPT = '/usr/bin/getcontrolpaneluserspackages'

CPUINFO = '/proc/cpuinfo'
CORE_WEIGHT = 10000
DEFAULT_PACKAGE = "VE_DEFAULT"

NOIOPS = False

UMOUNT = '/bin/umount'
EXCLUDE_MOUNTS_CONF = '/etc/container/exclude_mounts.conf'

MULTI_FORMAT = 'multi'
SINGLE_FORMAT = 'single'

IS_DEBUG = int(os.environ.get('PYLVE_DEBUG', 0))

XML_PLESK_ID = 'plesk_id'   # XML attribute name to bind package name in the ve.cfg with plesk DB id

if not is_panel_feature_supported(Feature.LVE):
    pylve = None
    lve = None
else:
    pylve = PyLve(debug=IS_DEBUG)
    lve = Lve(py=pylve)

class LiblveSettings(TypedDict):
    ls_cpu: int | str
    ls_cpus: int
    ls_io: int
    ls_enters: int
    ls_memory_phy: int
    ls_nproc: int
    ls_iops: int

def create_liblve_settings(**kwargs) -> LiblveSettings:
    defaults: LiblveSettings = {
        'ls_cpu': 0,
        'ls_cpus': 0,
        'ls_io': 0,
        'ls_enters': 0,
        'ls_memory_phy': 0,
        'ls_nproc': 0,
        'ls_iops': 0
    }
    defaults.update(kwargs)
    return defaults

def lvp_list():
    """Helper function for easy mocking in unittests"""
    if lve.reseller_limit_supported():
        return lve.proc.lvp_id_list()
    return []

def get_active_resellers():
    """
    Get list of resellers with activated reseller limits

:return: list of pairs (name, uid)
    """
    name_map = NameMap()
    name_map.link_xml_node()
    return name_map.load_from_node()

def is_active_reseller_limits(reseller_name):
    """
    Check whether giver reseller has activated reseller limits or not
    :return: bool
    """
    return reseller_name in (name for name, uid in get_active_resellers())

# TODO: py3 move it to cllib/long script
def raise_cpanel_encoding_error(e: EncodingError):
    """
    Since cPanel user can corrupt config file for some user with wrong encodings,
    we want to notify him that he should fix encoding problems with the link to documentation.
    Print error message and exit with code 1 or raise given exception if it isn't cPanel.
    :return: None
    """
    if not cldetectlib.is_cpanel():
        raise e

if JSON:
        json_format('multi',
                    ['ERROR', str(e)])
    else:
        print(e)
    sys.exit(1)

def get_global_lock(write=False):
    """
    ~~~~~~~~~~~~~~~~~~
    !!! DEPRECATED !!!
    ~~~~~~~~~~~~~~~~~~
    Please, use setup_global_lock instead if possible

Wrapper over setup_global_lock. If lock cannot be set,
    it will write message and close app

The only reason why it is here is legacy function
    check_result_and_exit that we use in TWO places
    :type write: bool
    :return: Nothing
    """
    try:
        setup_global_lock(write)
    except LockFailedException:
        check_result_and_exit(1, 'can`t get lock')

def check_result_and_exit(result, message):
    # on cl5 some func is unimplemented; so ENOSYS is not error;
    if result not in (0, -errno.ENOSYS):
        if JSON:
            json_format(MULTI_FORMAT, ['ERROR', f'lvectl: {message}'])
        else:
            print(f'lvectl: Error: {message}')
        sys.exit(result)

# Default parameters for lve
LVE_DEFAULT = {
    'cpu': 25,
    'ncpu': 1,
    'io': 25,
    'ep': 20,
    'mem': 0,
    'pmem': 262144,
    'nproc': 0,
    'iops': 1024

}
MEM_DEFAULT_CL5 = 262144
# Default parameters for lvp
LVP_DEFAULT = {
    'cpu': 100,
    'ncpu': 1,
    'io': 0,
    'ep': 0,
    'mem': 0,
    'pmem': 0,
    'nproc': 0,
    'iops': 0
}

LIMITS_LIST_NAME = ['ncpu', 'cpu', 'io', 'mem', 'pmem', 'nproc', 'iops', 'ep']

LVE_VERSION = 4
JSON = False
BYTES_FLAG = False

# defined structures for liblve and turples for functions
lve_settings = ''
setup_data = ''  # type: dict

# dict with user-packages relations
# keys = int UID or str package name
# data = string package name
packages_users = {}

# defined ve.cfg variables
ve_cfg = ''
ve_lveconfig = ''
ve_default = ''
ve_lve = ''
ve_lvp = ''  # for resellers limits
ve_defaults = ''  # type: dict
ve_package = ''
ubc = 'false'  # TODO: looks like not used anymore, check and remove it
ve_enter_by_name = ''
ve_binary = ''
ve_cfg_version = ''

# Set JSON if json output required
def set_json(json_flag):
    global JSON
    JSON = json_flag

def set_bytes(bytes_flag):
    global BYTES_FLAG
    BYTES_FLAG = bytes_flag

def get_fields():
    if NOIOPS and LVE_VERSION == 8:
        version = 'noiops_8'
    elif LVE_VERSION == 8:
        version = '8'
    elif LVE_VERSION == 6:
        version = '6'
    else:                       # LVE_VERSION == 4
        version = '4'
    fields = {
        'noiops_8': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO'],
        '8': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO','IOPS'],
        '6': ['ID','SPEED','PMEM','VMEM','EP','NPROC','IO'],
        '4': ['ID','SPEED','VMEM','EP','IO']
    }[version]
    if JSON:
        speed_idx = fields.index('SPEED') + 1
        return (fields[:speed_idx] + ["CPU"] + fields[speed_idx:])
    return fields

# Create structure
def init(lve_ver=None):
    global LVE_VERSION
    if lve_ver is None:
        lve_ver = clcommon.get_lve_version()
        if lve_ver[0] is None:
            raise RuntimeError('get_lve_version failed')
        LVE_VERSION = lve_ver[0]
    else:
        LVE_VERSION = lve_ver
    global lve_settings

lve_status = pylve.initialize()
    if not lve_status:
        raise RuntimeError('init_lve() failed.')

lve_settings = pylve.liblve_settings()

# we use /proc/cpuinfo to get cpu speed, but unfortunately
# it returns current CPU MHZ, which is different in lve environment
# and we cannot get right speed value there
# FIXME: LU-947
def _get_cpu_data_from_env():
    """Get cpu information from environment veriable"""
    packed_cpu_data = os.environ.get('CPU_DATA')
    if packed_cpu_data is None:
        return None

try:
        return json.loads(packed_cpu_data)
    except (TypeError, ValueError) as e:
        print('Invalid environment variable \'CPU_DATA\' format', str(e))
        sys.exit(1)

def get_cpu_data():
    """
    Parse /proc/cpuinfo

return [NumProc, frequency in MHZ]

"""

cpuinfo = {}
    procinfo = {}
    nprocs = 0
    try:
        # f = open(CPUINFO, 'r')
        with open(CPUINFO, 'r', encoding='utf-8') as f:
            for line in f:
                if not line.strip():
                    # end of one processor
                    cpuinfo[f'proc{nprocs}'] = procinfo
                    nprocs = nprocs + 1
                    # Reset
                    procinfo = {}
                else:
                    if len(line.split(':')) == 2:
                        procinfo[line.split(':')[0].strip()] = line.split(':')[1].strip()
                    else:
                        procinfo[line.split(':')[0].strip()] = ''
    except IOError:
        print(f'lvectl: Error: Can`t open {CPUINFO}.')
        sys.exit(1)

return [nprocs, cpuinfo['proc0']['cpu MHz']]

# It's extremely rare case when CPU changes at runtime so we will use cached
CPUINFO_DATA = _get_cpu_data_from_env() or get_cpu_data()

def convert_from_old_cpu(data, lncpu=0):
    """
    Try converting to kernel format from old CPU format (percentage of whole
    cpu) and optionally the NCPU format. Return whichever is less.

:param data: string presumably in old CPU format
    :param lncpu: integer number of cores limit
    """
    data = str(data)
    lncpu = lncpu or 0
    cpu_data = CPUINFO_DATA
    ncpu = int(cpu_data[0])
    cpu_percent = re.match(r'\d{1,2}0?$', data)  # 0-100
    if cpu_percent is not None:
        data = int(data)
        if 0 < data <= 100:
            from_cpu_limit = int(round(CORE_WEIGHT // 100 * ncpu * data))
            if lncpu == 0:
                return from_cpu_limit
            return min(lncpu * CORE_WEIGHT, from_cpu_limit)
    return None

def convert_from_speed_percent(data):
    """
    Try converting cpu limit from SPEED in percentage of one CORE format to
    kernel format.
    """
    data = str(data)
    cpu_data = CPUINFO_DATA
    ncpu = int(cpu_data[0])
    percent = re.match(r'\d+(?:\.\d+)?%$', data)  # *%
    if percent is not None:
        percent = float(data.replace('%', ''))
        if percent > ncpu * 100:
            percent = ncpu * 100
        if percent > 0:
            return int(round(CORE_WEIGHT // 100 * percent))
        return None
    return None

def convert_from_speed_hz(data):
    """
    Try converting cpu limit from SPEED in mhz/gzh format to kernel format.
    """
    data = str(data)
    cpu_data = CPUINFO_DATA
    ncpu = int(cpu_data[0])
    cpu_freq = float(cpu_data[1])
    pattern = re.compile(r'(?P<freq>\d+(?:\.\d+)?)(?P<suffix>mhz|ghz)+$', re.IGNORECASE)
    match = pattern.match(data)  # *mhz\ghz
    if match is not None:
        suffix = match.group('suffix')
        freq = float(match.group('freq'))
        if suffix.upper() == 'GHZ':
            freq = freq * 1000
        if freq > cpu_freq * ncpu:
            freq = cpu_freq * ncpu
        if freq > 0:
            return int(round(freq * CORE_WEIGHT / cpu_freq))
    return None

def convert_from_speed(data):
    """
    Try converting cpu limit value from either SPEED limit format
    (percentage of CORE or mhz/ghz) to kernel format.
    """
    return (
        convert_from_speed_percent(data)
        or convert_from_speed_hz(data)
    )

def convert_to_kernel_format(data, lncpu=0):
    """
    Convert different variants of cpu limit to kmod ver 8 variant

:param data: Value in old CPU format or SPEED with % or mhz/ghz.
    :param lncpu: Limit in old NCPU format.
    :return: CPU limit in kmod ver 8+ format or None for bad format
    """
    from_cpu = convert_from_old_cpu(data, lncpu)
    if from_cpu is not None:
        return from_cpu
    from_speed_percent = convert_from_speed_percent(data)
    if from_speed_percent is not None:
        return from_speed_percent
    from_speed_hz = convert_from_speed_hz(data)
    if from_speed_hz is not None:
        return from_speed_hz

return None

def speed_to_old_cpu(speed):
    """
    convert speed to old cpu format
    args: cpu limit in speed value
    return: old cpu limit format
    """
    cpu_data = CPUINFO_DATA
    nproc = int(cpu_data[0])
    speed = str(speed)
    if '*' in speed:
        return '*' + str(int(round(int(speed.lstrip('*')) // nproc)))
    return str(int(round(int(speed) // nproc)))

@contextlib.contextmanager
def temporary_lve(settings):
    # type: (pylve.liblve_settings) -> contextlib.GeneratorContextManager
    """
    Run subprocess in lve with pseudo-random id and given limits
    """
    pylve.initialize()
    lve_id = pylve.get_available_lve_id()
    try:
        pylve.lve_setup(lve_id, settings)
    except PyLveError:
        syslog.syslog(syslog.LOG_ALERT, f"Unable to setup lve with id {lve_id}, "
                                        "something is wrong, check dmesg for details")
        raise

try:
        yield lve_id
    finally:
        pylve.lve_destroy(lve_id)

def make_liblve_settings(settings: LiblveSettings):
    # type: (LiblveSettings) -> pylve.liblve_settings
    """
    Just a nice user-friendly constructor of liblve_settings object
    You can pass the following ls_cpu and ls_cpus values:
    - in percents of one core (just ls_cpu='75%', ls_cpus will be ignored)
    - in old 'CPU' format (two arguments, ls_cpu and ls_cpus required, both int)
    """
    s = pylve.liblve_settings()
    s.ls_cpu = convert_to_kernel_format(settings['ls_cpu'], lncpu=settings['ls_cpus'])
    s.ls_io = settings['ls_io']
    s.ls_enters = settings['ls_enters']
    s.ls_nproc = settings['ls_nproc']
    s.ls_iops = settings['ls_iops']
    # convert memory from bytes to mempages
    s.ls_memory_phy = int(math.ceil(1. * settings['ls_memory_phy'] / PAGESIZE))
    return s

def get_ve_lve_user_uid(ve_lve_element):
    user_uid = str(ve_lve_element.getAttribute('id'))
    if not user_uid:
        user_name = ve_lve_element.getAttribute('user')
        user_uid = pwd.getpwnam(user_name).pw_uid
    return int(user_uid)

def json_format(error_type, data, extensions=None):
    """
    Print output in json as:
    {"status": "ERROR/OK", "msg": "Some Message", "ext1": "foo", "ext2": "bar"}
    where "status" and "msg" field are mandatory

:param str error_type: Either MULTI_ERROR or SINGLE_ERROR
    :param list data: List with a status string and a message string
    :param dict extensions: Some additional fields for the final json object
    :return: None
    """
    result = {'status': str(data[0])}
    if error_type == MULTI_FORMAT:
        result['msg'] = str(data[1])
    if extensions is not None:
        result.update(extensions)
    print(json.dumps(result))

def check_def_value(xml, ve_defaults, ve_cfg, val, default):
    try:
        ve_defaults[val] = int(ve_default.getElementsByTagName(val)[0].getAttribute('limit'))
    except (ValueError, IndexError, TypeError):
        ve_defaults[val] = default[val]
        node = ve_cfg.createElement(val)
        node.setAttribute('limit',str(default[val]))
        try:
            xml.appendChild(node)
        except Exception:
            pass

def xml_filter_tag(node, tag):
    return [_ for _ in node.childNodes if isinstance(_, xml.Element) and _.tagName == tag]

def xml_filter_first(node, tag, attr=None, attr_val=None):
    for child_node in xml_filter_tag(node, tag):
        if attr is not None and not child_node.hasAttribute(attr):
            continue
        if attr_val is not None and child_node.getAttribute(attr) != attr_val:
            continue
        return child_node

def get_child_tag_atrr(node, tag, attr):
    filtered_child_node = xml_filter_first(node=node, tag=tag, attr=attr)
    if filtered_child_node is None:
        raise IndexError()
    return filtered_child_node.getAttribute(attr)

def set_child_tag_atrr(node, tag, attr, val):
    """
    Find in children nodes node with tag and setup attribute
    insted el.getElementsByTagName not search recursiveli in tree
    """
    first_child_node = xml_filter_tag(node, tag)[0]
    first_child_node.setAttribute(attr, str(val))

def _load_config_wrapper():
    """Load config from ve.cfg"""
    global ve_cfg
    global ve_lveconfig

try:
        ve_cfg, ve_lveconfig = get_xml_config()
    except BadVeConfigException as e:
        if JSON:
            json_format(MULTI_FORMAT, ['ERROR', str(e)])
        else:
            print(str(e))
        sys.exit(1)

def _load_default_limits(lvp_id: int, lvp_defaults: bool):
    """Load default limits

:param int lvp_id: lvp id
    :param bool lvp_defaults: load reseller's default limits instead of global
    :return: dict with default limits
    """
    global ve_default
    global ubc

ubc = 'true'

try:
        if not lvp_id:
            ve_default = xml_filter_tag(ve_lveconfig, 'defaults')[0]
            return LVE_DEFAULT

if lvp_defaults:
            ve_default = ve_cfg.createElement('defaults')
            return LVP_DEFAULT

defaults_root_node = xml_filter_first(ve_lveconfig, LVP_XML_TAG_NAME, 'id', str(lvp_id))
        if defaults_root_node:  # if no such reseller with lvp in config
            ve_default = defaults_root_node.getElementsByTagName('defaults')[0]
        else:
            ve_default = ve_default.cloneNode(ve_default)
        return LVE_DEFAULT

except IndexError:
        if JSON:
            json_format('multi', ['WARNING', 'default section error in ve.cfg'])
            sys.exit(1)
        else:
            print('warning: default section error in ve.cfg')

return LVE_DEFAULT

def _all_config_elements_loaded():
    return all(x != '' for x in (ve_lve, ve_lvp, ve_package, ve_binary, ve_enter_by_name, ve_cfg_version))

def _load_config_elements():
    """Load all config elements from ve.cfg"""
    global ve_lve
    global ve_lvp
    global ve_package
    global ve_binary
    global ve_enter_by_name
    global ve_cfg_version

ve_lve = ve_lveconfig.getElementsByTagName("lve")
    ve_package = ve_lveconfig.getElementsByTagName("package")
    ve_lvp = ve_lveconfig.getElementsByTagName(LVP_XML_TAG_NAME)
    lve.map.name_map.link_xml_node(ve_lveconfig)

enter_by_name_elems = ve_lveconfig.getElementsByTagName('enter-by-name')
    if len(enter_by_name_elems) > 0:
        ve_enter_by_name = enter_by_name_elems[0]
    else:
        ve_enter_by_name = ve_cfg.createElement('enter-by-name')
        ve_lveconfig.appendChild(ve_enter_by_name)

ve_binary = ve_enter_by_name.getElementsByTagName('binary')

# expected version tag in next format
    #
    # <lveconfig>
    #   <version>2</version>
    #   <system>
    #   ....
    #   </system>
    #   ....
    # </lveconfig>

cfg_version_elems = ve_lveconfig.getElementsByTagName('version')
    ve_cfg_version = int(cfg_version_elems[0].firstChild.nodeValue) if len(cfg_version_elems) > 0 else 1

def _load_ve_defaults(default_limits: Dict[str, int]):
    """Create ve_defaults dict with default values for all limits

:param dict default_limits: default limits for lve or lvp
    """
    global ve_cfg
    global ve_defaults
    global ve_default

ve_defaults = {}
    check_def_val = partial(
        check_def_value,
        xml=ve_default,
        ve_defaults=ve_defaults,
        ve_cfg=ve_cfg,
        default=default_limits,
    )
    check_def_val(val='ncpu')

try:
        speed = ve_default.getElementsByTagName('cpu')[0].getAttribute('limit')
        ve_defaults['cpu'] = convert_to_kernel_format(speed, lncpu=ve_defaults['ncpu'])
    except (ValueError, IndexError, TypeError):
        ve_defaults['cpu'] = convert_to_kernel_format(default_limits['cpu'], lncpu=ve_defaults['ncpu'])
        cpu = ve_cfg.createElement('cpu')
        cpu.setAttribute('limit', str(default_limits['cpu']))
        try:
            ve_default.appendChild(cpu)
        except Exception:
            pass

try:
        ve_defaults['ep'] = int(ve_default.getElementsByTagName('other')[0].getAttribute('maxentryprocs'))
    except (ValueError, IndexError, TypeError):
        ve_defaults['ep'] = default_limits['ep']
        ep = ve_cfg.createElement('other')
        ep.setAttribute('maxentryprocs', str(default_limits['ep']))
        try:
            ve_default.appendChild(ep)
        except Exception:
            pass

check_def_val(val='io')
    if LVE_VERSION > 5:
        check_def_val(val='mem')
    else:
        check_def_val(val='mem', default={'mem': MEM_DEFAULT_CL5})  # pylint: disable=redundant-keyword-arg
    check_def_val(val='pmem')
    check_def_val(val='nproc')
    check_def_val(val='iops')

_check_defaults_for_nones()

def _check_defaults_for_nones():
    """Check that all default values are not None"""
    global ve_defaults

for key, value in ve_defaults.items():
        if value is not None:
            continue

err_msg = f'ERROR: Incorrect {key} default value'
        if JSON:
            json_format('multi', ['ERROR', err_msg])
        else:
            sys.stderr.write(f'{err_msg}\n')
        sys.exit(1)

def get_XML_cfg(lvp_id=0, lvp_defaults=False, load_config_elements=True):
    """
    :param bool lvp_defaults: load reseller's default limits instead of global
    :param int lvp_id: lvp id to load customise defaults
    """

_load_config_wrapper()
    default_limits = _load_default_limits(lvp_id, lvp_defaults)

# load config elements if load_config_elements=True or they were not loaded before
    if load_config_elements or not _all_config_elements_loaded():
        _load_config_elements()

_load_ve_defaults(default_limits)

def check_value(val, el, ve_defaults, setup_data):
    try:
        value = int(get_child_tag_atrr(el, tag=val, attr='limit'))
        setup_data[val] = value
        return value
    except (ValueError, IndexError, TypeError):
        return int(ve_defaults[val])

def _load_resellers_xml_data(reseller, xml_config_load_elements=True):
    """
    This function is a pure workaround for our ugly globals-based API which
    should be fixed partially with LU-496, because there is no clean way
    to retrieve reseller's data from ve.cfg without touching globals
    :param reseller: reseller name
    :return: Nothing. It just updates some globals
    """
    # TODO after LU-496 we should read and cache all reseller's settings
    name_map = lveapi.NameMap()
    name_map.link_xml_node()
    reseller_id = name_map.get_id(reseller)
    get_XML_cfg(reseller_id, load_config_elements=xml_config_load_elements)

def prepare_setup_data(plan_id=None, reseller=None, lve_id=None):
    # type: (Optional[Text], Optional[Text]) -> None
    """
    Put limit values that will be applied later in a global variable `setup_ve`.

:param plan_id: package
    :param reseller:
        If reseller is None we only inherit from admin packages.
        In that case we ignore all tags in ve.cfg with a "reseller" attribute.
    """

global setup_data
    setup_data = copy.copy(ve_defaults)
    if plan_id is not None:
        res_pkg_dict = get_reseller_packages_map()
        # LU-510: Investigate the problem with reseller's list, part 2.
        #   Fix applying limit for reseller user with admin package
        if reseller is not None and reseller in res_pkg_dict and plan_id in res_pkg_dict[reseller]:

def is_needed_plan(el):
                return el.getAttribute('id') == plan_id and el.getAttribute('reseller') == reseller

else:
            if cldetectlib.is_da() and lve_id is not None:
                try:
                    user_pwd = pwd.getpwuid(lve_id)
                    filename = f'/usr/local/directadmin/data/users/{user_pwd.pw_name}/user.conf'
                    with open(filename, encoding='utf-8') as f:
                        text = f.read()
                except Exception:
                    text = ''
                # LU-1663 --> LU-3410:
                #   it is normal situation for DA to set DA(not LVE) user package to "custom"
                #   in this case the package is still correct and this warning should not be logged
                if 'package=custom' not in text:
                    syslog.syslog(
                        syslog.LOG_ALERT,
                        f"Package for user with id {lve_id} is incorrect, please recover it using Note from "
                        "https://docs.cloudlinux.com/cloudlinux_os_components/#installation-enabling-and-disabling",
                    )

# Ignore all tags in ve.cfg with a "reseller" attribute.
            def is_needed_plan(el):
                return el.getAttribute('id') == plan_id and not el.getAttribute('reseller')

# Example command when `ve_package` is not empty:
        #   cloudlinux-packages set --json --for-reseller root --package kekage --pmem 994 --nproc 77
        # `ve_package` is set by reading ve.cfg in `get_XML_cfg`.
        # <package> tag is added to ve.cfg in `package_set_ext`.
        for el in ve_package:
            if is_needed_plan(el):
                lncpu = check_value('ncpu', el, ve_defaults, setup_data)
                try:
                    cpu = int(convert_to_kernel_format(get_child_tag_atrr(el, tag='cpu', attr='limit'), lncpu=lncpu))
                    setup_data['cpu'] = cpu
                except (ValueError, IndexError, TypeError):
                    pass
                check_value('io', el, ve_defaults, setup_data)
                if (ubc == 'true'):
                    check_value('mem', el, ve_defaults, setup_data)
                else:
                    setup_data['mem'] = 0
                try:
                    ep = int(get_child_tag_atrr(el, tag='other', attr='maxentryprocs'))
                    setup_data['ep'] = ep
                except (ValueError, IndexError, TypeError):
                    pass
                check_value('nproc', el, ve_defaults, setup_data)
                check_value('pmem', el, ve_defaults, setup_data)
                check_value('iops', el, ve_defaults, setup_data)

def umount_dir(path):
    try:
        # run the "umount" command and suppress it's output, return True when child exit code is not zero
        with subprocess.Popen(
            [UMOUNT, "-l", path],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        ) as proc:
            proc.communicate()
            return proc.returncode != 0
    except OSError:
        check_result_and_exit(-1, f'failed to run "{UMOUNT} -l {path}"')

def prepare_mounts():
    """
    Unmount all paths from /proc/mounts that match regular expressions from /etc/container/exclude_mounts.conf file
    """
    if not os.path.isfile(EXCLUDE_MOUNTS_CONF):
        return
    reg_exp_list = []
    try:
        with open(EXCLUDE_MOUNTS_CONF, 'r', encoding='utf-8') as conf:
            for r in conf:
                pattern = r.strip()
                if pattern:
                    reg_exp_list.append(re.compile(pattern))
    except IOError:
        check_result_and_exit(-1, f'failed to read {EXCLUDE_MOUNTS_CONF}')
    if not reg_exp_list:
        return
    unshare.unshare(unshare.CLONE_NEWNS)
    try:
        with open('/proc/mounts', 'r', encoding='utf-8') as f:
            mounts = [m.split()[1] for m in f.readlines()]
    except (IndexError, IOError):
        check_result_and_exit(-1, 'failed to parse /proc/mounts')
    ATTEMPTS = 10
    for _ in range(ATTEMPTS):
        error = False
        for mount in mounts:
            for reg_exp in reg_exp_list:
                m = reg_exp.search(mount)
                if m:
                    error = umount_dir(mount) or error
                    break
        if not error:
            break

def lve_start():
    """
    Start LVE engine and initialize default mount namespace for LVE
    """
    MOUNT_CMD = '/bin/mount --make-rprivate / >/dev/null 2>&1'
    try:
        subprocess.call(MOUNT_CMD, shell=True, executable='/bin/bash')
    except OSError:
        print('Error: failed to execute', MOUNT_CMD)
    prepare_mounts()
    pylve.lve_start(err_msg='Can`t init lve default settings')

def lve_create(lve_id, ignore_error=False):
    """
    Create LVE container for given ID
    :type lve_id: int
    :type ignore_error: bool
    :return: Nothing
    """
    pylve.lve_create(lve_id,
                     err_msg=f'lvectl: Can`t create lve with id {lve_id}; error code {{code}}',
                     ignore_error=ignore_error)

def lvp_create(lvp_id, ignore_error=False):
    """
    Create LVP container for given ID
    :type lvp_id: int
    :type ignore_error: bool
    :return: Nothing
    """
    pylve.lve_lvp_create(lvp_id,
                         err_msg=f'lvectl: Can`t create lvp with id {lvp_id}; error code {{code}}',
                         ignore_error=ignore_error)

def destroy_lvp_all():
    logger = get_subprocess_logger('lvectllib')

destroyed_list = []
    for lvp_id in lvp_list():
        logger.debug('destroy_lvp_all: destroying LVP with id %s', lvp_id)
        pylve.lve_lvp_destroy(lvp_id,
                              err_msg=f'lvectl: Can`t destroy lvp with id {lvp_id}; error code {{code}}')
        destroyed_list.append(lvp_id)
    return destroyed_list

def lvp_destroy(lvp_id):
    logger = get_subprocess_logger('lvectllib')

if lvp_id == 'all':
        destroy_lvp_all()
        return

logger.debug('lvp_destroy: destroying LVP with id %s', lvp_id)
    pylve.lve_lvp_destroy(lvp_id,
                          err_msg=f'lvectl: Can`t destroy lvp with id {lvp_id}; error code {{code}}')

# Destroy LVE container for ID
def lve_destroy(lve_id):
    logger = get_subprocess_logger('lvectllib')

cant_remove_msg = f'Can\'t remove lve {lve_id} from kernel - error code -3'
    cant_destroy_msg = f'Can`t destroy lve with id {lve_id}; error code {{code}}'
    destroyed = False

if lve_id == 'all':
        if lve.reseller_limit_supported():
            destroyed = bool(destroy_lvp_all())  # destroy all top level containers
        if len(list(lve.proc.lve_id_list())) > 0:
            for id_ in lve.proc.lve_id_list():
                logger.debug('lve_destroy all: destroying LVE with id %s', id_)
                lve.lve_destroy(id_, err_msg=cant_destroy_msg)
                destroyed = True
        else:
            destroyed = True  # empty list - all lve already destroyed
    else:
        if lve.proc.check_inside_list(lve_id) or \
                (lve.proc.resellers_supported() and lve.proc.detect_inside_lvp(lve_id) is not None):
            logger.debug('lve_destroy: destroying LVE with id %s', lve_id)
            lve.lve_destroy(lve_id, err_msg=cant_destroy_msg)
            destroyed = True
        elif not lve.proc.check_inside_list(lve_id):
            destroyed = True  # lve_id doesn`t exist so it`s already destroy!

if destroyed:
        if JSON:
            json_format(SINGLE_FORMAT, ['OK'])
    else:
        if JSON:
            json_format(MULTI_FORMAT, ['WARN', cant_remove_msg])
        else:
            print(f'warning: {cant_remove_msg}')

# Setup LVE for ID
def lve_setup(lve_id, lvp_id=0):
    if lvp_id and not lve.proc.exist_lvp(lvp_id):
        # create lve top container if not exist
        lvp_create(lvp_id)

with warnings.catch_warnings():  # convert all warning to exceptions
        warnings.filterwarnings('error')
        try:
            lve_settings.ls_io = int(setup_data['io'])
            lve_settings.ls_cpu = int(setup_data['cpu'])
            lve_settings.ls_cpus = int(setup_data['ncpu'])
            lve_settings.ls_memory = int(setup_data['mem'])
            lve_settings.ls_enters = int(setup_data['ep'])

if LVE_VERSION > 5:
                lve_settings.ls_memory_phy = int(setup_data['pmem'])
                lve_settings.ls_nproc = int(setup_data['nproc'])
            if LVE_VERSION > 6:
                lve_settings.ls_iops = int(setup_data['iops'])
            if lvp_id:
                if lve_id == 0:
                    pylve.lve_set_default(
                        lvp_id, lve_settings, err_msg=f'Can`t setup default settings for LVP {lvp_id}')
                else:
                    pylve.lve_lvp_setup(
                        lvp_id, lve_settings, err_msg=f'Can`t setup lvp with id {lvp_id}; error code {{code}}')
            elif lve_id == 0:
                pylve.lve_set_default(lve_settings, err_msg='Can`t setup default settings')
            else:
                pylve.lve_setup(
                    lve_id, lve_settings, err_msg=f'Can`t setup lve with id {lve_id}; error code {{code}}')

except RuntimeWarning as rw:
            # exit if caught a warning - can`t set limits to lve
            check_result_and_exit(1,'Can`t setup lve ' + str(lve_id) + '. RuntimeWarning excepted: ' + str(rw))

def get_package_and_reseller_by_lve_id(lve_id):
    """
    Get pair of package, reseller for lve_id
    :param lve_id: lve_id, UID with package, reseller
    :return: tuple of (package, reseller); Both can be None
    """
    global packages_users
    package = None
    reseller = None
    # hate this! backup global ref to packages_users
    old_packages_users = packages_users
    GetControlPanelUsers('list-users')
    # now packages_users should be dict of dicts:
    # {lve_id : {'package' : , 'reseller':}}
    temp_package = packages_users
    # restore global ref to packages_users
    packages_users = old_packages_users
    if isinstance(temp_package, dict):
        try:
            reseller = temp_package[lve_id]['reseller']
            # if reseller is empty then reseller is root/admin. return to None
            if not reseller:
                reseller = None
        except KeyError:
            reseller = None
        try:
            package = temp_package[lve_id]['package']
            # if package is empty  - return to undefined state
            if not package:
                package = None
        except KeyError:
            package = None

return package, reseller

# (rprilipskii): Honestly, relying on global variables for storage like this leaves a bad
# taste in my mouth, but I'd have to rewrite a more lvectl code than I'd prefer if I
# wanted to do it in a more pythonic way. Technical debt is a problem.
CACHED_EFFECTIVE_LIMITS = {}
EFFECTIVE_CACHE_FILE = "/var/run/cloudlinux/effective-normal-limits"
BURSTABLE_LIMITS_FLAG_FILE = "/opt/cloudlinux/flags/enabled-flags.d/burstable-limits.flag"

def write_effective_cache(reset=False):
    """
    Save calculated effective normal limits for an LVE to a cache file.

The cache file is stored in /var/run (tmpfs) for faster operations.
    It disappears on reboot, but lvectl apply all runs in the lvectl systemd service anyway,
    so the file will always reappear if the service functions normally.

If additional speed is desired, consider replacing the standard JSON lib
    with a faster one like rapidjson or orjson.

:param reset: If True, recreate the cache file from scratch with data available in the dict,
    instead of updating it, defaults to False
    :type reset: bool, optional
    """
    # NOTE: When running `lvectl apply all`, this write will happen at the very end
    # of the operation. Applying the limits, however, happens before this - right after
    # they're calculated.
    # This means a substantial time gap and a potential race condition if someone calls
    # `lvectl set` while `apply all` is not yet finished (e.g. cache and kernel having different limits).
    # Could be solved by makin limit application happen after all effective limits are
    # done calculating, and wrapping limit application and cache writing in the same filelock.

# Do nothing if the corresponding feature flag is not set
    if not os.path.exists(BURSTABLE_LIMITS_FLAG_FILE):
        return

# If there's no /var/run/cloudlinux directory, create it.
    try:
        effective_dir = os.path.dirname(EFFECTIVE_CACHE_FILE)
        if not os.path.isdir(effective_dir):
            var_run_dir = os.path.dirname(effective_dir)
            create_dir_secure(effective_dir, 0o600, 0, 0, var_run_dir)
    except OSError as e:
        print(f"Error: failed to create the folder {effective_dir}: {e}")
        raise

# Load already cached effective limits and merge them with
    # the limits calculated during the current lvectl run.
    # Unless we want to discard them and start over, that is.
    effective_cache_lock = acquire_lock(f"{EFFECTIVE_CACHE_FILE}.lock")
    with effective_cache_lock:
        if reset:
            effective_limits = {}
        else:
            try:
                if os.path.isfile(EFFECTIVE_CACHE_FILE):
                    with open(EFFECTIVE_CACHE_FILE, "r", encoding="utf8") as readfile:
                        effective_limits = json.load(readfile)
                else:
                    effective_limits = {}
            except json.JSONDecodeError as e:
                print(f"Error: failed to parse the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}")
                raise
            except OSError as e:
                print(f"Error: failed to read the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}")
                raise
        effective_limits.update(CACHED_EFFECTIVE_LIMITS)

try:
            write_file_via_tempfile(json.dumps(effective_limits), EFFECTIVE_CACHE_FILE, 0o600)
        except OSError as e:
            print(f"Error: failed to save the effective normal limit cache file {EFFECTIVE_CACHE_FILE}: {e}")

def cache_effective_limits(lve_id):
    """
    Cache the calculated effective normal limits for an LVE to a dictionary.

setup_data is a global dict that contains the LVE configuration last applied.
    Holds only one entry - during operations like lvectl apply all, we enter this
    function repeatedly with setup_data containing info for one processed LVE each time.

These limits can be used by the Burstable Limits components
    to set/reset burst or normal limits without having to go
    through effective limit calculation or invoking lvectl as a middleman.
    Comparing the cached limits to current active limits (in kernel)
    also shows whether or not the LVE has burst limits active at the moment.

This function gets called in lve_apply, which means it also runs within:
    * lvectl apply all
    * lvectl apply-many
    * lvectl set
    * and others, that also call lve_apply

:param lve_id: LVE ID for which the limits are being saved.
    :type lve_id: int
    """
    global CACHED_EFFECTIVE_LIMITS
    CACHED_EFFECTIVE_LIMITS[str(lve_id)] = setup_data

# pylint: disable-msg=too-many-arguments
# Apply user settings
def lve_apply(lve_id, plan_id=None, result=False, reseller=None, out_node=None, lvp_id=0):
    """
    Aplly limits to LVE lve_id
    :param lve_id: lve id
    :type lve_id: int
    :param plan_id: package for user with lve_id. deprecated
    :type plan_id: string
    :param result: if True = don't apply limits. only create setup_data with actual limits
    :type result: boolean
    :param reseller: if True = plan_id is resellers plan. deprecated
    :type reseller: boolean
    :param out_node: node with limits for lve_id
    :type out_node: xml_node
    :param lvp_id: reseller container id; host container if 0
    """
    global setup_data
    global packages_users

old_packages_users = packages_users
    if lve_id != 0:
        GetControlPanelUsers('userid', lve_id)
        new_packages_users, packages_users = packages_users, old_packages_users

if ve_cfg == '':
        get_XML_cfg(lvp_id=lvp_id)

el = None
    if out_node is not None:
        el = out_node
    else:
        node_list = ve_lvp if lvp_id else ve_lve
        el = next(
            filter(lambda node: get_ve_lve_user_uid(ve_lve_element=node) == lve_id, node_list),
            None
        )

# reseller (lvp_id!=0) should not use package and default limits
    if lvp_id == 0:
        try:
            plan_id = new_packages_users[lve_id]['package']
        except (NameError, KeyError):
            plan_id = None

try:
            reseller = new_packages_users[lve_id]['reseller']
        except (NameError, KeyError):
            reseller = None

if el is not None:
        # get limits from package
        prepare_setup_data(plan_id, reseller=reseller)

# prepare custom limits for lve
        lncpu = check_value('ncpu', el, ve_defaults, setup_data)
        try:
            setup_data['cpu'] = convert_to_kernel_format(get_child_tag_atrr(el, tag='cpu', attr='limit'), lncpu=lncpu)
        except (ValueError, IndexError, TypeError):
            pass
        if setup_data['cpu'] is None:
            setup_data['cpu'] = ve_defaults['cpu']

check_value('io', el, ve_defaults, setup_data)

if ubc == 'true':
            check_value('mem', el, ve_defaults, setup_data)
        else:
            setup_data['mem'] = 0

try:
            setup_data['ep'] = int(get_child_tag_atrr(el, tag='other', attr='maxentryprocs'))
        except (ValueError, IndexError, TypeError):
            pass

check_value('nproc', el, ve_defaults, setup_data)
        check_value('pmem', el, ve_defaults, setup_data)
        check_value('iops', el, ve_defaults, setup_data)

else:
        # apply default limits
        prepare_setup_data(plan_id, reseller=reseller, lve_id=lve_id)
        if ubc == 'false':
            setup_data['mem'] = 0

if not result:
        cache_effective_limits(lve_id)
        # apply limits
        lve_setup(lve_id, lvp_id=lvp_id)

def _pprint(*fields):
    """
    Print data with the last column 30 symbols wide.
    Useful for printing data that contains package names.
    """
    formatted_string = _format_fields(fields, wide_indices=[len(fields) - 1])
    print(formatted_string)

def _pprint_f(*fields):
    """
    Print data with the two last columns 30 symbols wide.
    Useful for printing full data of every user with package name
    and reseller name.
    """
    formatted_string = _format_fields(
        fields,
        wide_indices=[len(fields) - 2, len(fields) - 1],
    )
    print(formatted_string)

def _pprint_p(*fields):
    """
    Print data with the first column 30 symbols wide.
    Useful for printing packages data.
    """
    formatted_string = _format_fields(fields, wide_indices=[0])
    print(formatted_string)

def _pprint_r(*fields):
    """
    Print data with the first and last columns 30 symbols wide.
    Useful for printing data with user names and package names.
    """
    formatted_string = _format_fields(fields, wide_indices=[0, len(fields) - 1])
    print(formatted_string)

def _format_fields(fields: tuple, wide_indices: list, width: int = 30) -> str:
    """
    Helper function to format fields
    based on specified indices for wide columns.

Args:
        fields: The fields to format.
        wide_indices: List of indices in the fields that should be wide.
        width: The width of the wide columns.

Returns:
        A formatted string with specified fields widened.
    """
    formatted_fields = []

for index, field in enumerate(fields):
        # Convert field to string to ensure compatibility with formatting
        field_str = str(field)

if index in wide_indices:
            formatted_fields.append(f"{field_str:>{width}}")
        else:
            formatted_fields.append(f"{field_str:>8}")

return ''.join(formatted_fields)

def _pmem_vmem_to_bytes_value(value):
    """
    Convert pmem or vmem limits to bytes value

:param value: pmem or vmem limits in kbytes value
    :return: bytes value of limit

"""
    # if value was changed we remove asterisk from value for counted this
    value = str(value)
    was_changed = isinstance(value, str) and value.startswith('*')
    value = value.replace('*', '')
    if value:
        value = int(value)
        value *= 4096
    else:
        value = 0
    # return asterisk in value if this was changed
    value = f'*{value}' if was_changed else value
    return value

def _mb_mem(value):
    """
    Convert amount of RAM to M format

:param string value: amount of memory in KB
    :rtype: string
    :return: amount of memory in MB like "1234M"
    """
    result = ''
    was_changed = False
    if isinstance(value, str) and value.startswith('*'):
        value = value[1:]
        was_changed = True
    try:
        v = int(value)
    except ValueError:
        return ""
    if was_changed:
        result = '*'
    value = v * 4 // 1024
    if value > 0:
        result = f'{result}{value}M'
    else:
        result = f'{result}{v * 4}K'
    return result

def _formatter(printer, default_id="0", default_package="default", more_fields=None):
    """
    Generate header and default package data either as print to stdout or as json string
    """
    defaults = ve_defaults.copy()
    #  convert from kernel format for output
    defaults['cpu'] = defaults['cpu'] // 100

def get_data(key):
        return defaults.get(key, '')

_cpu = speed_to_old_cpu(get_data("cpu")) if get_data("cpu") != '' else ''

def convert_mem_limits(value):
        return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value)

fields_map = {
        'ID': default_id, 'SPEED': str(get_data('cpu')),
        'CPU': str(_cpu), 'NCPU': str(get_data('ncpu')),
        'PMEM': str(convert_mem_limits(get_data('pmem'))), 'VMEM': str(convert_mem_limits(get_data('mem'))),
        'EP': str(get_data('ep')), 'NPROC': str(get_data('nproc')), 'IO': str(get_data('io')),
        'IOPS': str(get_data('iops')), 'PACKAGE': default_package
    }
    res = []
    fields = get_fields()
    if more_fields is not None:
        fields += more_fields
    if JSON:
        line = ','.join(f'"{f}":"{fields_map.get(f, "")}"' for f in fields)
        res = [f'{{{line}}}']
    else:
        printer(*fields)
        printer(*[fields_map.get(f, "") for f in fields])
    return res

def _user_formatter(fields, printer=_pprint):
    """
    Generate inner function with closured fields names and printer function

:param list fields: List of strings that represent names of fields in final output
    :param callable printer: Function to format and print data for every entry
    :rtype: callable
    :return: function to format data for every user
    """
    def wrapper(user):
        """
        :param string user: Find and format data for this User ID
        :rtype: list
        :return: List of given user's statistics data line or empty list
        """
        data = ''
        package = packages_users[user]["package"]
        reseller = packages_users[user]["reseller"]
        if reseller == '':
            reseller = None
        if ve_cfg_version <= 1:
            # Reseller's default limits will not be inherited by its end-users.
            # Backward compatibility - show some reseller packages in
            # paneluserslimits.
            prepare_setup_data(package, reseller=None)
        else:
            if reseller is not None:
                # We can set xml_config_load_elements=False
                # because get_XML_cfg was called with True before _user_formatter called
                _load_resellers_xml_data(reseller, xml_config_load_elements=False)
            else:
                # It's important to re-read admin's limits here or we will use
                # limits from previous reseller for next users in list
                # We can set xml_config_load_elements=False
                # because get_XML_cfg was called with True before _user_formatter called
                get_XML_cfg(load_config_elements=False)
            prepare_setup_data(package, reseller=reseller)
        data = copy.copy(setup_data)        # only after reading reseller's xml
        lve_apply(user, plan_id=package, reseller=reseller, result=True)

def check_changed(key):
            return '*' + str(setup_data[key]) if str(data[key]) != str(setup_data[key]) else str(data[key])

def convert_mem_limits(value):
            return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value)

data['id'] = str(user)
        data['cpu'] = str(check_changed('cpu'))
        data['ncpu'] = str(check_changed('ncpu'))
        data['pmem'] = str(convert_mem_limits(check_changed('pmem')))
        data['vmem'] = str(convert_mem_limits(check_changed('mem')))
        data['ep'] = str(check_changed('ep'))
        data['io'] = str(check_changed('io'))
        data['nproc'] = str(check_changed('nproc'))
        data['iops'] = str(check_changed('iops'))
        if JSON:
            data['package'] = _normalize_str(package)
        else:
            data['package'] = package
        if reseller is None:
            data['reseller'] = 'N/A' if JSON else ''
        else:
            data['reseller'] = reseller

if '*' in data['cpu']:
            data['cpu'] =   '*' + str(int(data['cpu'].lstrip('*')) // 100)
        else:
            data['cpu'] = int(data['cpu']) // 100
        data['speed'] = data['cpu']
        data['cpu'] = str(speed_to_old_cpu(data['speed']))

res = []
        if JSON:
            line = ','.join(f'"{f}":"{data[f.lower()]}"' for f in fields)
            res = [f'{{{line}}}']
        else:
            printer(*[data[f.lower()] for f in fields])
        return res
    return wrapper

# Show current user's limits for control panel
# 'lvectl paneluserslimits' or 'lvectl paneluserlimits lve_id'
def paneluserslimits(userid=None, reseller=None):
    get_XML_cfg()
    try:
        # create cache for userid_calls
        GetControlPanelUsers('list-users')
        # use explicit compare, because userid may be zero!
        # if userid == 0, then show only default limits
        # LU-374
        if userid is not None and userid:
            GetControlPanelUsers('userid', userid)
        # LU-530
        elif reseller is not None:
            GetControlPanelUsers('list-reseller-users', reseller=reseller)
        else:
            GetControlPanelUsers('list-users')
    except Exception:
        pass
    more_fields = ["PACKAGE"]
    result = _formatter(_pprint, more_fields=more_fields)
    fields = get_fields() + more_fields
    formatter = _user_formatter(fields)
    for user in packages_users:
        result += formatter(user)
    if JSON:
        print('{"data":[' + ','.join(result) + ']}')

def paneluserslist():
    # type: () -> List[Tuple[int, str, str]]
    """Get list of tuples[lve_id, reseller, package] from control panel"""
    GetControlPanelUsers('list-users')
    result = []
    for str_uid, payload in packages_users.items():
        result.append((int(str_uid), payload['reseller'], payload['package']))
    return result

def panelpackagesdict():
    # type: () -> Dict[str, List[str]]
    """Get dict of pairs[provider, list[package_name]] from control panel"""
    from clveconfig import DEFAULT_PROVIDER  # NOQA
    packages = {}

GetControlPanelUsers('list-packages')
    # admin's packages are already in bytes...
    packages[DEFAULT_PROVIDER] = list(packages_users.keys())
    GetControlPanelUsers('list-resellers-packages')
    # ..but we must convert reseller's package to bytes, because cl-summary
    # expects bytes and print warnings about unicode comparison
    packages.update(packages_users)

return packages

# lvectl all-user-list
def all_users_limits():
    """
    Implements lvectl all-user-list command

:return: None, prints result to stdout
    """
    get_XML_cfg()
    GetControlPanelUsers('list-users')

result = _formatter(_pprint_f, more_fields=["PACKAGE", "RESELLER"])
    fields = get_fields() + ["PACKAGE", "RESELLER"]

formatter = _user_formatter(fields, printer=_pprint_f)
    for user in packages_users:
        result += formatter(user)

if JSON:
        print('{"data":[' + ','.join(result) + ']}')

def _filtering_da_admins(ve_dict):
    """
    Filtering DirectAdmin's admins for `lvectl apply all` command

:param ve_dict: dict with LVE
    :return: filtering dict

"""

if cldetectlib.getCPName() == 'DirectAdmin':
        # get list of uids DirectAdmin's admins
        uids_da_admins = [pwd.getpwnam(user).pw_uid for user in admins()]
        ve_dict = {key: value for key, value in ve_dict.items() if key not in uids_da_admins}

return ve_dict

def prepare_apply_data(lvp_id=0):
    try:
        # update packages_users global dict
        GetControlPanelUsers()
        packages_users_ = dict(packages_users)
        if lvp_id:  # filter for apply lve top containers
            cfg_lvp_id_list = lve.map.name_map.id_list()
            packages_users_ = {k: v for k, v in packages_users.items() if k in cfg_lvp_id_list}
    except Exception:
        packages_users_ = {}

if lvp_id is True:
        node_list = ve_lvp
        id_list = lvp_list()
    else:
        node_list = ve_lve
        id_list = lve.proc.lve_id_list(lvp_id=lvp_id)

# get xml node for each lve_id
    # ve_dict is a local dict with lve_id and node with limit for lve_id
    # keys - int lve_id
    # data - xml node or None
    ve_dict = {}
    for node in node_list:
        ve_dict[get_ve_lve_user_uid(ve_lve_element=node)] = {'node' : node, 'reseller' : None}
    for lve_id in id_list:
        if lve_id not in ve_dict:
            # add lve_id for LVE that are not in ve.cfg
            ve_dict[lve_id] = {'node': None, 'reseller': None}

if (len(packages_users_) != 0):
        #  filtering addon admins DA.
        #  package_users contain only users, not addon admins DA
        ve_dict = _filtering_da_admins(ve_dict)
        for uid in packages_users_:
            if uid not in ve_dict:
                node = None
            else:
                node = ve_dict[uid]['node']
            # add lve_id for users that have package assigned
            pkg = packages_users_[uid]
            resellers = guess_reseller_by_package(pkg)
            if len(resellers) > 0:
                ve_dict[uid] = {'node' : node, 'reseller' : resellers[0]}
            else:
                ve_dict[uid] = {'node' : node, 'reseller' : None}
    return ve_dict

def lve_destroy_and_recreate_all():
    lve_ve_dict, lve_lvp_map = _get_lve_ve_dict_and_lvp_map()

remaning_alive_lves = set(lve.proc.lve_id_list())
    # NOTE: First we handle all LVEs not belonging to any LVP.
    for lve_id in lve_ve_dict.keys():
        if lve_lvp_map.get(lve_id, 0) != 0:
            # This code path will never be triggered if reseller limits are disabled
            remaning_alive_lves.discard(lve_id)
            continue
        if lve_id in remaning_alive_lves:
            lve.lve_destroy(lve_id)
            remaning_alive_lves.discard(lve_id)
        # TODO(vlebedev): Is this context manager really necessary here?
        with lve.py.context_ignore_error(ignore_error=True):
            lve_apply(
                lve_id,
                out_node=lve_ve_dict[lve_id]['node'],
                reseller=lve_ve_dict[lve_id]['reseller'],
            )
    # NOTE: Reset the default LVE settings.
    lve_apply(lve_id=0)

if lve.reseller_limit_supported():
        lvp_ve_dict = prepare_apply_data(True)
        remaining_alive_lvps = set(lve.proc.lvp_id_list())

# NOTE: Now we handle LVPs and LVEs inside those LVPs.
        for lvp_id in lvp_ve_dict.keys():
            if lvp_id in remaining_alive_lvps:
                pylve.lve_lvp_destroy(lvp_id)
                remaining_alive_lvps.discard(lvp_id)
            _create_if_necessary_and_configure_lvp(lvp_ve_dict, lvp_id)

# apply reseller's users
            reseller_name = lve.map.get_reseller_name(lvp_id)
            kernel_mapping = lve.proc.map()
            for user in clcommon.cpapi.reseller_users(reseller_name):
                lve_id = pwd.getpwnam(user).pw_uid
                if lve_id in remaning_alive_lves:
                    lve.lve_destroy(lve_id)
                    remaning_alive_lves.discard(lve_id)

if kernel_mapping.get(lve_id, 0) != lvp_id:
                    lve.py.lve_lvp_move(
                        lvp_id,
                        lve_id,
                        err_msg=f'Can`t move lve_id={lve_id} to lvp_id={lvp_id}; error code {{code}}'
                    )
                lve_apply(lve_id=lve_id, reseller=reseller_name)

# NOTE: Destroy any remaining live LVP that was not handled above.
        for lvp_id in remaining_alive_lvps:
            pylve.lve_lvp_destroy(lvp_id)

# TODO(vlebedev): Not sure if this cleanup (and the one for LVPs above) is
    #                 really necesary but at least it faithfully replicates the
    #                 behaviour of `lvectl destroy all` part.
    # NOTE: Destroy any remaining live LVE that was not handled above.
    for lve_id in remaning_alive_lves:
        lve.lve_destroy(lve_id)

# Apply all users and resellers settings
def lve_apply_all():
    ve_dict, lve_lvp_map = _get_lve_ve_dict_and_lvp_map()

with lve.py.context_ignore_error(ignore_error=True):
        for lve_id in ve_dict.keys():
            if lve_lvp_map.get(lve_id, 0) == 0:
                lve_apply(lve_id, out_node=ve_dict[lve_id]['node'], reseller=ve_dict[lve_id]['reseller'])

lve_apply(lve_id=0)

# apply limits for all LVP and LVEs inside LVP
    if not lve.reseller_limit_supported():
        return

ve_dict = prepare_apply_data(True)
    kernel_mapping = lve.proc.map()

for lvp_id_ in ve_dict.keys():
        _create_if_necessary_and_configure_lvp(ve_dict, lvp_id_)

# apply reseller's users
        reseller_name = lve.map.get_reseller_name(lvp_id_)
        for user in clcommon.cpapi.reseller_users(reseller_name):
            lve_id_ = pwd.getpwnam(user).pw_uid
            # LU-511: create mapping before lve if needed
            if kernel_mapping.get(lve_id_, 0) != lvp_id_:
                lve.py.lve_lvp_move(
                    lvp_id_,
                    lve_id_,
                    err_msg=f'Can`t move lve_id={lve_id_} to lvp_id={lvp_id_}; error code {{code}}'
                )
            lve_apply(lve_id=lve_id_, reseller=reseller_name)

def _get_lve_ve_dict_and_lvp_map() -> tuple[dict, dict]:
    get_XML_cfg()
    GetControlPanelUsers('list-users')
    lve_ve_dict = prepare_apply_data()
    lve_lvp_map = dict(lve.lve_id_lvp_id_pairs()) if lve.reseller_limit_supported() else {}
    return lve_ve_dict, lve_lvp_map

def _create_if_necessary_and_configure_lvp(ve_dict, lvp_id) -> None:
    # apply lvp limits using defaults for lvp
    # load reseller defaults instead of gloabal
    get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True, load_config_elements=False)
    lve_apply(lve_id=lvp_id,
              out_node=ve_dict[lvp_id]['node'],
              reseller=ve_dict[lvp_id]['reseller'],
              lvp_id=lvp_id)

get_XML_cfg(lvp_id=lvp_id, load_config_elements=False)
    lve_apply(lve_id=0, lvp_id=lvp_id)

def _remove_reseller(lvp_id):
    """Remove reseller from ve.cfg and from procfs."""
    get_global_lock(True)
    get_XML_cfg(lvp_id=lvp_id)
    for el in ve_lvp:
        if get_ve_lve_user_uid(ve_lve_element=el) == lvp_id:
            users = lve.proc.map_lve_id_list(lvp_id)
            # move containers to host
            for lve_id in users[:]:
                pylve.lve_lvp_move(0, lve_id)
                try:
                    pwd.getpwuid(lve_id)
                except KeyError:
                    if lve.py.lve_exists(lve_id):
                        lve.lve_destroy(lve_id)
                    users.remove(lve_id)
            lvp_destroy(lvp_id)  # destroy container
            el.parentNode.removeChild(el)  # remove record
            save_xml(ve_cfg)
            # load defaults host settings for end users
            get_XML_cfg(lvp_id=0)
            # apply limits from config (including package limits)
            for lve_id in users:
                lve_apply(lve_id)
            get_XML_cfg(lvp_id=lvp_id)
            return True
    return False

def disable_reseller_limits(reseller_name, lvp_id):
    """Disable reseller limits and call hooks"""
    if _remove_reseller(lvp_id):
        reseller_limits_disabled_post.throw_event(reseller=reseller_name)
    else:
        if JSON:
            json_format('multi', ['WARNING', f'no configuration found for LVP {lvp_id}'])
            sys.exit(-1)
        else:
            print(f'warning: no configuration found for LVP {lvp_id}')

# Delete User from ve.cfg and set default lve settings
def lve_delete(lve_id):
    get_global_lock(True)
    get_XML_cfg()
    Deleted = False
    for el in ve_lve:
        if get_ve_lve_user_uid(ve_lve_element=el) == lve_id:
            Deleted = True
            lve_destroy(lve_id)
            lve_create(lve_id)
            el.parentNode.removeChild(el)
            save_xml(ve_cfg)
            get_XML_cfg()
            lve_apply(lve_id)
    if not Deleted:
        if JSON:
            json_format('multi', ['WARNING', f'no configuration found for VE {lve_id}'])
            sys.exit(-1)
        else:
            print(f'warning: no configuration found for VE {lve_id}')

def lve_enter_check():
    if not os.path.exists('/proc/lve/enter'):
        if JSON:
            json_format('multi', ['WARNING', 'enter by name not supported'])
        else:
            print('warning: enter by name not supported')
        sys.exit(-1)

def enter_apply(sign, binary):
    lve_enter_check()
    try:
        msg = sign + binary.strip()
        with open('/proc/lve/enter', 'w', encoding='utf-8') as f:
            f.write(msg)
    except Exception:
        pass

def list_binaries():
    get_XML_cfg()
    if JSON:
        result = '{"data":['
        first = True
        for el in ve_binary:
            path = el.getAttribute('path')
            if first:
                result += '"' + path + '"'
                first = False
            else:
                result += ',"' + path + '"'
        result += ']}'
        print(result)
    else:
        print("Binaries")
        for el in ve_binary:
            print(el.getAttribute('path'))

def load_binaries():
    get_XML_cfg()
    for el in ve_binary:
        enter_apply('+', el.getAttribute('path'))

def reload_binaries():
    lve_enter_check()
    with open('/proc/lve/enter', 'r', encoding='utf-8') as f:
        for line in f:
            enter_apply('-', line)
    load_binaries()

def del_binary(binary):
    global ve_binary
    get_global_lock(True)
    lve_enter_check()
    get_XML_cfg()
    deleted = False
    for el in ve_binary:
        if el.getAttribute('path') == binary:
            deleted = True
            enter_apply('-', binary)
            el.parentNode.removeChild(el)
            save_xml(ve_cfg)
            get_XML_cfg()
    if not deleted:
        if JSON:
            json_format('multi', ['WARNING', f'no configuration found for {binary}'])
        else:
            print(f'warning: no configuration found for {binary}')
        sys.exit(-1)

def set_binary(binary):
    global ve_binary
    global ve_enter_by_name
    get_global_lock(True)
    get_XML_cfg()
    for el in ve_binary:
        if el.getAttribute('path') == binary:
            return  # nothing to do, it is already there
    enter_apply('+', binary)
    bin_xml = ve_cfg.createElement('binary')
    bin_xml.setAttribute('path', binary)
    ve_enter_by_name.appendChild(bin_xml)
    save_xml(ve_cfg)
    get_XML_cfg()

def lve_set_default(set_data, package_flag, is_needed, lvp_id=0):
    """
    Set given lve or package to default values for given parameters

:param dict set_data: Arguments of lvectl call
    :param bool package_flag: Should we delete package or lve with given id
    :param callable is_needed: Function that takes xml element and set_data dict and returns
                               whether current xml element contains info about needed ID from set_data
    """
    try:
        if package_flag:
            data = ve_package
        elif lvp_id:
            data = ve_lvp
        else:
            data = ve_lve
        el = [e for e in data if is_needed(e, set_data)][0]
    except IndexError:
        return
    if lvp_id:
        # for lvectl set-reseller {id} --default=A,B,C; remove limit record in ve.cfg
        for tag_ in set_data['set-default']:
            if tag_ == 'ep':
                n = xml_filter_first(el, 'other', 'maxentryprocs')
            else:
                n = xml_filter_first(el, tag_, 'limit')
            if n:
                n.parentNode.removeChild(n)
        return
    to_keep = set(LIMITS_LIST_NAME) - set_data['set-default']
    for limit in to_keep:
        if limit == 'ep' and len(el.getElementsByTagName('other')) > 0:
            # dict.setdefault isn't lazy evaluated
            if limit not in set_data:
                set_data[limit] = el.getElementsByTagName('other')[0].getAttribute('maxentryprocs')
        elif len(el.getElementsByTagName(limit)) > 0:
            # dict.setdefault isn't lazy evaluated
            if limit not in set_data:
                set_data[limit] = el.getElementsByTagName(limit)[0].getAttribute('limit')
    # delete this lve or package
    if package_flag:
        plan_delete(set_data['ve_id'])
    else:
        lve_delete(set_data['ve_id'])

def _check_reseller_user_pair(uid, reseller_name):
    """
    Checks is uid owned by reseller
    :param uid: uid for check
    :param reseller_name: Reseller name, None treats as root
    :return: True - valid reseller/user pair, False - else
        Special case:
          if reseller_name is None (root) - always valid
    """
    if reseller_name in (None, 'root'):
        return True
    # reseller is not root
    # determine username
    username = get_main_username_by_uid(uid)
    if username in ('root', 'N/A'):
        # user is root or no such user -- error
        return False
    # determine users of supplied reseller's container.
    try:
        # Get reseller's users list
        reseller_users_list = reseller_users(reseller_name)
    except Exception:
        # any error - ignore, reseller is root
        reseller_users_list = []
    if username in reseller_users_list:
        return True
    return False

# Set limits for user
# TODO: split this method into several independent:
# - enable_reseller_limits
# - set_reseller_default_limits
# - set_reseller_limits
# - set_lve_limits
def lve_set(set_data, lvp_id=0):
    # set_data example:
    #  {'iops': 2222, 'reseller_name': 'res', 'save': False, 've_id': 1023, 'pmem': 524288}
    # 524288 * 4096 = 2G -- pmem=2G
    if lvp_id == 0:
        # Set limits for user's LVE, check reseller/user match
        reseller_name = set_data.get('reseller_name', None)
        lve_id = set_data['ve_id']
        if not _check_reseller_user_pair(lve_id, reseller_name):
            return False
    global setup_data
    get_global_lock(True)
    if lvp_id and lvp_id == set_data['ve_id']:
        # reseller's container limits
        if lve.proc.exist_lvp(lvp_id):
            # reseller's container exists... load info about his container
            get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True)
        else:
            # reseller's container does not exists... create new one with default limits
            get_XML_cfg(lvp_id=lvp_id)
    elif lvp_id == 0 and lve.reseller_limit_supported():
        # user's limits (reseller & not)
        get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id']))
    else:
        # default limits and limits for user when reseller's does not supported
        get_XML_cfg(lvp_id=lvp_id)

try:
        GetControlPanelUsers()
    except Exception:
        pass
    try:
        # LU-366. Fix reset user's limits to unlimited if user in reseller package
        # and reseller is not root/admin
        package = packages_users[set_data['ve_id']]
        resellers = guess_reseller_by_package(package)
        reseller = resellers[0] if resellers else ''
        prepare_setup_data(package, reseller=reseller)
    except Exception:
        setup_data = ve_defaults
    if set_data['ve_id'] != 0:
        has_ve = False
        # set default

def is_needed_user(el, set_data):
            return get_ve_lve_user_uid(ve_lve_element=el) == set_data['ve_id']

if 'set-default' in set_data:
            lve_set_default(set_data, package_flag=False, is_needed=is_needed_user, lvp_id=lvp_id)
        if 'ncpu' in set_data:
            lncpu = int(set_data['ncpu'])
        else:
            lncpu = ve_defaults['ncpu']

# check that cpu value in any format (cpu, speed=% or speed=[m|g]hz) is equal or not
        cpu_is_different = True
        if 'cpu' in set_data:
            setted_cpu = convert_to_kernel_format(set_data['cpu'], lncpu = lncpu)
            if setted_cpu == setup_data['cpu']:
                cpu_is_different = False

if lvp_id:
            el_list = ve_lvp
        else:
            el_list = ve_lve  # choose top level container for modifications
        for el in el_list:
            if is_needed_user(el, set_data):
                for key in LIMITS_LIST_NAME:
                    if key in set_data:
                        try:
                            if key == 'ep':
                                set_child_tag_atrr(el, 'other', 'maxentryprocs', set_data[key])
                            else:
                                set_child_tag_atrr(el, key, 'limit', set_data[key])
                        except (ValueError, IndexError, TypeError):
                            # we already checked cpu value, so use cpu_is_different result
                            if key == "cpu":
                                is_different = cpu_is_different
                            # otherwise compare with default in usual way
                            else:
                                is_different = setup_data[key] != set_data[key]
                            if is_different or set_data['save']:
                                if key == 'ep':
                                    node = ve_cfg.createElement('other')
                                    node.setAttribute('maxentryprocs',str(set_data[key]))
                                else:
                                    node = ve_cfg.createElement(key)
                                    node.setAttribute('limit',str(set_data[key]))
                                el.appendChild(node)

if not set_data.get('skip-update-cfg', False):
                    save_xml(ve_cfg)

has_ve = True
                if lvp_id and lvp_id == set_data['ve_id']:
                    # reseller's container limits
                    if lve.proc.exist_lvp(lvp_id):
                        # reseller's container does not exists... create new one with default limits
                        get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True)
                    else:
                        # reseller's container exists... load info about his container
                        get_XML_cfg(lvp_id=lvp_id)
                elif lvp_id == 0 and lve.reseller_limit_supported():
                    # user's limits (reseller & not)
                    get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id']))
                else:
                    # default limits and limits for user when reseller's does not supported
                    get_XML_cfg(lvp_id=lvp_id)
                lve_apply(set_data['ve_id'], lvp_id=lvp_id)
            else:
                pass

if not has_ve and set_data['ve_id']:
            el_name = LVP_XML_TAG_NAME if lvp_id else 'lve'
            el = ve_cfg.createElement(el_name)
            if lvp_id:  # for create resellers limit config
                # set reseller_name reseller_id map
                el.setAttribute('id', str(lvp_id))
                el.setAttribute('user', set_data['user'])
                el.appendChild(ve_default)  # copy default limits to reseller
            else:
                if set_data.get('save-username'):
                    el.setAttribute('user', pwd.getpwuid(set_data['ve_id']).pw_name)
                else:
                    el.setAttribute('id', str(set_data['ve_id']))

for key in LIMITS_LIST_NAME:
                if key in set_data:
                    # we already checked cpu value, so use cpu_is_different result
                    if key == "cpu":
                        is_different = cpu_is_different
                    # otherwise compare with default in usual way
                    else:
                        is_different = setup_data[key] != set_data[key]
                    if is_different or set_data['save']:
                        if key == 'ep':
                            node = ve_cfg.createElement('other')
                            node.setAttribute('maxentryprocs',str(set_data[key]))
                        else:
                            node = ve_cfg.createElement(key)
                            node.setAttribute('limit',str(set_data[key]))
                        el.appendChild(node)

added = False
            for el2 in ve_package:
                el2.parentNode.insertBefore(el,el2)
                added = True
                break
            if not added:
                ve_cfg.lastChild.appendChild(el)

if not set_data.get('skip-update-cfg', False):
                save_xml(ve_cfg)

if lvp_id:
                enables_reseller_limits = not lve.proc.exist_lvp(lvp_id)
                # load lvp defaults and lvp tag, set limits for reseller
                get_XML_cfg(lvp_id=lvp_id, lvp_defaults=True)
                lve_apply(set_data['ve_id'], lvp_id=lvp_id)
                # copy default limits from host container
                pylve.lve_set_default(lvp_id, pylve.lve_info(0))
                # load lvp tag and reseller's end user defaults
                get_XML_cfg(lvp_id=lvp_id)
                reseller_name = lve.map.get_reseller_name(lvp_id)
                for lve_id_ in lve.map.lvp_lve_id_list(lvp_id=lvp_id):
                    lve.py.lve_lvp_move(lvp_id, lve_id_)
                    lve_apply(lve_id_, reseller=reseller_name)
                # call hook if we enabled reseller limits
                if enables_reseller_limits:
                    reseller_limits_enabled_post.throw_event(reseller=reseller_name)
            else:
                if lve.reseller_limit_supported():
                    get_XML_cfg(lvp_id=lve.proc.detect_inside_lvp(set_data['ve_id']))
                else:
                    get_XML_cfg(lvp_id=lvp_id)
                lve_apply(set_data['ve_id'])
    else:
        for key in LIMITS_LIST_NAME:
            if key in set_data:
                if key == 'ep':
                    ve_default.getElementsByTagName('other')[0].setAttribute('maxentryprocs',str(set_data[key]))
                else:
                    ve_default.getElementsByTagName(key)[0].setAttribute('limit',str(set_data[key]))

if not set_data.get('skip-update-cfg', False):
            save_xml(ve_cfg)

get_XML_cfg(lvp_id=lvp_id)
        lve_apply(set_data['ve_id'], lvp_id=lvp_id)
    return True

def package_set(set_data, is_reseller=False):
    """
    Set package with some heuristic algorithm to simulate old package set behavior
    """
    get_global_lock(True)
    get_XML_cfg()
    # Removed in LU-351
    # set_data['ve_id'] = unicode(set_data['ve_id'].decode('utf-8'))
    reseller_list = guess_reseller_by_package(set_data['ve_id'])
    if len(reseller_list) == 0:
        reseller = None
    elif len(reseller_list) >= 1:
        # if dublicated packages found - use first as a reseller
        reseller = reseller_list[0]
    #if ve.cfg has tag <version>2</version> - trying to guess reseller name by package
    #if reseller is undef or ve.cfg has not tag version - work with ver 1
    if reseller is not None and ve_cfg_version > 1:
        set_data['reseller_name'] = reseller
        package_set_ext(set_data, is_reseller=True)
    else:
        package_set_ext(set_data, is_reseller=False)

# Set new package or modify exist
# package-set-ext
def package_set_ext(set_data, is_reseller=False):
    get_global_lock(True)
    get_XML_cfg()
    has_package = False

if is_reseller:
        def is_needed_plan(el, set_data):
            return (
                el.getAttribute('id') == set_data['ve_id'] and el.getAttribute('reseller') == set_data['reseller_name']
            )
    else:
        def is_needed_plan(el, set_data):
            return el.getAttribute('id') == set_data['ve_id'] and not el.getAttribute('reseller')

if 'set-default' in set_data:
        lve_set_default(set_data, package_flag=True, is_needed=is_needed_plan)

for el in ve_package:
        if is_needed_plan(el, set_data):
            for key in LIMITS_LIST_NAME:
                if key in set_data:
                    try:
                        if key == 'ep':
                            el.getElementsByTagName('other')[0].setAttribute('maxentryprocs',str(set_data[key]))
                        else:
                            el.getElementsByTagName(key)[0].setAttribute('limit',str(set_data[key]))
                    except (ValueError, IndexError, TypeError):
                        if key == 'ep':
                            node = ve_cfg.createElement('other')
                            node.setAttribute('maxentryprocs',str(set_data[key]))
                        else:
                            node = ve_cfg.createElement(key)
                            node.setAttribute('limit',str(set_data[key]))
                        el.appendChild(node)
            if cldetectlib.is_plesk():
                plesk_id = _plesk_get_package_id(set_data.get('reseller_name', ''), set_data['ve_id'])
                el.setAttribute(XML_PLESK_ID, str(plesk_id))
            has_package = True
    if not has_package:
        package_reseller = ''
        el = ve_cfg.createElement('package')
        el.setAttribute('id', set_data['ve_id'])
        if is_reseller:
            el.setAttribute('reseller', set_data['reseller_name'])
            package_reseller = set_data['reseller_name']
        if cldetectlib.is_plesk():
            plesk_id = _plesk_get_package_id(package_reseller, set_data['ve_id'])
            el.setAttribute(XML_PLESK_ID, str(plesk_id))
        for key in LIMITS_LIST_NAME:
            if key in set_data:
                if key == 'ep':
                    node = ve_cfg.createElement('other')
                    node.setAttribute('maxentryprocs',str(set_data[key]))
                else:
                    node = ve_cfg.createElement(key)
                    node.setAttribute('limit',str(set_data[key]))
                el.appendChild(node)
        ve_cfg.lastChild.appendChild(el)

save_xml(ve_cfg)
    get_XML_cfg()
    copy_package_settings_to_cpanel(set_data)

if 'ncpu' in set_data:
        lncpu = int(set_data['ncpu'])
    else:
        lncpu = ve_defaults['ncpu']

if 'cpu' in set_data:
        set_data['cpu'] = convert_to_kernel_format(set_data['cpu'], lncpu=lncpu)

reseller = set_data['reseller_name'] if is_reseller else None
    plan_apply(set_data['ve_id'], reseller=reseller)

def _plesk_get_package_id(reseller: str, package: str) -> Optional[int]:
    """
    Find the right package id from plesk DB query
    """
    panel = detect_panelclass()
    packages = panel.list_domain_packages_with_id()
    try:
        pack = next(filter(
            lambda x: x[0] in {reseller, 'root'} and x[1] == package,   # no reseller == reseller is root (admin)
            packages
        ))
        return pack[2]
    except StopIteration:
        return None

def get_reseller_packages_map():
    """
    Retrives resellers to packages map from panel using /usr/bin/getcontrolpaneluserspackages
    :return: Dictionary:
    { 'reseller1' -> ['pack1', 'pack2'], 'reseller2' -> ['pack'] }
    """
    global packages_users
    packages_users_copy = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages_map = packages_users
    packages_users = packages_users_copy
    return reseller_packages_map

def reseller_package_set(set_data):
    """
    Set reseller package limits
    :param set_data: input data dictionary
    :return: True - limits was set succesfully
            False - supplied provider has no supplied package
    """
    # set limits to package that belongs to given reseller
    reseller_name = set_data['reseller_name']
    package_name = set_data['ve_id']
    # Retrive resellers packages from panel
    reseller_packages_map = get_reseller_packages_map()
    # If reseller has supplied package -- set limit
    if reseller_name in reseller_packages_map and package_name in reseller_packages_map[reseller_name]:
        # Reseller/package pair valid - set limit
        package_set_ext(set_data, is_reseller=True)
        return True
    # ERROR: Supplied reseller has no supplied package
    return False

def copy_package_settings_to_cpanel(set_data):
    """
    Copy package limits from ve.cfg to cpanel packages data
    """
    package = set_data['ve_id']
    if not cldetectlib.is_cpanel():
        return  # skip func if panel not cPanel
    package_path = f'/var/cpanel/packages/{package}'
    if not os.path.isfile(package_path):
        return  # skip func if no cPanel packages found
    with open(package_path, 'r', encoding='utf-8') as f:
        cpanel_package_data = f.readlines()
    new_cpanel_package_data = cpanel_package_data[:]
    old_cpanel_data = {}

# proces old_cpanel_package_data - get old limits and remove stings from it
    # result of processing - cpanel_package_data_modify and old_cpanel_data
    for line in cpanel_package_data:
        if line.startswith('lve_'):
            line_parts = line.strip().split('=')
            limit_name = line_parts[0].replace('lve_', '').strip()
            if line_parts[1] != 'DEFAULT':
                old_cpanel_data[limit_name] = line_parts[1]  # get old_limits
            if limit_name in LIMITS_LIST_NAME:
                new_cpanel_package_data.remove(line)
        if line.startswith('_PACKAGE_EXTENSIONS') and 'lve' not in line:
            return      # skip func if no lve extention install to the package

for limit_name in ('pmem', 'mem', 'vmem'):
        if limit_name in old_cpanel_data:
            memory_page_value = clcommon.memory_to_page(old_cpanel_data[limit_name])
            old_cpanel_data[limit_name] = memory_page_value or ve_defaults[limit_name]

if is_limits_equals(old_cpanel_data, set_data):
        return              # skip writeting to file - limits are equals

# create and add to new cpanel_data_file limits lines like:
    # lve_ + limit_name + = + limit_value
    cpanel_data = create_cpanel_limits(package, ve_package)
    for limit_name in LIMITS_LIST_NAME:
        limit_value = cpanel_data[limit_name]
        limit_line = f'lve_{limit_name}={limit_value}\n'
        new_cpanel_package_data.append(limit_line)
    write_file_via_tempfile(''.join(new_cpanel_package_data), package_path, 0o644)

def is_limits_equals(old_limits, new_limits):
    """
    check if new set of limits for package are equals to used
    """
    for key in new_limits.keys():
        if key in ('ve_id', 'save'):
            continue    # ve_id == package name. skip this key
        try:
            if old_limits[key] != new_limits[key]:
                return False
        except KeyError:
            return False
    return True

def create_cpanel_limits(package_id, xml_packages):
    """
    create limits for cpanel package file
    use data from ve.cfg:
    limit = limit if found in ve.cfg or DEFAULT
    return dict
    """
    result_data = {}
    for el in xml_packages:
        if el.getAttribute('id') == package_id:
            for limit in LIMITS_LIST_NAME:
                try:
                    if limit == 'ep':
                        result_data[limit] = str(
                            el.getElementsByTagName('other')[0].getAttribute('maxentryprocs')
                        ).strip()
                    elif limit in ("mem", "vmem", "pmem"):
                        result_data[limit] = str(
                            clcommon.page_to_memory(
                                int(el.getElementsByTagName(limit)[0].getAttribute("limit"))
                            )
                        ).strip()
                    else:
                        result_data[limit] = str(el.getElementsByTagName(limit)[0].getAttribute('limit')).strip()
                except (ValueError, IndexError, TypeError):
                    result_data[limit] = 'DEFAULT'
    return result_data

# Delete plan from ve.cfg
def plan_delete(plan_id, reseller_name=None):
    get_global_lock(True)
    get_XML_cfg()
    Deleted = False
    if reseller_name is None:
        def is_needed_package(el):
            return el.getAttribute('id') == plan_id and not el.getAttribute('reseller')
    else:
        def is_needed_package(el):
            return el.getAttribute('id') == plan_id and el.getAttribute('reseller') == reseller_name

for el in ve_package:
        if is_needed_package(el):
            Deleted = True
            el.parentNode.removeChild(el)
            save_xml(ve_cfg)
            break
    if not Deleted:
        # try to guess reseller name only if no reseller name
        if reseller_name is None:
            resellers_list = guess_reseller_by_package(plan_id)
            if len(resellers_list) == 0:
                reseller = None
            else:
                # if some resellers found - use first
                reseller = resellers_list[0]
            # try to delete package only if we guess reseller
            if reseller is not None:
                plan_delete(plan_id, reseller)
                return
        if JSON:
            json_format(
                'multi',
                ['WARNING', f'no configuration found for plan {plan_id}']
            )
            sys.exit(-1)
        else:
            print(f'warning: no configuration found for plan {plan_id}')
    lve_apply_all()

def reseller_plan_delete(plan_id, reseller_name):
    plan_delete(plan_id, reseller_name=reseller_name)

def get_xml_limit(el, key):
    try:
        return str(el.getElementsByTagName(key)[0].getAttribute('limit'))
    except (ValueError, IndexError):
        #  convert from kernel format to output
        return ve_defaults[key] if key != 'cpu' else f'{ve_defaults[key] // 100}%'

def _normalize_str(data_str):
    """
    Normalize string for JSON output.
    Example:
     - Input string: -_&[{}]'"`te\\s/t\a
     - Output string: -_&[{}]'\"`te\\\\s/t\\a
    :param data_str: String for normalize
    :return: Normalied string
    """
    def _get_char_index(input_string, char_to_search, ordinal):
        """
        Get the index of the specified occurrence of character in string
        :param input_string: String
        :param char_to_search: Character to search
        :param ordinal: Required occurence number
        :return: Char index
        """
        count = 0
        for idx, ch in enumerate(input_string):
            if ch == char_to_search:
                # Char found
                count += 1
                if count == ordinal:
                    return idx
        # Char not found
        return -1

if data_str is None:
        return None
    json_str = json.dumps({'str': data_str})
    # json_str example: {"str": "-_&[{}]'\"`te\\\\s/t\\a"},
    # get '-_&[{}]'\"`te\\\\s/t\\a' from it
    # Get third " index
    trd_idx = _get_char_index(json_str, '"', 3)
    # Get Last " index
    last_idx = json_str.rfind('"')
    return json_str[trd_idx + 1:last_idx]

def _package_formatter(fields, is_reseller=False, printer=None):
    """
    Generate inner function with closured fields names, is_reseller flag and printer function

:param list fields: List of strings that represent names of fields in final output
    :param boolean is_reseller: Format output with info about reseller or not
    :param callable printer: Function to format and print data for every entry
    :rtype: callable
    :return: function to format data for every user
    """

printer = printer if printer is not None else _pprint_r if is_reseller else _pprint_p

def wrapper(package_name, reseller_name=None):
        """
        :param string package_name: Find and format data for this package name
        :param string reseller_name: reseller name, owner of supplied package
        :rtype: list
        :return: List of giver package's statistics data line or empty list
        """
        # in order to avoid unicode warnings here
        if is_reseller:
            # Reseller package
            _load_resellers_xml_data(reseller_name)
            data = copy.copy(ve_defaults)   # only after reading reseller's xml
            data['reseller'] = reseller_name

def is_needed_package(el):
                return package_name == el.getAttribute('id') and el.getAttribute('reseller') == reseller_name
        else:
            # Admin's package
            def is_needed_package(el):
                return package_name == el.getAttribute('id') and not el.getAttribute('reseller')

data = copy.copy(ve_defaults)

data['id'] = _normalize_str(package_name) if JSON else package_name

def convert_mem_limits(value):
            return _pmem_vmem_to_bytes_value(value) if BYTES_FLAG else _mb_mem(value)

data['vmem'] = convert_mem_limits(data['mem'])
        if LVE_VERSION > 4:
            data['pmem'] = convert_mem_limits(data['pmem'])
        for el in ve_package:
            if is_needed_package(el):
                lncpu = get_xml_limit(el, 'ncpu')
                data['ncpu'] = lncpu if lncpu != '' else str(ve_defaults['ncpu'])
                data['speed'] = str(convert_to_kernel_format(
                    get_xml_limit(el, 'cpu'), lncpu=int(data['ncpu'])) // 100
                )
                try:
                    data['ep'] = str(int(
                        el.getElementsByTagName('other')[0].getAttribute('maxentryprocs')
                    ))
                except (IndexError, ValueError):
                    pass
                data['pmem'] = convert_mem_limits(get_xml_limit(el, 'pmem'))
                data['vmem'] = convert_mem_limits(get_xml_limit(el, 'mem'))
                data['io'] = get_xml_limit(el, 'io')
                data['nproc'] = get_xml_limit(el, 'nproc')
                data['iops'] = get_xml_limit(el, 'iops')
        if data.get('speed') is None:
            #  convert from kernel format for output
            data['speed'] = data['cpu'] // 100
        data['cpu'] = str(speed_to_old_cpu(data['speed']))

res = []
        if JSON:
            line = ','.join(f'"{f}":"{data.get(f.lower(), "N/A")}"' for f in fields)
            res = [f'{{{line}}}']
        else:
            printer(*[data.get(f.lower(), '') for f in fields])
        return res
    return wrapper

# lvectl package-list
def get_packages_list():
    get_XML_cfg()
    GetControlPanelUsers('list-packages')
    packages = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages = packages_users.copy()

result = _formatter(_pprint_p, default_id=DEFAULT_PACKAGE)
    formatter = _package_formatter(get_fields(), is_reseller=False, printer=_pprint_p)
    for package in packages:
        result += formatter(package)
    if ve_cfg_version > 1:
        formatter = _package_formatter(get_fields(), is_reseller=True, printer=_pprint_p)
    # reseller_packages: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in reseller_packages.items():
        # On DA skip all non-admin's packages
        if cldetectlib.is_da() and reseller_name != 'admin':
            continue
        for reseller_package in packages_list:
            if ve_cfg_version > 1:
                result += formatter(reseller_package, reseller_name)
            else:
                result += formatter(reseller_package)

if JSON:
        print('{"data":[' + ','.join(result) + ']}')

# lvectl reseller-package-list
def get_resellers_packages_list():
    get_XML_cfg()
    GetControlPanelUsers('list-resellers-packages')

more_fields = ["RESELLER"]
    fields = get_fields() + more_fields
    result = _formatter(_pprint_p, default_id=DEFAULT_PACKAGE)

formatter = _package_formatter(fields, is_reseller=True)
    # packages_users: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in packages_users.items():
        for reseller_package in packages_list:
            result += formatter(reseller_package, reseller_name)

if JSON:
        print('{"data":[' + ','.join(result) + ']}')

# lvectl all-package-list
def get_all_packages_list():
    get_XML_cfg()
    GetControlPanelUsers('list-packages')
    packages = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages = packages_users.copy()

more_fields = ["RESELLER"]
    fields = get_fields() + more_fields
    # make header with default package
    result = _formatter(_pprint_r, default_id=DEFAULT_PACKAGE, more_fields=more_fields)

formatter = _package_formatter(fields, is_reseller=False, printer=_pprint_r)
    for package in packages:
        result += formatter(package)
    # Print resellers packages
    formatter = _package_formatter(fields, is_reseller=True)
    # reseller_packages: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in reseller_packages.items():
        for reseller_package in packages_list:
            result += formatter(reseller_package, reseller_name)

if JSON:
        print('{"data": [' + ','.join(result) + ']}')

cached_resellers_packages = None
cached_list_packages      = None
cached_users              = None
cached_reseller_users     = None
cached_default            = None

def _convert_packages_list(package_list):
    """
    Converts package list to internal format
    :param package_list: Package list. Example: ['BusinessPackage', 'Package2']
    :return: Package list as dictionary. Example: {'BusinessPackage': 'BusinessPackage', 'Package2': 'Package2'}
    """
    packages_users_dict = {}
    for package in package_list:
        packages_users_dict[package] = package
    return packages_users_dict

# Get users from control panel with plans
def GetControlPanelUsers(option='list-all', lve_package_id='', reseller=None):
    """
    Parse output from GET_CP_PACKAGE_SCRIPT and get package and lve relations

:param option: option for GET_CP_PACKAGE_SCRIPT.
     Option is one from the following possible values:  'userid', 'package', 'list-packages', 'list-resellers-packages'
    :type option: string
    :param lve_package_id: lve_id or package_name
    :type lve_package_id: string or int
    :param reseller:
    :type reseller: string
    """
    global cached_list_packages
    global cached_resellers_packages
    global cached_users
    global cached_reseller_users
    global cached_default
    global packages_users
    # Check arguments
    if option not in ('list-all', 'userid', 'package', 'list-packages',
                      'list-resellers-packages', 'list-users', 'list-reseller-users'):
        return False
    if option in ('userid', 'package') and lve_package_id == '':
        return False

from clcommon.cpapi import (  # pylint: disable=import-outside-toplevel
        admin_packages,
        get_reseller_users,
        get_uids_list_by_package,
        list_all,
        list_users,
        reseller_package_by_uid,
        resellers_packages,
    )

try:
        if option == 'userid':
            if cached_users is not None:
                try:
                    packages_users = {
                        lve_package_id: {
                            'package': cached_users[lve_package_id]['package'],
                            'reseller': cached_users[lve_package_id]['reseller']
                        }
                    }
                except KeyError:
                    packages_users = {lve_package_id: {'package': '', 'reseller': ''}}
            else:
                try:
                    reseller_name, package = reseller_package_by_uid(lve_package_id)
                except ValueError:
                    # this is possible on vm without control panel
                    reseller_name = package = ''
                packages_users = {lve_package_id: {'package': package, 'reseller': reseller_name}}
                return True
        elif option == 'package':
            # reseller - optional argument
            packages_users = {lve_package_id: get_uids_list_by_package(lve_package_id, reseller)}
            return True
        elif option == 'list-packages':
            # Result format:
            # {'BusinessPackage': 'BusinessPackage', 'Package2': 'Package2'}
            if cached_list_packages is None:
                package_list = admin_packages()
                # Convert to output format
                packages_users = _convert_packages_list(package_list)
                cached_list_packages = packages_users
            else:
                # list-packages data already present
                packages_users = cached_list_packages
            return True
        elif option == 'list-resellers-packages':
            # Result format:
            #  {'res2 SimplePackage': 'res2 Package',
            #   'res1 BusinessPackage': 'res1 UltraPackage'}
            if cached_resellers_packages is None:
                # in order to produce same results as code that works with Popen
                packages_users = resellers_packages()
                cached_resellers_packages = packages_users
            else:
                # list-resellers-packages data already present
                packages_users = cached_resellers_packages
            return True
        elif option == 'list-users':
            # Result format:
            # {1000: {'reseller': '', 'package': 'Package1'},
            #  1001: {'reseller': '', 'package': 'BusinessPackage'},
            # }
            if cached_users is None:
                cached_users = list_users()
            packages_users = cached_users
            return True
        elif option == 'list-reseller-users':
            # {1001: {'reseller': 'res1', 'package': 'BusinessPackage'},
            #  1004: {'reseller': 'res1', 'package': 'BusinessPackage'}}
            if cached_reseller_users is None:
                reseller_users_dict = get_reseller_users(reseller)
                # for uid, user_data in reseller_users_dict.iteritems():
                #     packages_users[uid] = user_data
                cached_reseller_users = reseller_users_dict
                packages_users = reseller_users_dict
            else:
                # list-reseller-users data already present
                packages_users = cached_reseller_users
            return True
        elif option == 'list-all':  # deprecated. TODO: Remove this option
            # Result format:
            # {1000: 'Package1', 1001: 'BusinessPackage'}
            if cached_default is None:
                cached_default = list_all()
            packages_users = cached_default
            return True
    except EncodingError as e:
        raise_cpanel_encoding_error(e)
    except OSError:
        pass
    return False

def get_panel_users_count():
    """
    Retrieves panel users count
    :return:
    """
    GetControlPanelUsers()
    return len(packages_users)

# Apply plan settings for users
def plan_apply(plan_id, reseller=None):
    # fill the cache to speedup `lvectl package-set-ext` with many users in one package
    GetControlPanelUsers("list-users")

if GetControlPanelUsers("package", plan_id, reseller=reseller):
        for uid in packages_users[plan_id]:
            lve_apply(int(uid), plan_id, reseller=reseller)

# Destroy many LVEs from stdin
def destroy_many(users_list):
    for line in users_list:
        line = line.replace('\n','')
        users = line.strip().split()
        for user in users:
            if (len(user) != 0):
                try:
                    user = int(user)
                    lve_destroy(user)
                except Exception:
                    pass

# Apply many LVEs from stdin
def apply_many(users_list):
    get_XML_cfg()
    try:
        GetControlPanelUsers()
    except Exception:
        pass

for line in users_list:
        line = line.replace('\n','')
        users = line.strip().split()
        for user in users:
            if (len(user) != 0):
                try:
                    user = int(user)
                    lve_apply(user)
                except Exception:
                    pass

# Put pid into LVE
def limit_pid(lve_id, pid, flags):
    pylve.lve_enter_pid_flags(
        int(lve_id), int(pid), flags,
        err_msg=f'Can`t put proccess with pid {pid} in lve {lve_id}; error code {{code}}'
    )

# Get pid from LVE
def release_pid(pid):
    pylve.lve_leave_pid(int(pid), err_msg=f'Can`t release process with pid {pid}')

def get_globals():
    global ve_cfg
    global ve_lveconfig
    global ve_default
    global ve_lve
    global ve_defaults
    global ve_package
    global ve_binary
    global ve_enter_by_name
    global ubc

return {'ve_cfg': ve_cfg, 've_lveconfig': ve_lveconfig,
            've_default': ve_default,
            've_lve': ve_lve, 've_defaults': ve_defaults,
            've_package': ve_package, 'ubc': ubc,
            've_enter_by_name': ve_enter_by_name}

def guess_reseller_by_package(package):
    reseller = []
    global packages_users
    pkg_users_old = packages_users.copy()
    GetControlPanelUsers('list-resellers-packages')
    reseller_packages = packages_users.copy()
    packages_users = pkg_users_old.copy()

# reseller_packages: {'reseller_name': ['pack1', 'pack2']}
    for reseller_name, packages_list in reseller_packages.items():
        for package_name_in_key in packages_list:
            if package == package_name_in_key:
                reseller.extend([reseller_name])
    return reseller

# LU-400
def call_endurance_custom_script(args):
    """
    Call Endurance's custom script

:param args: list of arguments for pass to Endurance's custom script
    :return: None

"""

endurance_custom_script = cldetectlib.get_param_from_file(cldetectlib.CL_CONFIG_FILE,
                                                              'ENDURANCE_CUSTOM_SCRIPT',
                                                              separator='=')
    if endurance_custom_script and os.path.isfile(endurance_custom_script):

ret_code, std_out = exec_utility(endurance_custom_script, args)
        if ret_code != 0:
            message = f'Error while executing Endurance\'s custom script\n{std_out}'
            if JSON:
                json_format('multi', ['ERROR', message])
            else:
                err_message = f"error: {message}"
                sys.stderr.write(f"{err_message}\n")
                sys.exit(ret_code)

def _page_to_memory_or_bytes(value):
    """
    Convert page value to human-readable value or bytes, depending on BYTES_FLAG;
    E.g.
    >>> _page_to_memory_or_bytes(1233254)  # BYTES_FLAG=False
    '100M'
    >>> _page_to_memory_or_bytes(1233254)  # BYTES_FLAG=True
    654321

:type value: int
    :rtype: str | int
    """
    if BYTES_FLAG:
        return int(round(value * mmap.PAGESIZE))
    return _mb_mem(value)

def remove_absent_resellers():
    """
    Remove from LVE all resellers, which are absent from panel
    :return: None
    """
    # Build resellers ids list for removal
    reseller_id_list_for_delete = []
    # Get resellers list from cpapi
    # Create a list from a generator for repeated membership testing
    cpapi_resellers_list = list(lve.map.resellers())
    get_XML_cfg()  # for loading reseller_name<=>reseller_id map form ve.cfg
    # If reseller present in LVE, but absent in panel - add it to list for removal
    for lve_reseller_id in lvp_list():
        try:
            lve_reseller_name = lve.map.get_reseller_name(lve_reseller_id)
            if lve_reseller_name not in cpapi_resellers_list:
                # Reseller does not exist in panel, remove it from LVE
                reseller_id_list_for_delete.append(lve_reseller_id)
        except (KeyError, OSError, IOError):
            # No such user, remove it from LVE
            reseller_id_list_for_delete.append(lve_reseller_id)
    # Remove all selected resellers ignoring all errors (for example no such reseller)
    for reseller_id_for_delete in reseller_id_list_for_delete:
        _remove_reseller(reseller_id_for_delete)

def remove_absent_users():
    """
    Remove from LVE all users, which absent in system
    :return: None
    """

for lve_id in lve.proc.map():
        try:
            pwd.getpwuid(lve_id)  # Check the existence of the user
        except KeyError:
            try:
                lve.lve_destroy(lve_id)  # Destroy lve, if user not exist
            except PyLveError:  # If lve not exist
                try:
                    lve.py.lve_create(lve_id)  # we create lve
                    lve.lve_destroy(lve_id)  # and destroy them again
                    # After destroing lve, mapping will be cleansed of absent lve_id
                except PyLveError:
                    pass

Back