<h2>Editing File: /opt/imunify360/venv/lib/python3.11/site-packages/clcommon/lib/mysql_governor_lib.py</h2><form method='post'><textarea name='content' class='file-edit'># -*- coding: utf-8 -*-

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

# mysql_governor.py - module for interfacing with dbctl utility for get/set MySQL limits and user's status

import os
import xml.dom.minidom as xml
import subprocess
import json
from xml.parsers.expat import ExpatError
from typing import Tuple, Optional, Dict  # NOQA

from packaging.version import Version

from clcommon.clexception import FormattedException
from clcommon.utils_cmd import run_command, ExternalProgramFailed


class GovernorStatus:
    ENABLED = 'enabled'
    ERROR = 'error'
    DISABLED = 'disabled'


class MySQLGovException(FormattedException):
    pass


class MySQLGovernorDisabled(MySQLGovException):
    &quot;&quot;&quot;
    Exception raised when dbgovernor daemon is offline
    &quot;&quot;&quot;
    def __init__(self):
        super().__init__({
            'message':
                &quot;%(util)s is disabled in the system. &quot;
                &quot;Please, run \&quot;%(command)s\&quot; to start the service&quot;,
            'context': {
                'util': 'MySQL governor',
                'command': 'service db_governor restart'
            }
        })


class MySQLGovernorAbsent(MySQLGovException):
    &quot;&quot;&quot;
    Exception raised when dbgovernor isn't installed
    &quot;&quot;&quot;
    def __init__(self):
        super().__init__({
            'message': &quot;%(util)s not present in system&quot;,
            'context': {'util': 'Governor'}
        })


def _get_exc_message(param):
    return {
        'message': &quot;Invalid %(param)s parameter&quot;,
        'context': {'param': param}
    }


class MySQLGovernor:
    &quot;&quot;&quot;
    MysqlGovernor library
    &quot;&quot;&quot;
    # Constants to calculate individial limits for cpu(current, short, middle period)
    #  and read(current, short, middle, long periods)
    # cpu:
    # c,  s = c*95%, m = c*87%, l = c*75%
    # io (read == write):
    # c = io / 2,  s = c*83%, m = c*76%, l = c*59%
    # refer to LU-167
    IO_PERCENTS = (0.83, 0.76, 0.59)  # percents for calculate read/write limits
    CPU_PERCENTS = (0.95, 0.87, 0.75)  # percents for calculate cpu limits
    _UTILITY_PATH = &quot;/usr/sbin/dbctl&quot;
    _PACKAGE_UTILITY_PATH = &quot;/usr/share/lve/dbgovernor/governor_package_limitting.py&quot;
    _GOVERNOR_BINARY_PATH = &quot;/usr/sbin/db_governor&quot;
    _CONTAINER_PATH = &quot;/etc/container/mysql-governor.xml&quot;  # available for root only
    _CONTAINER_PATH_V2 = &quot;/var/run/mysql-governor-config.xml&quot;  # available for everyone; governor-mysql &gt;= 1.1-14
    _S_DEFAULT = 'DEFAULT'

    def __init__(self):
        # List of ignored users in governor
        self._governor_ignored_users = None
        # MySQL Governor limits cache
        # Dictionary format: user_name -&gt; (cpu_limit: int, io_limit: int, cpu_limit_marked: str, io_limit_marked: str)
        # cpu_limit: int, io_limit: int - &quot;old&quot; limits, without package differs marks. For backward compatibility
        #   Both are integers always, IO limit - KB/s
        # cpu_limit_marked: str, io_limit_marked: str - &quot;new&quot; limits, with package differs marks.
        #   Both are strings always, IO limit - KB/s
        # Examples: (150, 3072, '150', '3072'), (200, 3072, '*200', '3072'), etc
        self._governor_limits = None
        # Error flag
        self._is_governor_error = False
        # MySQL Governor presence flag
        self._is_governor_present = os.path.isfile(self._UTILITY_PATH)
        self._governor_mode = self._detect_governor_mode()
        self._governor_version = self.get_governor_version()

    def get_governor_mode(self):
        return self._governor_mode

    def _detect_governor_mode(self):
        if self._is_governor_present:
            try:
                governor_cfg = self._get_xml_config()
                return governor_cfg.getElementsByTagName('lve')[0].getAttribute('use')
            except (MySQLGovException, IndexError):
                return None
        else:
            return None

    def _run_dbctl_with_args(self, args, check_exit_code=False):
        &quot;&quot;&quot;
        Run dbctl utility with given arguments and handle common errors:
        - governor is down: MySQLGovernorDisabled
        :param check_exit_code: whether we should raise exception
                                when dbctl returs code != 0
        &quot;&quot;&quot;
        ret_code, std_out, std_err = run_command(
            [self._UTILITY_PATH] + list(args),
            return_full_output=True)

        if &quot;can't connect to socket&quot; in std_out:
            # Governor not started
            self._is_governor_error = True
            self._governor_limits = None
            self._governor_ignored_users = None
            raise MySQLGovernorDisabled()
        elif check_exit_code and (ret_code or std_err):
            raise MySQLGovException({'message': &quot;dbctl error: %(output)s&quot;,
                                     'context': {'output': std_err}})
        return ret_code, std_out, std_err

    def get_governor_status(self):
        # type: () -&gt; Tuple[str, Optional[MySQLGovException]]
        return self._detect_governor_status()

    def _detect_governor_status(self):
        # type: () -&gt; Tuple[str, Optional[MySQLGovException]]
        if self.is_governor_present():
            try:
                self._load_info()
            except MySQLGovException as e:
                return GovernorStatus.ERROR, e
            except ExternalProgramFailed as e:
                return GovernorStatus.ERROR, MySQLGovException({
                    'message': str(e)
                })
            else:
                return GovernorStatus.ENABLED, None
        return GovernorStatus.DISABLED, None

    def _get_xml_config(self):
        # type: () -&gt; xml.Document
        config_path = self._get_config_path()
        try:
            with open(config_path, 'r', encoding='utf-8') as f:
                return xml.parseString(f.read())
        except (IOError, OSError) as e:
            self._is_governor_error = True
            raise MySQLGovException(
                {'message': &quot;An error occured while loading governor &quot;
                            &quot;config from %(path)s. Error: %(error)s&quot;,
                 'context': {'path': config_path, 'error': str(e)}}
            ) from e
        except ExpatError as e:
            self._is_governor_error = True
            raise MySQLGovException(
                {'message': &quot;An error occured while parsing governor &quot;
                            &quot;config from %(path)s. Error: %(error)s&quot;,
                 'context': {'path': config_path, 'error': str(e)}}
            ) from e

    @staticmethod
    def _load_ignore_users_from_xml(governor_cfg):
        &quot;&quot;&quot;
        Loads information about igrored users
        :type governor_cfg: xml.Document
        :return: list of ignore users
        &quot;&quot;&quot;
        ignore_users_list = []
        try:
            gov_data = governor_cfg.getElementsByTagName(&quot;governor&quot;)[0]
        except IndexError as e:
            raise MySQLGovException({
                'message': 'Malformed mysql-governor config. '
                           'Unable to find element \'%(element)s\'.',
                'context': {'element': 'governor'}
            }) from e
        users_data_list = gov_data.getElementsByTagName(&quot;user&quot;)
        for user_data in users_data_list:
            user_mode = user_data.getAttribute(&quot;mode&quot;)
            if user_mode == &quot;ignore&quot;:
                # ignore_users_list.append(user_data.getAttribute(&quot;mysql_name&quot;))
                # ignore only system users
                if user_data.getAttribute(&quot;name&quot;):
                    ignore_users_list.append(user_data.getAttribute(&quot;name&quot;))
        return ignore_users_list

    def _calc_rw_io_limits(self, io):
        &quot;&quot;&quot;
        Calculate db R/W IO limits based on governor version
        :param io: requested limits in MB/s
        :return: string values suitable to pass to &quot;dbctl set ...&quot;
        &quot;&quot;&quot;
        # Only Governor &gt;= 1.2-18 has support for limits in bytes
        # This check should be removed when new Governor will be in stable repo
        if self._is_governor_newer_then('1.2-17'):
            io_limits = io * 2 ** 20  # MB to Bytes
            limits_tmpl = &quot;%sb,%sb,%sb,%sb&quot;
        else:
            io_limits = io  # Use MBytes as is
            if io_limits == 1:
                # This should prevent dropping to defaults even for old
                # Governor, and just set 1MB instead
                io_limits = 2
            limits_tmpl = &quot;%s,%s,%s,%s&quot;

        read = limits_tmpl % self._percentage(int(io_limits // 2),
                                              self.IO_PERCENTS)
        write = limits_tmpl % self._percentage(int(io_limits // 2),
                                               self.IO_PERCENTS)
        return read, write

    def get_governor_version(self):
        if not self.is_governor_present():
            return None
        try:
            res = subprocess.check_output([self._GOVERNOR_BINARY_PATH, '--version'], text=True)
            # example of valid output
            # res = 'governor-mysql version 1.2-36'
            version = res.strip().split(' ')[2]
            return version
        except (subprocess.CalledProcessError, OSError):
            return None

    def _is_governor_newer_then(self, version):
        current = self._governor_version
        if not current:
            return False    # assume &quot;No&quot; if we can't determine version
        return Version(version) &lt; Version(current)

    @staticmethod
    def _parse_line(line):
        &quot;&quot;&quot;
        Convert data line from dbctl to list
        :param line:
            Data line could be like:
                &quot;default\t400/380/350/300\t953/791/724/562\t953/791/724/562&quot;
            or:
                &quot;default          400/380/350/300            1000/830/760/590                  1000/830/760/590&quot;
            depending on --kb/mb/bb option passed to dbctl
        :return: list: ['default', '400/380/350/300', '953/791/724/562', '953/791/724/562']
        &quot;&quot;&quot;
        return [part for part in line.split() if part]

    @staticmethod
    def _percentage(value, percents):
        &quot;&quot;&quot;
        Calculate full list of governor limits by one value and percents koeff
        &quot;&quot;&quot;
        res = [value]
        for k in percents:
            res.append(int(value*k))
        return tuple(res)

    def is_governor_present(self):
        &quot;&quot;&quot;
        Get governor presence flag
        :return:
        &quot;&quot;&quot;
        return self._is_governor_present

    def get_governor_status_by_username(self, username):
        &quot;&quot;&quot;
        Get MySQL governor status for supplied user
        :param username: Username for get status
        :return: Governor status: &quot;watched&quot;/&quot;ignored&quot; or None if error
        &quot;&quot;&quot;
        # Load Governor data
        self._load_info()
        if username in self._governor_ignored_users:
            return 'ignored'
        return 'watched'

    def get_limits_by_user(self, username, with_package_mark: bool = False) -&gt; Tuple[int, int]:
        &quot;&quot;&quot;
        Get MySQL governor limits for supplied user
        :param username: Username for read limits
        :param with_package_mark: False - without package limits difference mark (for compatibility with non-package
                                            governor, used in cloudlinux-top/cloudlinux-statistics),
                                  True - with package limits difference mark
        :return: Tuple (CPU limit, IO limit).
            Examples:
                (150, 3072) - with_package_mark == False
                ('*150', '3072') - with_package_mark == True
                ('*150', '*4096') - with_package_mark == True
            * - user has individual limits, differ from package limit
        MySQLGovException will be thrown if governor not present or error
        &quot;&quot;&quot;
        # Load Governor data if need
        self._load_info()
        if username in self._governor_limits:
            user_cpu_limit, user_io_limit_kb, user_cpu_limit_marked, user_io_limit_marked =\
                self._governor_limits[username]
        else:
            user_cpu_limit, user_io_limit_kb, user_cpu_limit_marked, user_io_limit_marked = \
                self._governor_limits['default']
        if with_package_mark:
            limits_for_return = (user_cpu_limit_marked, user_io_limit_marked)
        else:
            limits_for_return = (user_cpu_limit, user_io_limit_kb)
        return limits_for_return

    def set_governor_status_for_user(self, username, status):
        &quot;&quot;&quot;
        Set MySQLGovernor status for single user
        :param: `str` username: Username for set status
        :param: `bool` status: True for &quot;monitor&quot;, False for &quot;ignore&quot;
        :return: `bool`: operation status result
        &quot;&quot;&quot;
        self._load_info()
        status_cmd = &quot;monitor&quot; if status else &quot;ignore&quot;
        ret, std_out, std_err = self._run_dbctl_with_args([status_cmd, username])
        if std_err or ret:
            exc_message = {'message': &quot;Set governor status error(%(ret)s): %(output)s&quot;,
                           'context': {'ret': ret, 'output': std_err or std_out}}
            raise MySQLGovException(exc_message)
        return 0

    def set_restricted_status_for_user(self, username, status):
        &quot;&quot;&quot;
        Set user restricted with dbctl utility
        :param: `str` username: Username for set restricted status
        :param: `bool` status: True for &quot;restricted&quot;, False for &quot;unrestricted&quot;
        :return: `bool`: operation status result
        &quot;&quot;&quot;
        self._load_info()
        status_cmd = &quot;restrict&quot; if status else &quot;unrestrict&quot;
        if username in [&quot;root&quot;, &quot;admin&quot;]:
            username = &quot;default&quot;
        ret, std_out, std_err = self._run_dbctl_with_args([status_cmd, username])
        if std_err or ret:
            exc_message = {'message': &quot;Set user restrict error(%(ret)s): %(output)s&quot;,
                           'context': {'ret': ret, 'output': std_err or std_out}}
            raise MySQLGovException(exc_message)
        return 0

    def set_unrestricted_status_for_all_users(self):
        &quot;&quot;&quot;
        Set user restricted with dbctl utility
        :return: `bool`: operation status result
        &quot;&quot;&quot;
        self._load_info()
        ret, std_out, std_err = self._run_dbctl_with_args([&quot;unrestrict-all&quot;])
        if std_err or ret:
            exc_message = {'message': &quot;Set all users unrestrict status error(%(ret)s): %(output)s&quot;,
                           'context': {'ret': ret, 'output': std_err or std_out}}
            raise MySQLGovException(exc_message)
        return 0

    def get_restrict_status_by_username(self, username):
        &quot;&quot;&quot;
        Get MySQL governor status for supplied user
        :param username: Username for get status
        :return: Governor restricted status: &quot;restricted&quot;/&quot;unrestricted&quot;
        &quot;&quot;&quot;
        # Load Governor data
        self._load_info()
        if username in [&quot;root&quot;, &quot;admin&quot;]:
            username = &quot;default&quot;
        if username in self._governor_restricted_users:
            return 'restricted'
        return 'unrestricted'

    def set_limits_for_user(self, username, cpu=None, io=None):
    # this function interface for full edit mode
    # def set_limits_for_user(self, username, cpu=None, io=None, read=None,
    #                         write=None):
        &quot;&quot;&quot;
        Set MySQLGovernor limits for user
        :param: username `str`: username for set limits
        :param: `int`|`list` cpu: governor cpu limit. when it param int -
                                  calculate by percentage other params
        :param: `int`|`list` io: io value means that read and write limits similar
        :param: `int`|`list` read: read limit
        :param: `int`|`list` write: write limit
        :return: 0
        &quot;&quot;&quot;
        if cpu is None and io is None:  #  and read is None and write is None:
            return 0

        self._load_info()

        cmd = [&quot;set&quot;, username]
        if cpu is not None:
            if isinstance(cpu, int):
                cpu = &quot;,&quot;.join(map(str, self._percentage(cpu, self.CPU_PERCENTS)))
            else:
                raise MySQLGovException(_get_exc_message('cpu'))
            # uncomment this lines for add full edit mode
            # elif isinstance(cpu, (list, tuple)) and len(cpu) == 4:
            #     cpu = &quot;,&quot;.join(map(str, cpu))
            # else:
            #     raise MySQLGovException(_get_exc_message('cpu'))
            cmd.append(f&quot;--cpu={cpu}&quot;)

        if io is not None:
        # uncomment this line for add full edit mode
        # if io is not None or read is not None or write is not None:
            if isinstance(io, int):
                read, write = self._calc_rw_io_limits(io)
            else:
                raise MySQLGovException(_get_exc_message('io'))
            # uncomment this lines for add full edit mode
            # elif isinstance(io, (list, tuple)) and len(io) == 4:
            #     read = write = &quot;,&quot;.join(map(str, io))
            # else:
            #     if isinstance(read, int):
            #         read = &quot;%s,%s,%s,%s&quot; % (read, read*0.83, read*0.76, read*0.59)
            #     elif isinstance(read, (list, tuple)) and len(read) == 4:
            #         read = &quot;,&quot;.join(map(str, read))
            #     if isinstance(write, int):
            #         write = &quot;%s,%s,%s,%s&quot; % (write, write*0.83, write*0.76, write*0.59)
            #     elif isinstance(write, (list, tuple)) and len(write) == 4:
            #         write = &quot;,&quot;.join(map(str, write))
            #     else:
            #         raise MySQLGovException(_get_exc_message('limit'))

            cmd.append(f&quot;--read={read}&quot;)
            cmd.append(f&quot;--write={write}&quot;)
        try:
            ret, std_out, std_err = self._run_dbctl_with_args(cmd)
        except ExternalProgramFailed as e:
            raise MySQLGovException(str(e)) from e
        if std_err or ret:
            exc_message = {'message': &quot;Set all users unrestrict status error(%(ret)s): %(output)s&quot;,
                           'context': {'ret': ret, 'output': std_err or std_out}}
            raise MySQLGovException(exc_message)
        # Reset users limits cache
        self._governor_limits = None
        return 0

    def _get_package_raw_limits_from_utility(self, package_name: Optional[str]) -&gt; dict:
        &quot;&quot;&quot;
        Retrieve MySQL Governor package limits
        :param package_name: Package name. If None, get all packages name
        :return: Dict with limits. Example:
            {'pack1': {'cpu': 100, 'io': 900}, 'pack2': {'cpu': 100, 'io': 900}}
        &quot;&quot;&quot;
        if package_name:
            cmd_list = [self._PACKAGE_UTILITY_PATH, 'get',
                        f&quot;--package={package_name.encode().decode('unicode-escape')}&quot;,
                        '--format=kb']
        else:
            cmd_list = [self._PACKAGE_UTILITY_PATH, 'get', '--all', '--format=kb']
        ret_code, std_out, std_err = run_command(cmd_list, return_full_output=True)
        if ret_code != 0 or std_err:
            raise MySQLGovException({
                'message': &quot;'%(cmd)s' failed, stderr is: %(stderr)s&quot;,
                'context': {'cmd': ' '.join(cmd_list), 'stderr': std_err}
            })
        try:
            package_data_from_util = json.loads(std_out)
        except (json.JSONDecodeError, ) as e:
            raise MySQLGovException({
                'message': &quot;%(util)s output invalid, error is: %(error)s&quot;,
                'context': {'util': self._PACKAGE_UTILITY_PATH, 'error': str(e)}
            }) from e
        return package_data_from_util

    def _calc_package_limits_from_raw(self, cpu_limits_list: list, read_limits_list: list,
                                      write_limits_list: list) -&gt; Tuple[int, int]:
        &quot;&quot;&quot;
        Calculate package limits from raw governor limits
        :param cpu_limits_list: CPU limits list
        :param read_limits_list: Read limits list
        :param write_limits_list: Write limits list
        :return: Tuple: (cpu_limit, io_limit)
        &quot;&quot;&quot;
        return int(cpu_limits_list[0]), self._get_user_io_limit(str(read_limits_list[0]), str(write_limits_list[0]))

    def get_package_limits(self, package_name: Optional[str] = None) -&gt; Optional[Dict]:
        &quot;&quot;&quot;
        Retrieve MySQL Governor package limits
        :param package_name: Package name. If None, get all packages name
        :return: Dict with limits. Example:
            {'pack1': {'cpu': 100, 'io': 900}, 'pack2': {'cpu': 100, 'io': 900}}
        &quot;&quot;&quot;
        try:
            package_data_from_util = self._get_package_raw_limits_from_utility(package_name)
        except MySQLGovException:       # pylint: disable=try-except-raise
            raise
        # Convert limits from governor
        packages_data = {}
        for pack_name, pack_limits in package_data_from_util.items():
            cpu_limit, io_limit = self._calc_package_limits_from_raw(pack_limits['cpu'], pack_limits['read'],
                                                                     pack_limits['write'])
            packages_data[pack_name] = {'cpu': cpu_limit, 'io': io_limit}
        return packages_data

    def reset_user_limits_to_defaults(self, username: str, limit_names: list):
        &quot;&quot;&quot;
        Reset users limits to default
        :param username: User name ro reset limits
        :param limit_names: Limit names list to reset
        &quot;&quot;&quot;
        # /usr/share/lve/dbgovernor/governor_package_limitting.py reset_individual \
        #   --user=res1 --limits=mysql-cpu,mysql-io
        limits_string = ','.join(limit_names)
        cmd_list = [self._PACKAGE_UTILITY_PATH, 'reset_individual', f'--user={username}',
                    f'--limits={limits_string}']
        ret_code, _, std_err = run_command(cmd_list, return_full_output=True)
        # For reliability we check both retcode and std_err
        if ret_code != 0 or std_err:
            raise MySQLGovException({
                'message': &quot;'%(cmd)s' is failed, stderr is: %(stderr)s&quot;,
                'context': {'cmd': ' '.join(cmd_list), 'stderr': std_err}
            })

    def _set_governor_limits_to_cpanel_package(  # pylint: disable=too-many-branches
        self,
        package_name: str,
        cpu_limit: Optional[int],
        io_limit: Optional[int]
    ):
        &quot;&quot;&quot;
        Set MySQL Governor to cPanel package file. If limits not changed, package file will not be written
        :param package_name: Package name
        :param cpu_limit: MySQL CPU limit to set
        :param io_limit: MySQL IO limit to set
        &quot;&quot;&quot;
        from clcommon.utils import get_file_lines, write_file_lines  # pylint: disable=import-outside-toplevel
        package_path = f'/var/cpanel/packages/{package_name}'
        try:
            cpanel_package_lines = get_file_lines(package_path)
        except (OSError, IOError, ):
            return
        if len(cpanel_package_lines) == 0:
            return
        lines_to_write = []
        is_change_made = False
        # Find and change limit lines:
        # lve_mysql_cpu=4000
        # lve_mysql_io=4096
        for line in cpanel_package_lines:
            if line.startswith('lve_mysql_cpu') and cpu_limit is not None:
                parts = line.strip().split('=')
                if len(parts) != 2:
                    continue
                if cpu_limit == 0:
                    s_cpu_limit = self._S_DEFAULT
                else:
                    s_cpu_limit = str(cpu_limit)
                s_old_cpu_limit = parts[1].strip()
                if s_old_cpu_limit != s_cpu_limit:
                    lines_to_write.append(f'lve_mysql_cpu={s_cpu_limit}\n')
                    is_change_made = True
                else:
                    # MYSQL CPU limit unchanged, save old line
                    lines_to_write.append(f'{line}\n')
            elif line.startswith('lve_mysql_io') and io_limit is not None:
                if io_limit == 0:
                    s_io_limit = self._S_DEFAULT
                else:
                    s_io_limit = str(io_limit)
                parts = line.strip().split('=')
                if len(parts) != 2:
                    continue
                value = parts[1].strip()
                if value != s_io_limit:
                    lines_to_write.append(f'lve_mysql_io={s_io_limit}\n')
                    is_change_made = True
                else:
                    # MYSQL IO limit unchanged, save old line
                    lines_to_write.append(f'{line}\n')
            else:
                lines_to_write.append(line)
        if is_change_made:
            write_file_lines(package_path, lines_to_write, 'w')

    def _apply_package_limits_to_cpanel(self, package_name: str, cpu_limit: Optional[int], io_limit: Optional[int]):
        &quot;&quot;&quot;
        Apply all MySQL Governor packages limits to cpanel package file. In not cPanel, do nothing
        :param package_name: Package name to update
        :param cpu_limit: MySQL CPU limit to set, None - not change
        :param io_limit: MySQL IO limit to set, None - not change
        &quot;&quot;&quot;
        from clcommon.cpapi import getCPName  # pylint: disable=import-outside-toplevel
        if getCPName() != 'cPanel':
            return
        self._set_governor_limits_to_cpanel_package(package_name, cpu_limit, io_limit)

    def set_package_limits(self, package_name: str, cpu_limit: Optional[int] = None,
                           io_limit: Optional[int] = None):
        &quot;&quot;&quot;
        Set limits for Governor package
        :param package_name: Package name for set
        :param cpu_limit: MySQL CPU limit to set
        :param io_limit: MySQL CPU limit to set
        &quot;&quot;&quot;
        # Argument validation
        # /usr/share/lve/dbgovernor/governor_package_limitting.py set --package pack2 --cpu=200,201,202,203
        #  --read=500,501,502,503 --write=400,401,402,403
        if package_name is None or cpu_limit is None and io_limit is None:
            raise MySQLGovException(&quot;MySQLGovernor.set_package_limits arguments error: &quot;
                                    &quot;Package name and at least one limit &quot;
                                    f&quot;should be provided. Current arguments: package name: {package_name}; &quot;
                                    f&quot;cpu limit is {cpu_limit}; IO limit is {io_limit};&quot;)

        cmd_list = [self._PACKAGE_UTILITY_PATH, 'set', '--package', package_name]
        # Check arguments and prepare command line parameters for governor utility
        if cpu_limit is not None:
            if isinstance(cpu_limit, int):
                cpu = &quot;,&quot;.join(map(str, self._percentage(cpu_limit, self.CPU_PERCENTS)))
            else:
                raise MySQLGovException(_get_exc_message('cpu_limit'))
            cmd_list.append(f&quot;--cpu={cpu}&quot;)

        if io_limit is not None:
            if isinstance(io_limit, int):
                read, write = self._calc_rw_io_limits(io_limit)
            else:
                raise MySQLGovException(_get_exc_message('io_limit'))
            cmd_list.append(f&quot;--read={read}&quot;)
            cmd_list.append(f&quot;--write={write}&quot;)
        ret_code, std_out, std_err = run_command(cmd_list, return_full_output=True)
        if ret_code != 0 or std_err:
            raise MySQLGovException({
                'message': &quot;'%(command)s' is failed, stdout is: '%(stdout)s', stderr is: '%(stderr)s'&quot;,
                'context': {'command': ' '.join(cmd_list), 'stdout': std_out, 'stderr': std_err}
            })
        # Apply limits to cPanel package
        self._apply_package_limits_to_cpanel(package_name, cpu_limit, io_limit)

    def _get_config_path(self):
        &quot;&quot;&quot;
        Get config path for mysql-governor;
        :rtype: str|None
        &quot;&quot;&quot;
        if os.path.isfile(self._CONTAINER_PATH_V2):
            return self._CONTAINER_PATH_V2
        return self._CONTAINER_PATH

    def _read_ignore_users(self):
        &quot;&quot;&quot;Load ignore users list from container file&quot;&quot;&quot;
        try:
            governor_xml = self._get_xml_config()
            self._governor_ignored_users = \
                self._load_ignore_users_from_xml(governor_xml)
        except MySQLGovException:
            self._governor_limits = None
            self._governor_ignored_users = None
            raise

    def _load_info(self):
        &quot;&quot;&quot;
        Loads users info from MySQL governor
        :return: None
        &quot;&quot;&quot;
        # Exit if governor data already loaded
        if self._governor_ignored_users is not None and self._governor_limits is not None:
            return

        # Exit if governor not present
        if not self._is_governor_present:
            self._is_governor_error = True
            raise MySQLGovernorAbsent()

        utility_exc_message = {'message': &quot;%(utility)s output is invalid&quot;,
                               'context': {'utility': self._UTILITY_PATH}}

        self._read_ignore_users()

        # Load governor limits
        is_kb_limits_ok, gov_data_str = self._run_dbctl_list()
        _, gov_restricted_str, _ = \
            self._run_dbctl_with_args(['list-restricted'], check_exit_code=True)

        self._governor_restricted_users = [
            line.split()[0] for line in gov_restricted_str.strip().split('\n')[1:]
        ]
        # Parse dbctl output
        gov_data_lines = gov_data_str.split('\n')
        self._governor_limits = self._parse_dbctl_data_lines(gov_data_lines, is_kb_limits_ok, utility_exc_message)
        # Check default settings presence
        if 'default' not in self._governor_limits:
            self._is_governor_error = True
            self._governor_limits = None
            self._governor_ignored_users = None
            exc_message = {
                'message': &quot;There is no %(what)s found in %(where)s&quot;,
                'context': {'what': 'default settings', 'where': f'{self._UTILITY_PATH} output'}
            }
            raise MySQLGovException(exc_message)

    @staticmethod
    def _get_user_io_limit(read_limit: str, write_limit: str):
        &quot;&quot;&quot;
            Calculates the io limit.
            Handles the situation when user's write or read io limit is less than 1mb/s (PTCLLIB-85).
        :type write_limit: str
        :type read_limit: str
        :rtype: int
        &quot;&quot;&quot;
        try:
            user_io_limit = int(read_limit) + int(write_limit)
        except ValueError:
            if read_limit == write_limit == &quot;&lt;1&quot;:
                user_io_limit = 1
            elif write_limit == &quot;&lt;1&quot;:
                user_io_limit = int(read_limit)
            else:
                user_io_limit = int(write_limit)
        return user_io_limit

    def _run_dbctl_list(self, _is_incorrect_syntax=False):
        &quot;&quot;&quot;
        Executes dbctl list-marked --kb or dbctl list-marked
        :param _is_incorrect_syntax: True is emulate dbctl error. Only for testing!
        :return: Cortege (is_kb_limits_ok, stdout_str), where
            is_kb_limits_ok == True, if dbctl returned limits in KB, else - False
            stdout_str - dbctl stdout string
        &quot;&quot;&quot;
        ret_code, gov_data_str, _ = self._run_dbctl_with_args(
            ['list-marked', '--kb'], check_exit_code=True)
        # Check is KB limits supported
        is_kb_limits_ok = True
        if _is_incorrect_syntax or 'Incorrect syntax' in gov_data_str:
            # --kb option not supported, call without it
            _, gov_data_str, _ = self._run_dbctl_with_args(
                ['list'], check_exit_code=True)
            is_kb_limits_ok = False
        return is_kb_limits_ok, gov_data_str

    def _parse_dbctl_data_lines(self, data_lines_list, is_kb_limits_ok: bool, utility_exc_message: dict) -&gt; dict:
        &quot;&quot;&quot;
        Converts data lines from dbctl stdout to dictionary
        :param data_lines_list: List of lines from dbctl stdout
        :param is_kb_limits_ok: Is limits already in KB/s
        :param utility_exc_message: Message dict for exception
        :return: Tuple(dict, dict) dbctl data dictionary. Example:
            {'default': (400, 1953124, '400', '1953124'),
             'cltest1': (350, 2025138, '*350', '2025138')
             }
        &quot;&quot;&quot;
        governor_limits = {}
        for line in data_lines_list:
            line = line.strip()
            # Pass header line and empty lines
            if not line or 'cpu(%)' in line:
                continue
            # List: [0] - username, [1] - CPU limits, [2] - read limits, [3] - write limits, [4] - package limits marks
            user_limits_list = MySQLGovernor._parse_line(line)
            # Data format verification
            if len(user_limits_list) != 5 or len(user_limits_list[4]) != 2:
                self._is_governor_error = True
                self._governor_limits = None
                self._governor_ignored_users = None
                raise MySQLGovException(utility_exc_message)
            cpu_limits_list = user_limits_list[1].split('/')  # '400/380/350/300'
            read_limits_list = user_limits_list[2].split('/')  # '1000/830/760/590'
            write_limits_list = user_limits_list[3].split('/')  # '1000/830/760/590'
            if len(cpu_limits_list) != 4 or len(read_limits_list) != 4 or len(write_limits_list) != 4:
                self._is_governor_error = True
                self._governor_limits = None
                self._governor_ignored_users = None
                raise MySQLGovException(utility_exc_message)
            user_name = user_limits_list[0]
            # Determine CPU limit as [0] limit
            user_cpu_limit = int(cpu_limits_list[0])
            # Determine IO limit as read_limit[0] + write_limit[0]
            user_io_limit = self._get_user_io_limit(read_limits_list[0], write_limits_list[0])
            # limit if is_kb_limits_ok else limit*1024
            user_io_limit_kb = user_io_limit if is_kb_limits_ok else user_io_limit*1024
            # Process package differ marks
            # Package limits marks are placed in user_limits_list[4] and shows as '+-' (for example)
            # There are 2 marks always, each or '-' or '+'
            # 1st mark - CPU limit
            # 2nd mark - IO limit
            # If mark is '-' - package limit, if '+' - individual (should be marked by '*')
            marks = user_limits_list[4]
            if marks[0] == '+':
                user_cpu_limit_marked = f'*{cpu_limits_list[0]}'
            else:
                user_cpu_limit_marked = cpu_limits_list[0]
            if marks[1] == '+':
                user_io_limit_marked = f'*{user_io_limit_kb}'
            else:
                user_io_limit_marked = str(user_io_limit_kb)
            # Add limits to dictionary
            governor_limits[user_name] = (user_cpu_limit, user_io_limit_kb, user_cpu_limit_marked, user_io_limit_marked)
        return governor_limits
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fimunify360%2Fvenv%2Flib%2Fpython3.11%2Fsite-packages%2Fclcommon%2Flib'>Back to Directory</a>