-
Notifications
You must be signed in to change notification settings - Fork 2
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
fix: mysql support #115
fix: mysql support #115
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -16,6 +16,8 @@ | |
|
||
from typing import TYPE_CHECKING | ||
|
||
import mysql | ||
|
||
from aws_wrapper.errors import QueryTimeoutError | ||
|
||
if TYPE_CHECKING: | ||
|
@@ -61,10 +63,13 @@ class PgExceptionHandler(ExceptionHandler): | |
def is_network_exception(self, error: Optional[Exception] = None, sql_state: Optional[str] = None) -> bool: | ||
if isinstance(error, QueryTimeoutError) or isinstance(error, ConnectionTimeout): | ||
return True | ||
if sql_state is None: | ||
error_sql_state = getattr(error, "sqlstate") | ||
if error_sql_state is not None: | ||
sql_state = error_sql_state | ||
|
||
if sql_state: | ||
if sql_state in self._NETWORK_ERRORS: | ||
return True | ||
if sql_state is not None and sql_state in self._NETWORK_ERRORS: | ||
return True | ||
|
||
if isinstance(error, OperationalError): | ||
if len(error.args) == 0: | ||
|
@@ -81,6 +86,12 @@ def is_login_exception(self, error: Optional[Exception] = None, sql_state: Optio | |
if isinstance(error, InvalidAuthorizationSpecification) or isinstance(error, InvalidPassword): | ||
return True | ||
|
||
if sql_state is None and hasattr(error, "sqlstate") and error.sqlstate is not None: | ||
sql_state = error.sqlstate | ||
|
||
if sql_state is not None and sql_state in self._ACCESS_ERRORS: | ||
return True | ||
|
||
if isinstance(error, OperationalError): | ||
if len(error.args) == 0: | ||
return False | ||
|
@@ -91,10 +102,62 @@ def is_login_exception(self, error: Optional[Exception] = None, sql_state: Optio | |
or self._PAM_AUTHENTICATION_FAILED_MSG in error_msg: | ||
return True | ||
|
||
if sql_state: | ||
if sql_state in self._ACCESS_ERRORS: | ||
return False | ||
|
||
|
||
class MySQLExceptionHandler(ExceptionHandler): | ||
_PAM_AUTHENTICATION_FAILED_MSG = "PAM authentication failed" | ||
_UNAVAILABLE_CONNECTION = "MySQL Connection not available" | ||
|
||
_NETWORK_ERRORS: List[int] = [ | ||
2001, # Can't create UNIX socket | ||
2002, # Can't connect to local MySQL server through socket | ||
2003, # Can't connect to MySQL server | ||
2004, # Can't create TCP/IP socket | ||
2006, # MySQL server has gone away | ||
2012, # Error in server handshake | ||
2013, # unexpected error | ||
2026, # SSL connection error | ||
2055, # Lost connection to MySQL server | ||
] | ||
|
||
def is_network_exception(self, error: Optional[Exception] = None, sql_state: Optional[str] = None) -> bool: | ||
if isinstance(error, QueryTimeoutError): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Does MySQL have a ConnectionTimeoutError that we should check for like PG? Or is the connection timeout caught in the logic below? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. |
||
return True | ||
|
||
if sql_state is None: | ||
if hasattr(error, "sqlstate"): | ||
error_sql_state = getattr(error, "sqlstate") | ||
if error_sql_state is not None: | ||
sql_state = error_sql_state | ||
|
||
if sql_state is not None: | ||
if sql_state.startswith("08") or sql_state.startswith("HY"): | ||
# Connection exceptions may also be returned as a generic error | ||
# e.g. 2013 (HY000): Lost connection to MySQL server during query | ||
return True | ||
|
||
if isinstance(error, mysql.connector.errors.OperationalError): | ||
if error.errno in self._NETWORK_ERRORS: | ||
return True | ||
if error.msg is not None and self._UNAVAILABLE_CONNECTION in error.msg: | ||
return True | ||
|
||
if len(error.args) == 1: | ||
return self._UNAVAILABLE_CONNECTION in error.args[0] | ||
|
||
return False | ||
|
||
def is_login_exception(self, error: Optional[Exception] = None, sql_state: Optional[str] = None) -> bool: | ||
if sql_state is None: | ||
if hasattr(error, "sqlstate"): | ||
error_sql_state = getattr(error, "sqlstate") | ||
if error_sql_state is not None: | ||
sql_state = error_sql_state | ||
|
||
if "28000" == sql_state: | ||
return True | ||
|
||
return False | ||
|
||
|
||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -25,8 +25,6 @@ | |
from logging import getLogger | ||
from typing import Any, Callable, Dict, List, Optional, Set | ||
|
||
from psycopg import OperationalError | ||
|
||
from aws_wrapper.errors import (AwsWrapperError, FailoverFailedError, | ||
FailoverSuccessError, | ||
TransactionResolutionUnknownError) | ||
|
@@ -55,6 +53,16 @@ class FailoverPlugin(Plugin): | |
"force_connect", | ||
"notify_host_list_changed"} | ||
|
||
_METHODS_REQUIRE_UPDATED_TOPOLOGY: Set[str] = { | ||
"Connection.commit", | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. how did we come up with this method list, and why is it necessary? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Mysql specific thing awslabs/aws-mysql-jdbc#364 |
||
"Connection.autocommit", | ||
"Connection.autocommit_setter", | ||
"Connection.rollback", | ||
"Connection.cursor", | ||
"Cursor.callproc", | ||
"Cursor.execute" | ||
} | ||
|
||
def __init__(self, plugin_service: PluginService, props: Properties): | ||
self._plugin_service = plugin_service | ||
self._properties = props | ||
|
@@ -134,7 +142,8 @@ def execute(self, target: type, method_name: str, execute_func: Callable, *args: | |
self._invalid_invocation_on_closed_connection() | ||
|
||
try: | ||
self._update_topology(False) | ||
if self._requires_update_topology(method_name): | ||
self._update_topology(False) | ||
return execute_func() | ||
except Exception as ex: | ||
msg = Messages.get_formatted("FailoverPlugin.DetectedException", str(ex)) | ||
|
@@ -375,9 +384,6 @@ def _should_exception_trigger_connection_switch(self, ex: Exception) -> bool: | |
logger.debug(Messages.get_formatted("FailoverPlugin.FailoverDisabled")) | ||
return False | ||
|
||
if isinstance(ex, OperationalError): | ||
return True | ||
|
||
return self._plugin_service.is_network_exception(ex) | ||
|
||
@staticmethod | ||
|
@@ -401,6 +407,7 @@ def _is_node_still_valid(node: str, changes: Dict[str, Set[HostEvent]]): | |
def _can_direct_execute(method_name): | ||
# TODO: adjust method names to proper python method names | ||
return method_name == "Connection.close" or \ | ||
method_name == "Cursor.close" or \ | ||
method_name == "Connection.abort" or \ | ||
method_name == "Connection.isClosed" | ||
|
||
|
@@ -412,6 +419,9 @@ def _allowed_on_closed_connection(method_name: str): | |
method_name == "Connection.getSchema" or \ | ||
method_name == "Connection.getTransactionIsolation" | ||
|
||
def _requires_update_topology(self, method_name: str): | ||
return method_name in FailoverPlugin._METHODS_REQUIRE_UPDATED_TOPOLOGY | ||
|
||
|
||
class FailoverPluginFactory(PluginFactory): | ||
def get_instance(self, plugin_service: PluginService, props: Properties) -> Plugin: | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Temporary change to run integration tests