[ PARENT ]

EDIT :: ve_config_reader.py

# coding=utf-8 # # 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 from collections import namedtuple from typing import Tuple, Optional, Type, List, Iterator # NOQA from lxml import etree # NOQA from clcommon.clpwd import ClPwd from .ve_config import get_xml_config_etree Limits = namedtuple('Limits', ['cpu', 'ncpu', 'io', 'vmem', 'pmem', 'nproc', 'ep', 'iops']) InheritedLimits = namedtuple('InheritedLimits', ['cpu', 'ncpu', 'io', 'vmem', 'pmem', 'nproc', 'ep', 'iops']) EMPTY_LIMITS = Limits( None, None, None, None, None, None, None, None ) # users whose owner is admin we mark with owner='root' # in order to simplify matching user<->package # let's use owner='root' for admin packages too DEFAULT_PROVIDER = 'root' class XmlConfigReader: """ This class parses our xml into user-friendly primitive structures (dicts, tuples, namedtuples) This class may do: - data conversion (cpu & ncpu in config -> speed) - validation This class should NOT: - take care of user/package/reseller existence - write anything to xml """ def __init__(self, _xml_config_etree=None): self.clpwd = ClPwd() if _xml_config_etree is None: self._xml_config = get_xml_config_etree() else: self._xml_config = _xml_config_etree self._users_map = dict(self._users_limits()) self._packages_map = dict(self._packages_limits()) self._resellers_limits_map = dict(self._resellers_limits()) self._resellers_defaults_map = dict(self._resellers_defaults()) def version(self): # type: () -> int return int(self._xml_config.find('version').text) def defaults(self): # type: () -> Limits return self._parse_limits_section( self._xml_config.find('defaults')) @staticmethod def _get_attribute_by_xpath_or_none(element, xpath, type_=int): # type: (etree.ElementBase, str, Type) -> Optional[object] try: value, = element.xpath(xpath) return type_(value) except (AttributeError, ValueError): return None @classmethod def _parse_limits_section(cls, element): # type: (etree.ElementBase) -> Limits return Limits( cpu=cls._get_attribute_by_xpath_or_none(element, 'cpu/@limit', str), ncpu=cls._get_attribute_by_xpath_or_none(element, 'ncpu/@limit'), io=cls._get_attribute_by_xpath_or_none(element, 'io/@limit'), vmem=cls._get_attribute_by_xpath_or_none(element, 'mem/@limit'), pmem=cls._get_attribute_by_xpath_or_none(element, 'pmem/@limit'), nproc=cls._get_attribute_by_xpath_or_none(element, 'nproc/@limit'), ep=cls._get_attribute_by_xpath_or_none(element, 'other/@maxentryprocs'), iops=cls._get_attribute_by_xpath_or_none(element, 'iops/@limit'), ) def get_user_limits(self, user_id): # type: (int) -> Limits return self._users_map.get(user_id) def get_package_limits(self, name, owner=DEFAULT_PROVIDER): # type: (str, str) -> Limits return self._packages_map.get((name, owner)) def get_reseller_default_limits(self, name): # type: (str) -> Limits return self._resellers_defaults_map.get(name) def get_reseller_limits(self, name): # type: (str) -> Limits return self._resellers_limits_map.get(name) def users_lve_ids(self): # type: () -> List[int] result = [] for lve in self._xml_config.findall('lve'): # lve tag has user attribute instead id attribute if use command lvectl --save-username if lve.get('id') is None: try: result.append(self.clpwd.get_uid(lve.get('user'))) except self.clpwd.NoSuchUserException: # We skip limits record with attribute `user` # if that user isn't exists in system pass else: result.append(int(lve.get('id'))) return result def _users_limits(self): # type: () -> Tuple[int, Limits] for lve in self._xml_config.findall('lve'): uid = lve.get('id') uid = self.clpwd.get_uid(lve.get('user')) if uid is None else int(uid) limits = self._parse_limits_section(lve) yield uid, limits def _packages_limits(self): # type: () -> Tuple[Tuple[str, str], Limits] ve_package = self._xml_config.findall("package") for package in ve_package: name = package.get('id') owner = package.get('reseller') or None limits = self._parse_limits_section(package) yield (name, owner or DEFAULT_PROVIDER), limits def _resellers_limits(self): # type: () -> Tuple[str, Limits] ve_reseller = self._xml_config.findall('reseller') for reseller in ve_reseller: name = reseller.get('user') limits = self._parse_limits_section(reseller) yield (name, limits) def _resellers_defaults(self): # type: () -> Tuple[str, Limits] ve_defaults = self._xml_config.findall('reseller/defaults') for defaults in ve_defaults: name = defaults.find('..').get('user') defaults = self._parse_limits_section(defaults) yield (name, defaults)

SAVE CANCEL