"""
OneAgent MySQL plugin
=====================
Gathers stats from a given MySQL server instance. The stats are selected values from 2 queries:
* show global status (http://dev.mysql.com/doc/refman/5.7/en/show-variables.html)
* show global variables (http://dev.mysql.com/doc/refman/5.7/en/show-status.html)
ruxit_MySQLdb module is used to access the MySQL instance. It is based on mysql-ctypes - ported to work with python 3
and using bundled libmariadb instead of libmysql as the underlying C library.
To avoid confusion the package is distributed under a unique name with `ruxit_` prefix.
Running the plugin requires specifying user credentials and server port in the configuration.
This is a singleton type plugin - one instance will be created even if multiple MySQL process
groups instances are detected.
The plugin can only query one MySQL server due to limitations of ruxit plugin configuration mechanism. In case
of multiple MySQL server running on a one host, the one monitored will the one listening on port specified in
the configuration.
"""
from ruxit.api.data import PluginMeasurement
from ruxit.api.base_plugin import BasePlugin
from ruxit.api.exceptions import AuthException, ConfigException
from ruxit.api.selectors import ListenPortSelector
from ruxit.api.snapshot import parse_port_bindings
import itertools
try:
import ruxit_MySQLdb as MySQLdb
except OSError as mysql_load_ex:
raise ConfigException("Unable to load mysql connector library") from mysql_load_ex
[docs]class MySqlPlugin(BasePlugin):
"""
MySQL plugin class. This plugin is stateful - initialize and close methods are defined
in order to maintain one server connection for the lifetime of the plugin.
Class fields mirror measurements declared in the plugin.json
and declare some few parameters i.e: connection timeout, error codes
"""
_DEFAULT_TIMEOUT = 5
# see http://dev.mysql.com/doc/refman/5.7/en/error-messages-server.html
_MYSQL_ACCESS_DENIED_ERRORS = [1044, 1045, 1142, 1143, 1227, 1370, 1698, 1873, 3118]
# see https://dev.mysql.com/doc/refman/5.7/en/error-messages-client.html
_MYSQL_CONNECTION_ERRORS = [2003]
_STATUS_RELATIVE_VALUES = {
'threads_connected',
'connection_errors_max_connections',
'queries',
'slow_queries',
'questions',
'created_tmp_tables',
'created_tmp_disk_tables',
'table_locks_waited',
'innodb_buffer_pool_pages_total',
'innodb_buffer_pool_pages_dirty',
'innodb_buffer_pool_pages_free',
'innodb_buffer_pool_pages_data',
'threads_created',
'threads_running',
'table_locks_immediate'
}
_STATUS_PER_SECOND_VALUES = {
'com_delete',
'com_delete_multi',
'com_insert',
'com_insert_select',
'com_update',
'com_update_multi',
'com_select',
'com_replace_select',
'qcache_queries_in_cache',
'innodb_data_reads',
'innodb_data_writes',
'qcache_hits',
'qcache_not_cached'
}
_STATUS_ABSOLUTE_VALUES = {
'innodb_buffer_pool_pages_total',
'innodb_buffer_pool_pages_dirty',
'innodb_buffer_pool_pages_free',
'innodb_buffer_pool_pages_data',
'threads_connected',
'threads_created',
'threads_running',
'qcache_free_memory',
'qcache_queries_in_cache'
}
# quotient = numerator/denominator
_STATUS_RATE_VALUES = {
'slow_queries_rate':{'numerator':'slow_queries','denominator':'queries'}
}
_VARS_ABSOLUTE_VALUES = {
'innodb_buffer_pool_size'
}
_STATUS_VARIABLES = _STATUS_RELATIVE_VALUES.union(_STATUS_ABSOLUTE_VALUES).union(_STATUS_PER_SECOND_VALUES)
_ABSOLUTE_VALUES = _STATUS_ABSOLUTE_VALUES.union(_VARS_ABSOLUTE_VALUES)
[docs] def initialize(self, **kwargs):
"""
Initializes the mysql plugin.
This parses the configuration of the plugin and initiates a connection object so that
every query does not need to establish a new connection.
The connection is established over tcp by specifying 127.0.0.1 as connection address. Using
localhost would cause the connection to be made via unix sockets which would require additional
discovery of the socket. Note that this is the limitation of the C connector library and
other package can behave differently.
Raises:
ruxit.api.exceptions.AuthException: mysql server returned error indicating auth issues
ruxit.api.exceptions.ConfigException: unable to connect for other reasons
"""
self.process_types = kwargs["activation_context"].value.process_types
config = kwargs["config"]
self.port = int(config["connection_port"])
#Find assiociated entity and ports bindigs data for valid connections IP address
host = "127.0.0.1"
port_found = False
if (self.query_snapshot is not None) and (self.query_snapshot.entries is not None):
for se in self.query_snapshot.entries:
if se.process_type not in self.process_types:
continue
port_bindings = parse_port_bindings (se)
for binding_data in port_bindings:
if self.port == binding_data [1]:
host = binding_data [0]
port_found = True
break
if port_found:
break
connection_config = {
'user': config["connection_user"].encode('ascii'),
'passwd': config["connection_password"].encode('ascii'), 'port': self.port,
'host': host.encode("ascii"), 'connect_timeout': self._DEFAULT_TIMEOUT
}
try:
self.connection = MySQLdb.connect(**connection_config)
self.cursor = self.connection.cursor()
except MySQLdb.OperationalError as ex:
if ex.args[0] in self._MYSQL_ACCESS_DENIED_ERRORS:
raise AuthException from ex
elif ex.args[0] in self._MYSQL_CONNECTION_ERRORS:
raise ConfigException("Unable to connect to MySQL server") from ex
elif len(ex.args) > 1:
raise ConfigException(ex.args[1]) from ex
raise
# build list of locally calculated relative metrics
self._rate_metrics = set()
for m in self._STATUS_RATE_VALUES:
mdef = self._STATUS_RATE_VALUES[m]
self._rate_metrics.add(mdef['denominator'])
self._rate_metrics.add(mdef['numerator'])
self._rate_old_results = {}
self._rate_results = {}
# this below assumes that rate metrics are dimension-less
def _store_relative_metric(self,metric):
if metric.key in self._rate_metrics:
self._rate_results[metric.key] = metric
def _flush_rate_metrics(self):
# calculate deltas and store them for next step
tmp_rate_metrics = {}
for metric, result in self._rate_results.items():
old_result = self._rate_old_results.get(metric, None)
if old_result is not None:
value_delta = result.value - old_result.value
if value_delta < 0:
continue
tmp_rate_metrics[metric] = value_delta
# walk all rate metrics defitions and calculate values based on tmp_rate_metrics
for mkey, mdef in self._STATUS_RATE_VALUES.items():
mdef_numerator = tmp_rate_metrics.get(mdef['numerator'], None)
mdef_denominator = tmp_rate_metrics.get(mdef['denominator'], None)
if mdef_numerator is not None and mdef_denominator is not None:
if mdef_numerator >=0 and mdef_denominator > 0:
self.results_builder.add_absolute_result( PluginMeasurement(
key=mkey,
value=100.0* mdef_numerator/mdef_denominator,
entity_selector=ListenPortSelector(self.port, self.process_types)
)
)
self._rate_old_results = self._rate_results
self._rate_results = {}
[docs] def query(self, **kwargs):
"""
Gathers the mysql data from global status and global variables queries.
The measurements use a ruxit.api.selectors.ListenPortSelector to associate data with correct
entity id (the mysql server process group) in case of multiple MySQL instances running
on a host.
Raises:
ruxit.api.exceptions.AuthException: mysql server returned error indicating auth issues
ruxit.api.exceptions.ConfigException: unable to connect for other reasons
"""
try:
variables_query_fragment = ', '.join('\'%s\'' % it for it in self._STATUS_VARIABLES)
status_query = 'show global status where variable_name in (%s)' % variables_query_fragment
status_query = status_query.encode('ascii')
self.cursor.execute(status_query)
global_status_rows = self.cursor.fetchall()
self.cursor.execute(
b'show global variables where variable_name in (\'innodb_buffer_pool_size\')'
)
global_variables_rows = self.cursor.fetchall()
except MySQLdb.OperationalError as ex:
if ex.args[0] in self._MYSQL_ACCESS_DENIED_ERRORS:
raise AuthException from ex
elif ex.args[0] in self._MYSQL_CONNECTION_ERRORS:
raise ConfigException("Unable to connect to MySQL server") from ex
elif len(ex.args) > 1:
raise ConfigException(ex.args[1]) from ex
raise
results = [
PluginMeasurement(
key=row[0].decode('ascii').lower(),
value=row[1],
entity_selector=ListenPortSelector(self.port, self.process_types)
)
for row in itertools.chain(global_status_rows, global_variables_rows)
]
for r in results:
if r.key in self._ABSOLUTE_VALUES:
self.results_builder.add_absolute_result(r)
elif r.key in self._STATUS_PER_SECOND_VALUES:
self.results_builder.add_per_second_result(r)
else:
self.results_builder.add_relative_result(r)
self._store_relative_metric(r)
self._flush_rate_metrics()
[docs] def close(self, **kwargs):
self.connection.close()
