File Manager

Editing File: /opt/cloudlinux/venv/lib/python3.11/site-packages/clveconfig/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)

Back to Directory File Manager