Viewing file:  mysql_governor_lib.py (22.32 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & 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
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import os
import xml.dom.minidom as xml
import subprocess
from distutils.version import LooseVersion
from xml.parsers.expat import ExpatError
from typing import Tuple, Optional # NOQA
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):
"""
Exception raised when dbgovernor daemon is offline
"""
def __init__(self):
super(MySQLGovernorDisabled, self).__init__({
'message':
"%(util)s is disabled in the system. "
"Please, run \"%(command)s\" to start the service",
'context': {
'util': 'MySQL governor',
'command': 'service db_governor restart'
}
})
class MySQLGovernorAbsent(MySQLGovException):
"""
Exception raised when dbgovernor isn't installed
"""
def __init__(self):
super(MySQLGovernorAbsent, self).__init__({
'message': "%(util)s not present in system",
'context': {'util': 'Governor'}
})
class MySQLGovernor:
"""
MysqlGovernor library
"""
# 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 = "/usr/sbin/dbctl"
_GOVERNOR_BINARY_PATH = "/usr/sbin/db_governor"
_CONTAINER_PATH = "/etc/container/mysql-governor.xml" # available for root only
_CONTAINER_PATH_V2 = "/var/run/mysql-governor-config.xml" # available for everyone; governor-mysql >= 1.1-14
def __init__(self):
# List of ignored users in governor
self._governor_ignored_users = None
# MySQL Governor limits
# Dictionary format: user_name -> (cpu_limit, io_limit)
# Both limits are integers, MB/s
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):
"""
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
"""
ret_code, std_out, std_err = run_command(
[self._UTILITY_PATH] + list(args),
return_full_output=True)
if "can't connect to socket" 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': "dbctl error: %(output)s",
'context': {'output': std_err}})
return ret_code, std_out, std_err
def get_governor_status(self):
# type: () -> Tuple[str, Optional[MySQLGovException]]
return self._detect_governor_status()
def _detect_governor_status(self):
# type: () -> 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: () -> xml.Document
config_path = self._get_config_path()
try:
with open(config_path, 'r') as f:
return xml.parseString(f.read())
except (IOError, OSError) as e:
self._is_governor_error = True
raise MySQLGovException(
{'message': "An error occured while loading governor "
"config from %(path)s. Error: %(error)s",
'context': {'path': config_path, 'error': str(e)}})
except ExpatError as e:
self._is_governor_error = True
raise MySQLGovException(
{'message': "An error occured while parsing governor "
"config from %(path)s. Error: %(error)s",
'context': {'path': config_path, 'error': str(e)}})
@staticmethod
def _load_ignore_users_from_xml(governor_cfg):
"""
Loads information about igrored users
:type governor_cfg: xml.Document
:return: list of ignore users
"""
ignore_users_list = list()
try:
gov_data = governor_cfg.getElementsByTagName("governor")[0]
except IndexError:
raise MySQLGovException({
'message': 'Malformed mysql-governor config. '
'Unable to find element \'%(element)s\'.',
'context': {'element': 'governor'}
})
users_data_list = gov_data.getElementsByTagName("user")
for user_data in users_data_list:
user_mode = user_data.getAttribute("mode")
if user_mode == "ignore":
# ignore_users_list.append(user_data.getAttribute("mysql_name"))
# ignore only system users
if user_data.getAttribute("name"):
ignore_users_list.append(user_data.getAttribute("name"))
return ignore_users_list
def _calc_rw_io_limits(self, io):
"""
Calculate db R/W IO limits based on governor version
:param io: requested limits in MB/s
:return: string values suitable to pass to "dbctl set ..."
"""
# Only Governor >= 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 = "%sb,%sb,%sb,%sb"
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 = "%s,%s,%s,%s"
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 "No" if we can't determine version
return LooseVersion(version) < LooseVersion(current)
@staticmethod
def _parse_line(line):
"""
Convert data line from dbctl to list
:param line:
Data line could be like:
"default\t400/380/350/300\t953/791/724/562\t953/791/724/562"
or:
"default 400/380/350/300 1000/830/760/590 1000/830/760/590"
depending on --kb/mb/bb option passed to dbctl
:return: list: ['default', '400/380/350/300', '953/791/724/562', '953/791/724/562']
"""
return [part for part in line.split() if part]
@staticmethod
def _percentage(value, percents):
"""
Calculate full list of governor limits by one value and percents koeff
"""
res = [value]
for k in percents:
res.append(int(value*k))
return tuple(res)
def is_governor_present(self):
"""
Get governor presence flag
:return:
"""
return self._is_governor_present
def get_governor_status_by_username(self, username):
"""
Get MySQL governor status for supplied user
:param username: Username for get status
:return: Governor status: "watched"/"ignored" or None if error
"""
# Load Governor data
self._load_info()
if username in self._governor_ignored_users:
return 'ignored'
return 'watched'
def get_limits_by_user(self, username):
"""
Get MySQL governor limits for supplied user
:param username: Username for read limits
:return: Cortege (CPU limit, IO limit). (None, None) if governor not present or error
"""
# Load Governor data if need
self._load_info()
if username in self._governor_limits:
return self._governor_limits[username]
return self._governor_limits['default']
def set_governor_status_for_user(self, username, status):
"""
Set MySQLGovernor status for single user
:param: `str` username: Username for set status
:param: `bool` status: True for "monitor", False for "ignore"
:return: `bool`: operation status result
"""
self._load_info()
status_cmd = "monitor" if status else "ignore"
ret, std_out, std_err = self._run_dbctl_with_args([status_cmd, username])
if std_err or ret:
exc_message = {'message': "Set governor status error(%(ret)s): %(output)s",
'context': {'ret': ret, 'output': std_err or std_out}}
raise MySQLGovException(exc_message)
return 0
def set_restricted_status_for_user(self, username, status):
"""
Set user restricted with dbctl utility
:param: `str` username: Username for set restricted status
:param: `bool` status: True for "restricted", False for "unrestricted"
:return: `bool`: operation status result
"""
self._load_info()
status_cmd = "restrict" if status else "unrestrict"
if username in ["root", "admin"]:
username = "default"
ret, std_out, std_err = self._run_dbctl_with_args([status_cmd, username])
if std_err or ret:
exc_message = {'message': "Set user restrict error(%(ret)s): %(output)s",
'context': {'ret': ret, 'output': std_err or std_out}}
raise MySQLGovException(exc_message)
return 0
def set_unrestricted_status_for_all_users(self):
"""
Set user restricted with dbctl utility
:return: `bool`: operation status result
"""
self._load_info()
ret, std_out, std_err = self._run_dbctl_with_args(["unrestrict-all"])
if std_err or ret:
exc_message = {'message': "Set all users unrestrict status error(%(ret)s): %(output)s",
'context': {'ret': ret, 'output': std_err or std_out}}
raise MySQLGovException(exc_message)
return 0
def get_restrict_status_by_username(self, username):
"""
Get MySQL governor status for supplied user
:param username: Username for get status
:return: Governor restricted status: "restricted"/"unrestricted"
"""
# Load Governor data
self._load_info()
if username in ["root", "admin"]:
username = "default"
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):
"""
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
"""
if cpu is None and io is None: # and read is None and write is None:
return 0
self._load_info()
cmd = ["set", username]
get_exc_message = lambda param: {'message': "Invalid %(param)s parameter",
'context': {'param': param}}
if cpu is not None:
if isinstance(cpu, int):
cpu = "%s,%s,%s,%s" % 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 = ",".join(map(str, cpu))
# else:
# raise MySQLGovException(get_exc_message('cpu'))
cmd.append("--cpu=%s" % cpu)
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 = ",".join(map(str, io))
# else:
# if isinstance(read, int):
# read = "%s,%s,%s,%s" % (read, read*0.83, read*0.76, read*0.59)
# elif isinstance(read, (list, tuple)) and len(read) == 4:
# read = ",".join(map(str, read))
# if isinstance(write, int):
# write = "%s,%s,%s,%s" % (write, write*0.83, write*0.76, write*0.59)
# elif isinstance(write, (list, tuple)) and len(write) == 4:
# write = ",".join(map(str, write))
# else:
# raise MySQLGovException(get_exc_message('limit'))
cmd.append("--read=%s" % read)
cmd.append("--write=%s" % write)
try:
ret, std_out, std_err = self._run_dbctl_with_args(cmd)
except ExternalProgramFailed as e:
raise MySQLGovException(str(e))
if std_err or ret:
exc_message = {'message': "Set all users unrestrict status error(%(ret)s): %(output)s",
'context': {'ret': ret, 'output': std_err or std_out}}
raise MySQLGovException(exc_message)
return 0
def _get_config_path(self):
"""
Get config path for mysql-governor;
:rtype: str|None
"""
if os.path.isfile(self._CONTAINER_PATH_V2):
return self._CONTAINER_PATH_V2
return self._CONTAINER_PATH
def _read_ignore_users(self):
"""Load ignore users list from container file"""
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):
"""
Loads users info from MySQL governor
:return: None
"""
# 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': "%(utility)s output is invalid",
'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 = [l.split()[0] for l 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': "There is no %(what)s found in %(where)s",
'context': {'what': 'default settings', 'where': '%s output' % self._UTILITY_PATH}
}
raise MySQLGovException(exc_message)
@staticmethod
def _get_user_io_limit(read_limit, write_limit):
"""
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
"""
try:
user_io_limit = int(read_limit) + int(write_limit)
except ValueError:
if read_limit == write_limit == "<1":
user_io_limit = 1
elif write_limit == "<1":
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):
"""
Executes dbctl --list --kb or dbctl --list
: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
"""
ret_code, gov_data_str, _ = self._run_dbctl_with_args(
['list', '--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, utility_exc_message):
"""
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 for exception
:return: dbctl data dictionary. Example:
{'default': (400, 1953124), 'cltest1': (400, 2025138)}
"""
governor_limits = dict()
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
user_limits_list = MySQLGovernor._parse_line(line)
# Data format verification
if len(user_limits_list) != 4:
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
# Add limits to dictionary
governor_limits[user_name] = (user_cpu_limit, user_io_limit_kb)
return governor_limits
|