D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
lib
/
mysqlsh
/
python-packages
/
mysql_gadgets
/
common
/
Filename :
server.py
back
Copy
# # Copyright (c) 2016, 2024, Oracle and/or its affiliates. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License, version 2.0, # as published by the Free Software Foundation. # # This program is designed to work with certain software (including # but not limited to OpenSSL) that is licensed under separate terms, # as designated in a particular file or component or in included license # documentation. The authors of MySQL hereby grant you an additional # permission to link the program and your derivative works with the # separately licensed software that they have either included with # the program or referenced in the documentation. # # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See # the GNU General Public License, version 2.0, for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA # """ This module contains an abstraction of a MySQL server object used by multiple gadgets. It also contains methods and functions for common server operations. """ import logging import os import random import re import socket import threading import time import sys import subprocess import mysqlsh from mysql_gadgets import MIN_MYSQL_VERSION, MAX_MYSQL_VERSION from mysql_gadgets.exceptions import (GadgetCnxInfoError, GadgetCnxError, GadgetQueryError, GadgetServerError, GadgetError) from mysql_gadgets.common.connection_parser import (parse_connection, hostname_is_ip, clean_IPv6,) from mysql_gadgets.common.logger import CustomLevelLogger from mysql_gadgets.common.tools import (get_abs_path, is_executable, run_subprocess, shell_quote) CR_SERVER_LOST = 2013 ER_OPTION_PREVENTS_STATEMENT = 1290 _FOREIGN_KEY_SET = "SET foreign_key_checks = {0}" _AUTOCOMMIT_SET = "SET AUTOCOMMIT = {0}" _GTID_ERROR = ("The server {host}:{port} does not comply to the latest GTID " "feature support. Errors:") logging.setLoggerClass(CustomLevelLogger) _LOGGER = logging.getLogger(__name__) class Secret(object): def __init__(self, s): super(Secret, self).__init__() self.value = s def __str__(self): return str(self.value) def __repr__(self): return repr(self.value) class Query(object): def __init__(self, query_string, *params): super(Query, self).__init__() self._query_string = query_string self._params = params def extend(self, query_string, *params): self._query_string += " " + query_string self._params.extend(params) def query(self): return self._query_string def params(self, mask=False): if mask: return tuple(map(lambda x: "<secret>" if isinstance(x, Secret) else x, self._params)) return tuple(map(lambda x: x.value if isinstance(x, Secret) else x, self._params)) def __str__(self): if self._params: return "{0}, with params: {1}".format(self.query(), self.params(True)) return self.query() def log(self): return self.__str__() def _to_str(value, charset="utf-8"): """Cast value to str except when None :param value: Value to be cast to str :type value: bytearray :param charset: the charset to decode the value. Used only on python 3 :type charset: string :return: value as string instance or None. :rtype: string or None """ if sys.version_info > (3, 0): # for Python 3 use bytes instead of str return None if value is None else bytes(value).decode(charset) else: return None if value is None else str(value) def get_mysqld_version(mysqld_path): """Get the version of the mysql server through the mysqld executable. :param mysqld_path: absolute path to the mysqld executable :type mysqld_path: str :return: tuple with with major, minor and release number and version string :rtype tuple((int, int, int), str) """ cmd = u"{0} --version".format(shell_quote(mysqld_path)) version_proc = run_subprocess(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=False, universal_newlines=True) output, error = version_proc.communicate() match = re.match(r'^.*mysqld.*?\s+(Ver\s+(\d+\.\d+(?:\.\d+)*).*)', output) if match: return tuple(int(n) for n in match.group(2).split('.')), match.group(1) else: error_msg = '' if error: error_msg = " Error executing '{0}': {1}".format(cmd, error) raise GadgetError( "Unable to parse version output '{0}' from mysqld executable '{1}'" ".{2}".format(output, mysqld_path, error_msg)) def is_valid_mysqld(mysqld_path): """Check if the provided mysqld is valid. Check the version of the given mysqld and return True if valid and False otherwise. :param mysqld_path: Path to mysqld to check. :type mysqld_path: string :return: True if the provided mysqld is valid, or False otherwise. :rtype: bool """ # Check if mysqld is executable and version is valid. if is_executable(mysqld_path): mysqld_ver, _ = get_mysqld_version(mysqld_path) if MIN_MYSQL_VERSION <= mysqld_ver < MAX_MYSQL_VERSION: _LOGGER.debug("Valid mysqld found with version %s: '%s'", ".".join(str(num) for num in mysqld_ver), mysqld_path) return True # mysqld is not valid. _LOGGER.debug("Invalid mysqld version (or not executable): '%s'", mysqld_path) return False class MySQLUtilsCursorResult(object): def __init__(self, result): self._result = result self.with_rows = result.has_data() self.column_names = result.column_names def fetchone(self): row = self._result.fetch_one() if row: return tuple(str(row[i]) if row[i] is not None else 'NULL' for i in range(row.length)) return None def fetchall(self): rows = [] row = self.fetchone() while row: rows.append(row) row = self.fetchone() return rows def close(self): pass def get_connection_dictionary(conn_info, ssl_dict=None): """Get the connection dictionary. Convert the given connection information into a dictionary. The method accepts one of the following types for the connection information: - dictionary containing connection information including: (user, passwd, host, port, socket) - connection string in the form: user:pass@host:port:socket - an instance of the Server class :param conn_info: Connection information to be converted. :type conn_info: dictionary or Server object or string :param ssl_dict: A dictionary with the ssl options. :type ssl_dict: dictionary :return dictionary with connection information (user, passwd, host, port, socket). :rtype: dictionary """ if conn_info is None: return conn_info if isinstance(conn_info, dict): # Not update conn_info if already has any ssl certificate. if (ssl_dict is not None and not (conn_info.get("ssl_ca", None) or conn_info.get("ssl_cert", None) or conn_info.get("ssl_key", None) or conn_info.get("ssl", None))): conn_info.update(ssl_dict) conn_val = conn_info elif isinstance(conn_info, Server): # get server's dictionary conn_val = conn_info.get_connection_values() elif isinstance(conn_info, str): # parse the string conn_val = parse_connection(conn_info, options=ssl_dict) else: raise GadgetCnxInfoError("Cannot determine connection information" " type.") return conn_val def check_hostname_alias(server1_cnx_values, server2_cnx_values): """Check to see if the servers are the same machine by host name. :param server1_cnx_values: connection values for server 1 :type server1_cnx_values: dictionary :param server2_cnx_values: connection values for server 2 :type server2_cnx_values: dictionary :return: True if both servers are the same otherwise False. :rtype: boolean """ server1 = Server({'conn_info': server1_cnx_values}) server2 = Server({'conn_info': server2_cnx_values}) return (server1.is_alias(server2.host) and int(server1.port) == int(server2.port)) def get_server(server_info, ssl_dict=None, connect=True): """Get a server instance from server connection information | string or a Server instance. The method accepts one of the following types for server_info: - dictionary containing connection information including: (user, passwd, host, port, socket) - connection string in the form: user:pass@host:port:socket or login-path:port:socket - an instance of the Server class :param server_info: Connection information :type server_info: dict | Server | str :param ssl_dict: A dictionary with the ssl certificates :type ssl_dict: dict :param connect: Attempt to connect to the server :type connect: boolean :raise GadgetCnxInfoError: if the connection information on server_info could not be parsed :raise GadgetServerError: if a connection fails. @returns a Server instance :rtype: Server """ if server_info is None: return server_info if isinstance(server_info, dict) and 'host' in server_info: # Don't update server_info if already has any ssl certificate. if (ssl_dict is not None and not (server_info.get("ssl_ca", None) or server_info.get("ssl_cert", None) or server_info.get("ssl_key", None) or server_info.get("ssl", None))): server_info.update(ssl_dict) options = {"conn_info": server_info} server = Server(options) elif isinstance(server_info, Server): server = server_info elif isinstance(server_info, str): # parse the string conn_val = parse_connection(server_info, options=ssl_dict) options = {"conn_info": conn_val} server = Server(options) else: raise GadgetCnxInfoError("Cannot determine connection information" " type.") if connect: server.connect() return server def generate_server_id(strategy='RANDOM'): """Generate a server ID based on the provided strategy. Two strategies are supported to generate a server_id: TIME and RANDOM. The TIME strategy allow to generate a server_id based on the current timestamp, ensuring that a unique server_id can be generated every second during approximately 31 years (until 2038, epoch end for the time.time() function in Unix). If different server_ids are generated during the same second they will all have the same value. Minimum value generated is 1 and maximum 999999999. The RANDOM strategy allow to generate a pseudo-random server_id, with a value between 1 and 4294967295. Two generated values have a low probability of being the same (independently of the time they are generated). Minimum value generated is 1 and maximum 4294967295. For the random generation (assuming random generation is uniformly) the probability of the same value being generated is given by: P(n, t) = 1 - t!/(t^n * (t-n)!) where t is the total number of different values that can be generated, and n is the number of values that are generated. In this case, t = 4294967295 (max number of values that can be generated), and for example the probability of generating the same id for 15, 100, and 1000 servers (n=15, n=100, and n=1000) is approximately: P(15, 4294967295) = 2.44 * 10^-8 (0.00000244 %) P(100, 4294967295) = 1.15 * 10^-6 (0.000115 %) P(1000, 4294967295) = 1.16 * 10^-4 (0.0116 %) Note: Zero is not a valid sever_id. :param strategy: Strategy used to generate the server id value. Supported values: 'RANDOM' to generate a random ID and 'TIME' to generate the ID based on the current timestamp. By default: 'RANDOM'. :type strategy: String :return: The generated server_id. :rtype: string :raises GadgetError: If an unsupported strategy is specified. """ strategy = strategy.upper() if strategy == 'RANDOM': # Generate random int between the min valid and max allowed value. # Note: 0 is not a valid server_id. return str(random.randint(1, 4294967295)) elif strategy == 'TIME': # Only the nine last digits from the time can be used to avoid # generating a value higher than the max allowed server_id value. server_id = str(int(time.time()))[-9:] # 0 is not a valid server_id, in that case return 1 (1 second later). if int(server_id) == 0: return '1' return server_id else: raise GadgetError("Invalid strategy used to generate the server_id " "(supported values: 'RANDOM' or 'TIME'): " "{0}.".format(strategy)) # pylint: disable=too-many-public-methods class Server(object): """The Server class can be used to connect to a running MySQL server. It provides several features, such as: - connect/disconnect to the server - get the server version - Retrieve a server variable - Execute a query, commit, and rollback - Return list of all databases - Read SQL statements from a file and execute - check for specific plugin support - etc. """ def __init__(self, options): """Constructor The constructor accepts one of the following types for the connection information provided trough the options parameter, more specifically for the options['conn_info']: - dictionary containing connection information including: (user, passwd, host, port, socket) - connection string in the form: user:pass@host:port:socket - an instance of the Server class :param options: Options and definitions used to create the Server object. Supported key values: conn_info a dictionary containing connection information (user, passwd, host, port, socket) role Name or role of server (e.g., server, master) verbose print extra data during operations (optional) default value = False charset Default character set for the connection. (default None) :type options: dictionary """ if options is None: options = {} if options.get("conn_info") is None: raise GadgetCnxInfoError( "Server connection information missing. The option parameter" "must contain a 'conn_info' entry.") self.verbose = options.get("verbose", False) self.db_conn = None self.host = None self.role = options.get("role", "Server") self.has_ssl = False conn_values = get_connection_dictionary(options.get("conn_info")) try: self.host = conn_values["host"] self.user = conn_values["user"] self.passwd = conn_values["passwd"] \ if "passwd" in conn_values else None self.socket = conn_values["unix_socket"] \ if "unix_socket" in conn_values else None self.port = 3306 if conn_values["port"] is not None: self.port = int(conn_values["port"]) self.charset = options.get("charset", conn_values.get("charset", None)) # Optional values self.ssl_ca = conn_values.get('ssl_ca', None) self.ssl_cert = conn_values.get('ssl_cert', None) self.ssl_key = conn_values.get('ssl_key', None) self.ssl = conn_values.get('ssl', False) if self.ssl_cert or self.ssl_ca or self.ssl_key or self.ssl: self.has_ssl = True except KeyError as err: raise GadgetCnxInfoError( "Server connection dictionary format not recognized. " "Mandatory value missing ({0}): " "{1}".format(str(err), conn_values)) self.connect_error = None # Set to TRUE when foreign key checks are ON. Check with # foreign_key_checks_enabled. self.fkeys = None self.autocommit = True # Set autocommit to True by default. self.read_only = False self.aliases = set() self.grants_enabled = None self._version = None self._version_full = None @classmethod def from_server(cls, server, conn_info=None): """ Create a new server instance from an existing one. Factory method that will allow the creation of a new server instance from an existing server. :param server: Source object used to create the new server. It must be an instance of the Server class or a subclass. :type server: Server object :param conn_info: Connection information for the new server. If provided this information will overwrite the one from the source server. :type conn_info: dictionary :return: An new instance of the Server class based on the provided source server snf (optional) connection information.. :rtype: Server object """ if isinstance(server, Server): options = {"role": server.role, "verbose": server.verbose, "charset": server.charset} if conn_info is not None and isinstance(conn_info, dict): options["conn_info"] = conn_info else: options["conn_info"] = server.get_connection_values() return cls(options) else: raise TypeError("The server argument's type is neither Server nor " "a subclass of Server") def is_alive(self): """Determine if connection to server is alive. :returns: True if server is alive (and responding) or False if an error occurred when trying to connect to the server. :rtype: boolean """ res = True try: if self.db_conn is None: res = False else: # ping and is_connected only work partially, try exec_query # to make sure connection is really alive retval = self.db_conn.is_open() if retval: self.exec_query("SELECT 1") else: res = False except Exception: # pylint: disable=W0703 res = False return res def _update_alias(self, ip_or_hostname, suffix_list): """Update list of aliases for the given IP or hostname. Gets the list of aliases for host *ip_or_hostname*. If any of them matches one of the server's aliases, then update the list of aliases (self.aliases). It also receives a list (tuple) of suffixes that can be ignored when checking if two hostnames are the same. :param ip_or_hostname: IP or hostname to test. :type ip_or_hostname: string :param suffix_list: Tuple with list of suffixes that can be ignored. :type suffix_list: list :returns: True if ip_or_hostname is a server alias, otherwise False. :rtype: boolean """ host_or_ip_aliases = self._get_aliases(ip_or_hostname) host_or_ip_aliases.add(ip_or_hostname) # Check if any of aliases matches with one the servers's aliases common_alias = self.aliases.intersection(host_or_ip_aliases) if common_alias: # There are common aliases, host is the same self.aliases.update(host_or_ip_aliases) return True else: # Check with and without suffixes no_suffix_server_aliases = set() no_suffix_host_aliases = set() for suffix in suffix_list: # Add alias with and without suffix from self.aliases for alias in self.aliases: if alias.endswith(suffix): host, _ = alias.rsplit('.', 1) no_suffix_server_aliases.add(host) no_suffix_server_aliases.add(alias) # Add alias with and without suffix from host_aliases for alias in host_or_ip_aliases: if alias.endswith(suffix): host, _ = alias.rsplit('.', 1) no_suffix_host_aliases.add(host) no_suffix_host_aliases.add(alias) # Check if there is any alias in common common_alias = no_suffix_host_aliases.intersection( no_suffix_server_aliases) if common_alias: # Same host, so update self.aliases self.aliases.update( no_suffix_host_aliases.union(no_suffix_server_aliases) ) return True return False def _get_aliases(self, host): """Gets the aliases for the given host. :param host: the host name or IP :type host: string :return: aliases for the given host. :rtype: list """ aliases = set([clean_IPv6(host)]) if hostname_is_ip(clean_IPv6(host)): # IP address try: my_host = socket.gethostbyaddr(clean_IPv6(host)) aliases.add(my_host[0]) # socket.gethostbyname_ex() does not work with ipv6 if (not my_host[0].count(":") < 1 or not my_host[0] == "ip6-localhost"): host_ip = socket.gethostbyname_ex(my_host[0]) else: addrinfo = socket.getaddrinfo(my_host[0], None) host_ip = ([socket.gethostbyaddr(addrinfo[0][4][0])], [fiveple[4][0] for fiveple in addrinfo], [addrinfo[0][4][0]]) except (socket.gaierror, socket.herror, socket.error) as err: host_ip = ([], [], []) if self.verbose: _LOGGER.warning( "IP lookup by address failed for %s, reason: %s", host, err.strerror) else: try: # server may not really exist. host_ip = socket.gethostbyname_ex(host) except (socket.gaierror, socket.herror, socket.error) as err: if self.verbose: _LOGGER.warning( "hostname: %s may not be reachable,reason: %s", host, err.strerror) return aliases aliases.add(host_ip[0]) addrinfo = socket.getaddrinfo(host, None) local_ip = None error = None for addr in addrinfo: try: local_ip = socket.gethostbyaddr(addr[4][0]) break except (socket.gaierror, socket.herror, socket.error) as err: error = err if local_ip: host_ip = ([local_ip[0]], [fiveple[4][0] for fiveple in addrinfo], [addrinfo[0][4][0]]) else: host_ip = ([], [], []) if self.verbose: _LOGGER.warning( "IP lookup by name failed for %s, reason: %s", host, error.strerror) aliases.update(set(host_ip[1])) aliases.update(set(host_ip[2])) return aliases def is_alias(self, host_or_ip): """Determine if the given host is an alias of the server host. :param host_or_ip: host or IP address to check. :type host_or_ip: string :returns: True is the given host or IP address is an alias of the server host, otherwise False. :rtype: boolean """ # List of possible suffixes suffixes = ('.local', '.lan', '.localdomain') host_or_ip = clean_IPv6(host_or_ip.lower()) # for quickness, verify in the existing aliases, if they exist. if self.aliases: if host_or_ip.lower() in self.aliases: return True else: # get the alias for the given host_or_ip return self._update_alias(host_or_ip, suffixes) # no previous aliases information # First, get the local information hostname_ = socket.gethostname() try: local_info = socket.gethostbyname_ex(hostname_) local_aliases = set([local_info[0].lower()]) # if dotted host name, take first part and use as an alias try: local_aliases.add(local_info[0].split('.')[0]) except Exception: # pylint: disable=W0703 pass local_aliases.update(['127.0.0.1', 'localhost', '::1', '[::1]']) local_aliases.update(local_info[1]) local_aliases.update(local_info[2]) local_aliases.update(self._get_aliases(hostname_)) except (socket.herror, socket.gaierror, socket.error) as err: if self.verbose: _LOGGER.warning("Unable to find aliases for hostname '%s', " "reason: %s", hostname_, str(err)) # Try with the basic local aliases. local_aliases = set(['127.0.0.1', 'localhost', '::1', '[::1]']) # Get the aliases for this server host self.aliases = self._get_aliases(self.host) # Check if this server is local for host in self.aliases.copy(): if host in local_aliases: # Is local then save the local aliases for future. self.aliases.update(local_aliases) break # Handle special suffixes in hostnames. for suffix in suffixes: if host.endswith(suffix): # Remove special suffix and attempt to match with local # aliases. host, _ = host.rsplit('.', 1) if host in local_aliases: # Is local then save the local aliases for future. self.aliases.update(local_aliases) break # Check if the given host_or_ip is alias of the server host. if host_or_ip in self.aliases: return True # Check if any of the aliases of ip_or_host is also an alias of the # host server. return self._update_alias(host_or_ip, suffixes) def user_host_exists(self, user_name, host_name): """Check if the 'user_name'@'host_name' account exists. This function checks if the specified 'user_name' and 'host_name' match an existing account on the server, considering the respective parts of the account name 'user_name'@'host_name'. Wildcard (%) matches are also taken into consideration for the 'host_name' part of the account name. For example, if an account 'myname'@'%' exists then any user with the name 'myname' will match the account independently of the host name. :param user_name: user name of the account. :type user_name: string :param host_name: hostname or IP address of the account. Note: wildcard '%' can be used. :type host_name: string :return: True if the given 'user_name' and 'host_name' match an existing account, otherwise False. :rtype: boolean :raise GadgetServerError: If an error occurs getting the user accounts information. """ res = self.exec_query(Query("SELECT host FROM mysql.user WHERE user = ? " "AND ? LIKE host", user_name, host_name)) if res: return True return False def get_connection_values(self): """Return a dictionary of connection values for the server. :return: Return the connection information for the server. :rtype: dictionary """ conn_vals = { "user": self.user, "host": self.host } if self.passwd: conn_vals["passwd"] = self.passwd if self.socket: conn_vals["unix_socket"] = self.socket if self.port: conn_vals["port"] = self.port if self.ssl_ca: conn_vals["ssl_ca"] = self.ssl_ca if self.ssl_cert: conn_vals["ssl_cert"] = self.ssl_cert if self.ssl_key: conn_vals["ssl_key"] = self.ssl_key if self.ssl: conn_vals["ssl"] = self.ssl return conn_vals def connect(self): """Connect to the server. Attempts to connect to the server according to its connection parameters. Note: This method must be called before executing statements. :raise GadgetServerError: if an error occurs during the connection. """ try: self.db_conn = self.get_connection() if self.ssl: res = self.exec_query("SHOW STATUS LIKE 'Ssl_cipher'") if res[0][1] == '': raise GadgetCnxError("Can not encrypt server connection.") except GadgetServerError: # Reset any previous value if the connection cannot be established, # before raising an exception. This prevents the use of a broken # database connection. self.db_conn = None raise self.connect_error = None self.read_only = self.show_server_variable("READ_ONLY")[0][1] def get_connection(self): """Return a new connection to the server. Attempts to connect to the server according to its connection parameters and returns a connection object. :return: The resulting MySQL connection object. :rtype: MySQLConnection object :raise GadgetCnxError: if an error occurred during the server connection process. """ try: parameters = { 'user': self.user, 'host': self.host, 'port': self.port, } if self.socket and os.name == "posix": parameters['socket'] = self.socket if self.passwd and self.passwd != "": parameters['password'] = self.passwd parameters['host'] = parameters['host'].replace("[", "") parameters['host'] = parameters['host'].replace("]", "") # Add SSL parameters ONLY if they are not None if self.ssl_ca is not None: parameters['ssl-ca'] = self.ssl_ca if self.ssl_cert is not None: parameters['ssl-cert'] = self.ssl_cert if self.ssl_key is not None: parameters['ssl-key'] = self.ssl_key if self.ssl == False: parameters['ssl-mode'] = "DISABLED" # The ca certificate is verified only if the ssl option is also # specified. if self.ssl and parameters['ssl-ca']: parameters['ssl-mode'] = "VERIFY_CA" if not "ssl-mode" in parameters: parameters['ssl-mode'] = "PREFERRED" db_conn = mysqlsh.mysql.get_classic_session(parameters) # Return MySQL connection object. return db_conn except mysqlsh.DBError as err: _LOGGER.debug("Connector Error: %s", err) raise GadgetCnxError( err.args[1], err.args[0], cause=err, server=self) except AttributeError as err: # Might be raised by mysql.connector.connect() raise GadgetCnxError(str(err), cause=err, server=self) def disconnect(self): """Disconnect from the server. """ if self.db_conn is None: raise GadgetCnxError("Cannot disconnect from a not connected" "server. You must use connect() first.") try: self.db_conn.close() except mysqlsh.DBError: # No error expected even if already disconnected, anyway ignore it. pass def get_version(self, full=False): """Return version number of the server. Get the server version. The respective instance variable is set with the result after querying the server the first time. The version is immediately returned when already known, avoiding querying the server at each time. :param full: If True return the full version information including suffixes (e.g., 5.7.14-log), otherwise only the major, minor and release number (e.g., 5.7.14) are returned. By default: False. :type full: boolean :return: List of version numbers(ints) if full = False or a string if full = True or None if an error occurs when trying to get the version information from the server. :rtype: string or int list or None """ # Return the local version value if already known. if self._version: if full: return self._version_full else: return self._version # Query the server for its version. try: res = self.show_server_variable("VERSION") if res: self._version_full = res[0][1] match = re.match(r'^(\d+\.\d+(\.\d+)*).*$', self._version_full.strip()) if match: self._version = [int(x) for x in match.group(1).split('.')] # Ensure a 3 elements list self._version = (self._version + [0])[:3] except (GadgetCnxError, GadgetQueryError): # Ignore errors and return version initialized with None. pass if full: return self._version_full else: return self._version def check_version_compat(self, t_major, t_minor, t_rel): """Checks version of the server against requested version. This method can be used to check for version compatibility. :param t_major: target server version (major) :type t_major: str or int :param t_minor: target server version (minor) :type t_minor: str or int :param t_rel: target server version (release) :type t_rel: str or int :return: True if server version is greater or equal (>=) than the specified version. False if server version is lower (<) than the specified version. :rtype: boolean """ version = self.get_version() if version is not None: return version >= [int(t_major), int(t_minor), int(t_rel)] else: return False def exec_query(self, query, options=None, exec_timeout=0): """Execute a query and return result. This is the singular method to execute queries. It should be the only method used as it contains critical error code to catch the issue with mysql.connector throwing an error on an empty result set. Notes: - It handles exception if query fails. - If 'fetch' is False in the options, the method returns the cursor instance. - By default a commit is performed at the end, unless 'commit' is set to False in the options. :param query: Query object or string with the SQL statement to execute. :type query: Query :type query: str :param options: Options to control the execution of the statement. The follow values are supported: columns Add column headings as first row. By default, False. fetch Execute the fetch as part of the operation and use a buffered cursor. By default, True. raw If True use a buffered raw cursor, meaning that all returned values are strings (i.e., not converted to the corresponding Python type). By default, True. commit Perform a commit (if needed) automatically at the end. By default, True. :type options: dictionary :param exec_timeout: Timeout value in seconds to kill the query execution if exceeded. Value must be greater than zero for this feature to be enabled. By default 0, meaning that the query will not be killed. :type exec_timeout: integer :return: List of rows (tuples) or Cursor with the result of the query. :rtype: list of tuples or Cursor object :raise GadgetCnxError: If an error occurs with the server connection or creating the cursor. :raise GadgetQueryError: If an error occurs when excuting the statement (query), fetching results or committing changes. """ if options is None: options = {} columns = options.get('columns', False) fetch = options.get('fetch', True) raw = options.get('raw', True) do_commit = options.get('commit', True) if isinstance(query, str): query = Query(query) # Guard for connect() prerequisite if not self.db_conn: raise GadgetCnxError( "You must call connect before executing a query.", server=self) # Execute query, handling parameters. q_killer = None try: if exec_timeout > 0: if query.params(): raise NotImplementedError() # Spawn thread to kill query if timeout is reached. # Note: set it as daemon to avoid waiting for it on exit. q_killer = QueryKillerThread(self, query.query(), exec_timeout) q_killer.daemon = True q_killer.start() # Execute query. cur = None _LOGGER.debug("MySQL query: " + query.log()) if query.params(): cur = MySQLUtilsCursorResult( self.db_conn.run_sql(query.query(), query.params())) else: cur = MySQLUtilsCursorResult( self.db_conn.run_sql(query.query())) except mysqlsh.DBError as err: if cur: cur.close() if err.args[0] == CR_SERVER_LOST and exec_timeout > 0: # If the connection is killed (because the execution timeout is # reached), then it attempts to re-establish it (to execute # further queries) and raise a specific exception to track this # event. # CR_SERVER_LOST = Errno 2013 Lost connection to MySQL server # during query. self.connect() raise GadgetQueryError("Timeout executing query", query.log(), errno=err.args[0], cause=err, server=self) else: raise GadgetQueryError("Query failed. {0}".format(err), query.log(), errno=err.args[0], cause=err, server=self) except Exception as err: if cur: cur.close() raise GadgetQueryError("Unknown error: {0}".format(err), query.log(), errno=0, cause=err, server=self) finally: # Stop query killer thread if alive. if q_killer and q_killer.is_alive(): q_killer.stop() # Fetch rows (only if available or fetch = True). if cur.with_rows: if fetch or columns: try: results = cur.fetchall() if columns: col_headings = cur.column_names col_names = [] for col in col_headings: col_names.append(col) results = col_names, results except mysqlsh.DBError as err: raise GadgetQueryError( "Error fetching all query data: {0}".format(err), query.log(), errno=err.args[0], cause=err, server=self) finally: cur.close() return results else: # Return cursor to fetch rows elsewhere (fetch = false). return cur else: # No results (not a SELECT) try: if do_commit: self.db_conn.run_sql("commit") except mysqlsh.DBError as err: raise GadgetQueryError( "Error performing commit: {0}".format(err), query.log(), errno=err.args[0], cause=err, server=self) finally: cur.close() return cur def commit(self): """Perform a COMMIT. :raise GadgetCnxError: If connection was not previously established. """ # Guard for connect() prerequisite if not self.db_conn: raise GadgetCnxError( "You must call connect before commit.", server=self) self.db_conn.run_sql("commit") def rollback(self): """Perform a ROLLBACK. :raise GadgetCnxError: If connection was not previously established. """ # Guard for connect() prerequisite if not self.db_conn: raise GadgetCnxError( "You must call connect before rollback.", server=self) self.db_conn.run_sql("rollback") def show_server_variable(self, variable): """Get the variable information using SHOW VARIABLES statement. Return one or more rows from the SHOW VARIABLES statement according to the specified variable parameter. :param variable: The variable name or wildcard string to match the returned variable information. :type variable: string :returns: Variables names and respective value matching the specified variable parameter. :rtype: list of tuples :raise GadgetServerError: If an error occur getting the variable value. """ return self.exec_query(Query("SHOW VARIABLES LIKE ?", variable)) def set_variable(self, var_name, var_value, var_type="session"): """Set server variable using the SET statement. This function sets the value of system variables using the SET statement. :param var_name: Name of the variable to set. :type var_name: str :param var_value: value to which we want to set the variable. :type var_value: str :param var_type: Type of the variable ('session', 'global', 'persist' 'persist_only'). By default session is used. :type var_type: str :raises GadgetDBError: if an invalid var_type is provided as argument. """ if var_type.lower() in ('global', 'session', 'persist', 'persist_only'): var_type = '{0}.'.format(var_type) # Add dot (.) else: raise GadgetError("Invalid variable type: {0}. Supported types: " "'global' and 'session'.".format(var_type)) # Execute SET @@var_type.var_name. self.exec_query("SET @@{0}{1}={2}".format(var_type, var_name, var_value)) def select_variable(self, var_name, var_type=None): """Get server system variable value using SELECT statement. This function displays the value of system variables using the SELECT statement. This can be used as a workaround for variables with very long values, as SHOW VARIABLES is subject to a version-dependent display-width limit. Note: Some variables may not be available using SELECT @@var_name, in such cases use SHOW VARIABLES LIKE 'var_name'. :param var_name: Name of the variable to display. :type var_name: string :param var_type: Type of the variable ('session' or 'global'). By default None (no type is used), meaning that the session value is returned if it exists and the global value otherwise. :type var_type: 'session', 'global', '', or None :return: value for the given server system variable. :rtype: string :raise GadgetServerError: If an unsupported variable type is specified. :raise GadgetQueryError: If the variable does not exist in the server. """ if var_type is None: var_type = '' elif var_type.lower() in ('global', 'session', ''): var_type = '{0}.'.format(var_type) # Add dot (.) else: raise GadgetServerError( "Invalid variable type: {0}. Supported types: " "'global' and 'session'.".format(var_type), server=self) # Execute SELECT @@[var_type.]var_name. # Note: An error is issued if the given variable is not known. res = self.exec_query("SELECT @@{0}{1}".format(var_type, var_name)) return res[0][0] def has_default_value(self, var_name): """Check if the variable has the default value or was already change. Return a boolean value indicating if the variable is set with the compiled default value (true), or if it was already changed explicitly by the user somehow (false), i.e. variable changed with the SET statement, a command line option, or the configuration file. NOTE: This method requires the performance_schema to be enabled. :param var_name: Name of the target variable to check. :type var_name: string :return: True if the variable value is the compiled default one, or False otherwise (meaning that the variable value was already changed by the user). :rtype: boolean """ res = self.exec_query(Query("SELECT variable_source " "FROM performance_schema.variables_info " "WHERE variable_name=?", var_name)) if res[0][0] == 'COMPILED': return True else: return False def flush_logs(self, log_type=None): """Execute the FLUSH [log_type] LOGS statement. Reload internal logs cache and closes and reopens all log files, or only of the specified log_type. Note: The log_type option is available starting from MySQL 5.5.3. :param log_type: Type of the log files to be flushed. Supported values: BINARY, ENGINE, ERROR, GENERAL, RELAY, SLOW. :type log_type: string :raise GadgetServerError: If an error occurs when executing the FLUSH LOGS statement. """ if log_type: self.exec_query("FLUSH {0} LOGS".format(log_type)) else: self.exec_query("FLUSH LOGS") def supports_gtid(self): """Check if the server supports GTIDs. :return: True if GTID is supported and turned on and False if supported but not enabled. :rtype: boolean :raise GadgetServerError: If GTID is not supported or the GTID mode cannot be obtained. """ # Check server for GTID support version_ok = self.check_version_compat(5, 6, 5) if not version_ok: raise GadgetServerError("GTIDs are only supported starting from" "MySQL 5.6.5", server=self) try: res = self.exec_query("SELECT @@GLOBAL.GTID_MODE") except (GadgetCnxError, GadgetQueryError) as err: raise GadgetServerError( "Unable to get GTID_MODE value: {0}".format(err.errmsg), cause=err, server=self) # Return result if res[0][0] == 'ON': return True elif res[0][0] == 'OFF': return False else: raise GadgetServerError( "Unexpected value for @@GLOBAL.GTID_MODE: {0}." "Expected: 'ON' or 'OFF'.".format(res[0][0]), server=self) def get_gtid_executed(self, skip_gtid_check=True): """Get the executed GTID set of the server. This function retrieves the (current) GTID_EXECUTED set of the server. :param skip_gtid_check: Flag indicating if the check for GTID support will be skipped or not. By default 'True' (check is skipped). :returns a string with the GTID_EXECUTED set for this server. :rtype str :raises GadgetError: if GTIDs are not supported or not enabled. """ if not skip_gtid_check: # Check server for GTID support. gtid_support = self.supports_gtid() if not gtid_support: raise GadgetServerError("Global Transaction IDs are not " "supported.", server=self) # Get GTID_EXECUTED. try: return self.exec_query("SELECT @@GLOBAL.GTID_EXECUTED")[0][0] except GadgetQueryError: if skip_gtid_check: # Query likely failed because GTIDs are not supported, # therefore skip error in this case. return "" else: # If GTID check is not skipped re-raise exception. raise except IndexError: # If no rows are returned by query then return an empty string. return "" def supports_plugin(self, plugin, state='ACTIVE'): """Check if the given plugin is supported. Check if the server supports the specified plugin. Return True if plugin is installed and active. :param plugin: Name of plugin to check :type plugin: string :param state: the expected plugin state to check, by default ACTIVE. :type state: string :returns: True if the plugin is supported and it has the given state, False otherwise. :rtype: boolean :raise GadgetServerError: If an error occurs when checking for the plugin support. """ _PLUGIN_QUERY = Query( "SELECT PLUGIN_NAME, PLUGIN_STATUS " "FROM INFORMATION_SCHEMA.PLUGINS WHERE PLUGIN_NAME LIKE ?", "{0}%".format(plugin)) res = self.exec_query(_PLUGIN_QUERY) if not res: # plugin not found. _LOGGER.debug("Plugin %s is not installed", plugin) return False elif res[0][1].upper() != state.upper(): # The state is not the desired. _LOGGER.debug("Plugin %s has state: %s and not the expected: %s", plugin, res[0][1], state) return False return True def get_all_databases(self, ignore_internal_dbs=True): """Get all databases from the server. Retrieve the list of all databases on the server, except for internal databases (e.g., INFORMATION_SCHEMA and PERFORMANCE_SCHEMA) if the 'ignore_internal_dbs' parameter is set to True. Note: New internal database 'sys' added by default for MySQL 5.7.7+. :param ignore_internal_dbs: Ignore internal databases. :type ignore_internal_dbs: boolean :returns: Result with the name of all the databases in the server. :rtype: list of tuples :raise GadgetServerError: If an error occurs when getting the list of all databases. """ if ignore_internal_dbs: _GET_DATABASES = """ SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME != 'INFORMATION_SCHEMA' AND SCHEMA_NAME != 'PERFORMANCE_SCHEMA' """ # Starting from MySQL 5.7.7, sys schema is installed by default. if self.check_version_compat(5, 7, 7): _GET_DATABASES = ("{0} AND SCHEMA_NAME != 'sys'" "".format(_GET_DATABASES)) else: _GET_DATABASES = """ SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA """ return self.exec_query(_GET_DATABASES) def read_and_exec_sql(self, input_file, verbose=False): """Read an input file containing SQL statements and execute them. :param input_file: The full path to the SQL file. :type input_file: string :param verbose: Log the read statements. By default, False. :type verbose: boolean :raise GadgetServerError: If an error occurs when executing the statements in the SQL file. """ with open(input_file) as f_input: while True: cmd = f_input.readline() if not cmd: break if len(cmd) > 1: if cmd[0] != '#': if verbose: _LOGGER.debug("%s", cmd) query_options = { 'fetch': False } self.exec_query(cmd, query_options) def binlog_enabled(self): """Check binary logging status. Check if binary logging is enabled on the server based on the value of the 'log_bin' variable. :return: False if binary logging is disabled and True otherwise. :rtype: boolean :raise GadgetServerError: If an error occur when trying to get the value of the 'log_bin' variable. """ try: res = self.show_server_variable("log_bin") except GadgetServerError as err: raise GadgetServerError("Cannot get value of 'log_bin' variable. " "{0}".format(err.errmsg), cause=err, server=self) if not res: raise GadgetServerError( "No value returned for 'log_bin' variable.", server=self) if res[0][1] in ("OFF", "0"): return False return True def toggle_global_read_lock(self, enable=True): """ Enable or disable read-only mode on the server. Note: user must have SUPER privilege :param enable: Indicate if read-only mode will be enabled (True) or disabled (False). If True (default) then flush all tables with read lock and set the [super_]read_only to 'ON'. If False then set [super_]read_only mode to 'OFF' and unlock all tables. :type enable: boolean """ # Starting with version 5.7.8 MySQL has a new super_read_only system # variable has_super_read_only = self.check_version_compat(5, 7, 8) var_name = "super_read_only" if has_super_read_only else "read_only" if enable: self.exec_query("FLUSH TABLES WITH READ LOCK") self.set_variable(var_name, "ON", "global") else: self.set_variable(var_name, "OFF", "global") self.exec_query("UNLOCK TABLES") def toggle_binlog(self, action='disable'): """Enable or disable binary logging. Note: User must have SUPER privilege. :param action: if 'disable' then turn off the binary logging. if 'enable' then turn on binary logging. If none of the previous action is specified then nothing is done (no action). :type action: 'disable' or 'enable' :raise GadgetServerError: If an error occur when setting the SQL_LOG_BIN value. """ if action.lower() == 'disable': self.exec_query("SET SQL_LOG_BIN=0") elif action.lower() == 'enable': self.exec_query("SET SQL_LOG_BIN=1") def get_server_id(self): """Retrieve the server ID. :return: Value of the 'server_id' variable. :rtype: integer :raise GadgetServerError: If an error occurs when trying to retrieve the server ID. """ try: res = self.show_server_variable("server_id") except GadgetServerError as err: raise GadgetServerError("Cannot retrieve 'server_id'. " "{0}".format(err.errmsg), cause=err, server=self) return int(res[0][1]) def get_server_uuid(self): """Retrieve the UUID of the server. :return: Value of the 'server_uuid' variable. If not available None is returned. :rtype: string :raises GadgetServerError: If an errors occurs when trying to retrieve the server UUID. """ try: res = self.show_server_variable("server_uuid") if res is None or res == []: return None except GadgetServerError as err: raise GadgetServerError("Cannot retrieve 'server_uuid': " "{0}".format(err.errmsg), cause=err, server=self) return res[0][1] def grant_tables_enabled(self): """Check if grant tables is enabled on the server. In other words, this functions checks if the privileges system is enabled on the server. Grant tables might be disabled if the --skip-grant-tables option is used when starting the server. :return: True if grant tables (privileges system) is enabled otherwise False (disabled). :rtype: boolean """ if self.grants_enabled is None: try: self.exec_query("SHOW GRANTS FOR 'snuffles'@'host'") self.grants_enabled = True except (GadgetCnxError, GadgetQueryError) as err: if (err.args[0] == ER_OPTION_PREVENTS_STATEMENT and "--skip-grant-tables" in err.errmsg): self.grants_enabled = False # Ignore other errors as they are not pertinent to the check else: self.grants_enabled = True return self.grants_enabled def get_server_binlogs_list(self, include_size=False): """Get the binary log file names listed on a server. Obtains the binlog file names available on the server by using the 'SHOW BINARY LOGS' statement, returning these file names as a list. :param include_size: Indicate if the returning list shall include the size of the file. :type include_size: boolean :return: List with the binary log file names available on the server. :rtype: list :raises GadgetServerError: If an errors occurs when executing SHOW BINARY LOGS. """ res = self.exec_query("SHOW BINARY LOGS") server_binlogs = [] for row in res: if include_size: server_binlogs.append(row) else: server_binlogs.append(row[0]) return server_binlogs def is_plugin_installed(self, plugin_name, is_active=False, silence_warnings=False): """Test if the given plugin is installed/loaded in this server. :param plugin_name: The name of the plugin to load. :type plugin_name: str. :param is_active: If True verifies the plugin is also in 'ACTIVE' state. :type is_active: bool :param silence_warnings: If True avoids logging warning messages. :type silence_warnings: bool (False by default) :return: True if the plugin is installed otherwise False. :rtype: bool. """ res = self.exec_query("show plugins") for row in res: if plugin_name in row[0]: if is_active: return True if "ACTIVE" in row[1] else False else: return True if not silence_warnings: _LOGGER.warning("The %s plugin has not been " "installed/loaded in %s", plugin_name, self) return False def install_plugin(self, plugin_name): """Attempts to install/load the given plugin in this server. :param plugin_name: The name of the plugin to load. :type plugin_name: str. :raise GadgetServerError: If the GR plugin could not be loaded. :return: True if the plugin is successfully loaded or was already loaded, else an exception is raised. :rtype: boolean :raise GadgetServerError: If the plugin could not be installed. """ msg = "Initializing {0} plugin on {1}".format(plugin_name, self) _LOGGER.info(msg) if self.is_plugin_installed(plugin_name, silence_warnings=True): return True else: try: if "WIN" in self.select_variable("version_compile_os").upper(): ext = ".dll" else: ext = ".so" self.exec_query("INSTALL PLUGIN {plugin_name} SONAME " "'{plugin_name}{ext}'" "".format(plugin_name=plugin_name, ext=ext)) _LOGGER.debug("The %s plugin has been successfully install " "in server: %s", plugin_name, self) except GadgetQueryError as err: if "already exists" in err.errmsg: _LOGGER.debug("The %s plugin is already installed: %s" "", plugin_name, err.errmsg) else: _LOGGER.error("An error was found trying to install the " "%s plugin: %s", plugin_name, err.errmsg) raise GadgetServerError("The {0} plugin could not be " "loaded in the server {1}" "".format(plugin_name, self), cause=err) return True def uninstall_plugin(self, plugin_name): """Attempts to uninstall the given plugin in this server. :param plugin_name: The name of the plugin to unload. :type plugin_name: str. :raise GadgetServerError: If the plugin could not be uninstalled. """ try: self.exec_query("UNINSTALL PLUGIN {plugin}" "".format(plugin=plugin_name)) except GadgetQueryError as err: if "does not exist" in err.errmsg: _LOGGER.debug("The %s plugin is not installed: %s" "", plugin_name, err.errmsg) else: _LOGGER.error("An error was found trying to uninstall the %s " "plugin: %s", plugin_name, err.errmsg) raise GadgetServerError("The {0} plugin could not be " "uninstalled in the server {1}" "".format(plugin_name, self), cause=err) def start_plugin(self, plugin_name): """Starts the given plugin :param plugin_name: The name of the plugin to load. :type plugin_name: str. """ self.exec_query("START {plugin}".format(plugin=plugin_name)) def stop_plugin(self, plugin_name): """Stops the given plugin :param plugin_name: The name of the plugin to load. :type plugin_name: str. """ self.exec_query("STOP {plugin}".format(plugin=plugin_name)) def __str__(self): """String representation of the class Server :return: representation the server with information of the host and port. :rtype: string """ if self.socket and os.name == "posix": return "'{0}:{1}'".format(self.host, self.socket) else: return "'{0}:{1}'".format(self.host, self.port) class QueryKillerThread(threading.Thread): """Class to run a thread to kill an executing query. This class is used to spawn a thread than will kill the execution (connection) of a query upon reaching a given timeout. """ def __init__(self, server, query, timeout): """Constructor. :param server: Server instance where the target query is executed. :type server: Server object :param query: Target query to kill. :type query: string :param timeout: Timeout value in seconds used to kill the query when reached. :type timeout: integer """ threading.Thread.__init__(self) self._stop_event = threading.Event() self._query = query self._timeout = timeout self._server = server self._connection = server.get_connection() server.get_version() def run(self): """Main execution of the query killer thread. Stop the thread if instructed as such """ connector_error = None # Kill the query connection upon reaching the given execution timeout. while not self._stop_event.is_set(): # Wait during the defined time. self._stop_event.wait(self._timeout) # If the thread was asked to stop during wait, it does not try to # kill the query. if not self._stop_event.is_set(): try: cur = None # Get process information from threads table when available # (for versions > 5.6.1), since it does not require a mutex # and has minimal impact on server performance. cur = MySQLUtilsCursorResult(self._connection.run_sql( "SELECT processlist_id " "FROM performance_schema.threads" " WHERE processlist_command='Query'" " AND processlist_info='{0}'".format(self._query))) result = cur.fetchall() try: process_id = result[0][0] except IndexError: # No rows are returned if the query ended in the # meantime. process_id = None # Kill the connection associated to que process id. # Note: killing the query will not work with # connector-python,since it will hang waiting for the # query to return. if process_id: self._connection.run_sql("KILL {0}".format(process_id)) except mysqlsh.DBError as err: # Hold error to raise at the end. connector_error = err finally: # Close cursor if available. if cur: cur.close() # Stop this thread. self.stop() # Close connection. try: self._connection.disconnect() except mysqlsh.DBError: # Only raise error if no previous error has occurred. if not connector_error: raise finally: # Raise any previous error that already occurred. if connector_error is not None: # pylint: disable=E0702 raise connector_error def stop(self): """Stop the thread. Set the event flag for the thread to stop as soon as possible. """ self._stop_event.set() class LocalErrorLog(object): """This class can be used to read the local Error log file. Note: In the case the log_error variable is set to "stderr", then it will not be possible to retrieve the messages. """ def __init__(self, server, raise_error=False): """Constructor Create the LocalErrorLog object setting the server to retrieve messages logged to its error log file. :param server: The server to retrieve messages from the error log. :type server: mysql_gadgets.common.server.Server :raise GadgetServerError: If the "log_error" var is set to stderr instead of a file. """ self.log_error = server.select_variable("log_error") if self.log_error == "stderr": if raise_error: raise GadgetServerError("The log error is set to stderr") self._log_file = None else: datadir = server.select_variable("datadir") self._log_file = get_abs_path(self.log_error, datadir) def get_size(self): """Get the current size of the log error file. Returns the current size of the server log error file. This method can be used as a starting point to read from the file. :return: Size of the error log file. :rtype: int """ if self._log_file: return os.stat(self._log_file).st_size return None def read(self, offset, errors_only=True): """Reads the Error log file. :param offset: Starting position. :type offset: int :param errors_only: If the messages returned should be "ERROR" messages only, By default True. :type errors_only: boolean :return: Messages logged at the logging file. :rtype: str """ if self._log_file: with open(self._log_file, "r") as e_file: if offset: e_file.seek(offset) result = [] for line in e_file: if errors_only and "[ERROR]" not in line: continue result.append(line) return "".join(result) return ""