Da

✏️ Editing: lvectl_packages_lib.py

# -*- coding: utf-8 -*-

# lvectl_packages_lib.py - module for interfacing with lvectl utility for get/set package LVE limits
#
# 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 json

from lveapi import NameMap
from lvectllib import lvp_list
from cllimits.lib import exec_utility
from cllimits.lib.utils import (_convert_memory_value_to_adaptive_format,
                                check_pmem_lower_bound_from_string)
from cllimits import LvectlException

class LveCtlPackage:
    def __init__(self):
        self._lve_version = None
        self._UTILITY_PATH = "/usr/sbin/lvectl"
        # Panel packages limits list
        # List item format:
        #   {
        #       "package": "Basic_16",
        #       "provider": "Reseller_Peter",
        #       "limits": { "vmem": 768, "pmem": 768, "ep": 25, "nproc": 50, "iops": 100, "speed": { "all": 50 },
        #                    "io": { "all": 5 }, "inodes": { "soft": 100000, "hard": 200000 }
        #   }
        self._packages_data_list = None
        self._name_map = NameMap()
        self._name_map.link_xml_node()

def _load_info(self, human_readable_numbers=False):
        """

Loads all package info from lvectl

:param human_readable_numbers: PMEM and VMEM limits in KBytes, MBytes or GBytes
        :return: None

"""
        # Do nothing, if data alreade present
        if self._packages_data_list is not None:
            return
        # lvectl all-package-list --json
        lvectl_cmd = ['all-package-list', '--json']
        if not human_readable_numbers:
            lvectl_cmd.append('--bytes')
        ret_code, lvectl_data_str = exec_utility(self._UTILITY_PATH, lvectl_cmd)
        loaded_json = json.loads(lvectl_data_str)
        if ret_code != 0:
            raise LvectlException(loaded_json['msg'])
        json_data = loaded_json['data']
        self._packages_data_list = []
        active_resellers = set(self._name_map.get_name(uid) for uid in lvp_list())
        for package_data in json_data:
            reseller_name = package_data['RESELLER']
            if reseller_name in ['N/A']:
                reseller_name = 'root'
            package_name = package_data['ID']
            # Build limits dictionary
            package_limits = {}
            package_limits['speed'] = int(package_data['SPEED'])
            package_limits['vmem'] = _convert_memory_value_to_adaptive_format(
                package_data['VMEM'],
                human_readable_numbers
            )
            if 'PMEM' in package_data:
                package_limits['pmem'] = _convert_memory_value_to_adaptive_format(
                    package_data['PMEM'],
                    human_readable_numbers
                )
            package_limits['ep'] = int(package_data['EP'])
            if 'NPROC' in package_data:
                package_limits['nproc'] = int(package_data['NPROC'])
            package_limits['io'] = {'all': int(package_data['IO'])}
            if 'IOPS' in package_data:
                package_limits['iops'] = int(package_data['IOPS'])
            # LU-505: add reseller status: either enabled or disabled reseller limits
            if 'provider' in ('root', ''):
                enabledLimits = False
            else:
                enabledLimits = reseller_name in active_resellers
            package_result_dict = {
                'package': package_name,
                'provider': reseller_name,
                'limits': package_limits,
                'enabledLimits': enabledLimits
            }
            self._packages_data_list.append(package_result_dict)

def get_reseller_package_limits(self, reseller_name_arg, package_name_arg, human_readable_numbers=False):
        """

Get limits for supplied reseller and package

:param reseller_name_arg: Reseller name. If None - all resellers
        :param package_name_arg: Package name. Only if reseller name is provided. If None - all packages
        :param human_readable_numbers: PMEM and VMEM limits in KBytes, MBytes or GBytes
        :return: Data list

List item format:
        {
            "package": "Basic_16",
            "provider": "Reseller_Peter",
            "limits": { "vmem": 768, "pmem": 768, "ep": 25, "nproc": 50, "iops": 100, "speed": { "all": 50 },
                        "io": { "all": 5 }, "inodes": { "soft": 100000, "hard": 200000 }
        }
        """
        self._load_info(human_readable_numbers=human_readable_numbers)
        if not reseller_name_arg:
            # if reseller not provided - return all resellers
            return self._packages_data_list
        # Reseller name provided
        out_data_list = []
        for package_data in self._packages_data_list:
            if package_data['provider'] == reseller_name_arg:
                # Supplied reseller found
                if package_name_arg:
                    # Package name is provided
                    if package_data['package'] == package_name_arg:
                        out_data_list.append(package_data)
                else:
                    # No package name provided
                    out_data_list.append(package_data)
        return out_data_list

def __set_error(self, param, package_name, err):
        return {'message': "%(what)s set error for package=%(package)s%(error)s",
                'context': {'what': param, 'package': package_name,
                            'error': f" [{err}]" if err else ""}}

def set_reseller_package_limits(self, reseller_name, package_name, limits_to_set):
        """
        Sets limits for supplied package
        :param reseller_name: Reseller name
        :param package_name: Package name
        :param limits_to_set: new LVE limits. Available keys: speed, vmem, pmem,
                        maxEntryProcs, io, nproc, iops
                    All other keys are ignoring. If some parameter absent on current LVE version
                    (for example pmem on LVE4), it will be ignored too.
        :return: None
        """
        if not limits_to_set:
            return

if 'pmem' in limits_to_set:
            error = check_pmem_lower_bound_from_string(limits_to_set['pmem'])
            if error:
                raise LvectlException(error)

# Call lvectl package-set-ext package_name --reseller=reseller_name ....
        # Form lvectl arguments list
        lvectl_cmd = ['package-set-ext', package_name]
        if reseller_name != 'root':
            lvectl_cmd.append(f'--reseller={reseller_name}')
        # Add limits from limits_to_set
        for limit_name, limit_value in limits_to_set.items():
            if limit_name == 'speed' and limit_value.isdigit():
                limit_value = f'{limit_value}%'
            lvectl_cmd.append(f'--{limit_name}={limit_value}')
        ret_code, lvectl_stdout, lvectl_stderr = exec_utility(self._UTILITY_PATH, lvectl_cmd, True)
        if ret_code != 0 or lvectl_stderr:
            # Set limits error
            raise LvectlException(self.__set_error('Limits', package_name, lvectl_stderr))

📁 Sid Gifari File Manager