diff --git a/.github/workflows/integration_tests.yml b/.github/workflows/integration_tests.yml index 937545b81..e8c99cdea 100644 --- a/.github/workflows/integration_tests.yml +++ b/.github/workflows/integration_tests.yml @@ -11,14 +11,67 @@ permissions: contents: read # This is required for actions/checkout jobs: - build-integration-tests: - name: Run Integration Tests + lts-integration-tests: + name: Run LTS Integration Tests runs-on: ubuntu-latest strategy: fail-fast: false matrix: python-version: [ "3.8", "3.11" ] - engine-version: [ "lts", "latest"] + environment: [ "mysql", "pg" ] + + steps: + - name: 'Clone repository' + uses: actions/checkout@v4 + + - name: 'Set up JDK 8' + uses: actions/setup-java@v4 + with: + distribution: 'corretto' + java-version: 8 + + - name: Install poetry + shell: bash + run: | + pipx install poetry==1.8.2 + poetry config virtualenvs.prefer-active-python true + + - name: Install dependencies + run: poetry install + + - name: 'Configure AWS Credentials' + uses: aws-actions/configure-aws-credentials@v4 + with: + role-to-assume: arn:aws:iam::${{ secrets.AWS_ACCOUNT_ID }}:role/${{ secrets.AWS_DEPLOY_ROLE }} + role-session-name: python_integration_tests + role-duration-seconds: 21600 + aws-region: ${{ secrets.AWS_DEFAULT_REGION }} + + - name: 'Run LTS Integration Tests' + run: | + ./gradlew --no-parallel --no-daemon test-python-${{ matrix.python-version }}-${{ matrix.environment }} --info + env: + RDS_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }} + RDS_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }} + AURORA_MYSQL_DB_ENGINE_VERSION: lts + AURORA_PG_ENGINE_VERSION: lts + + - name: 'Archive LTS results' + if: always() + uses: actions/upload-artifact@v4 + with: + name: pytest-integration-report-${{ matrix.python-version }}-${{ matrix.environment }}-lts + path: ./tests/integration/container/reports + retention-days: 5 + + latest-integration-tests: + name: Run Latest Integration Tests + runs-on: ubuntu-latest + needs: lts-integration-tests + strategy: + fail-fast: false + matrix: + python-version: [ "3.8", "3.11" ] environment: ["mysql", "pg"] steps: @@ -48,19 +101,19 @@ jobs: role-duration-seconds: 21600 aws-region: ${{ secrets.AWS_DEFAULT_REGION }} - - name: 'Run Integration Tests' + - name: 'Run Latest Integration Tests' run: | ./gradlew --no-parallel --no-daemon test-python-${{ matrix.python-version }}-${{ matrix.environment }} --info env: RDS_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }} RDS_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }} - AURORA_MYSQL_DB_ENGINE_VERSION: ${{ matrix.engine-version }} - AURORA_PG_ENGINE_VERSION: ${{ matrix.engine-version }} + AURORA_MYSQL_DB_ENGINE_VERSION: latest + AURORA_PG_ENGINE_VERSION: latest - - name: 'Archive results' + - name: 'Archive Latest results' if: always() uses: actions/upload-artifact@v4 with: - name: pytest-integration-report-${{ matrix.python-version }}-${{ matrix.environment }}-${{ matrix.engine-version }} + name: pytest-integration-report-${{ matrix.python-version }}-${{ matrix.environment }}-latest path: ./tests/integration/container/reports retention-days: 5 diff --git a/aws_advanced_python_wrapper/driver_dialect.py b/aws_advanced_python_wrapper/driver_dialect.py index a836e9ff5..c9df891a8 100644 --- a/aws_advanced_python_wrapper/driver_dialect.py +++ b/aws_advanced_python_wrapper/driver_dialect.py @@ -38,10 +38,11 @@ class DriverDialect(ABC): Driver dialects help the driver-agnostic AWS Python Driver interface with the driver-specific functionality of the underlying Python Driver. """ _QUERY = "SELECT 1" + _ALL_METHODS = "*" - _executor: ClassVar[Executor] = ThreadPoolExecutor() + _executor: ClassVar[Executor] = ThreadPoolExecutor(thread_name_prefix="DriverDialectExecutor") _dialect_code: str = DriverDialectCodes.GENERIC - _network_bound_methods: Set[str] = {"*"} + _network_bound_methods: Set[str] = {_ALL_METHODS} _read_only: bool = False _autocommit: bool = False _driver_name: str = "Generic" @@ -127,7 +128,7 @@ def execute( *args: Any, exec_timeout: Optional[float] = None, **kwargs: Any) -> Cursor: - if method_name not in self._network_bound_methods: + if DriverDialect._ALL_METHODS not in self.network_bound_methods and method_name not in self.network_bound_methods: return exec_func() if exec_timeout is None: diff --git a/aws_advanced_python_wrapper/failover_plugin.py b/aws_advanced_python_wrapper/failover_plugin.py index b959372b7..c4cb53740 100644 --- a/aws_advanced_python_wrapper/failover_plugin.py +++ b/aws_advanced_python_wrapper/failover_plugin.py @@ -351,22 +351,26 @@ def _invalidate_current_connection(self): """ conn = self._plugin_service.current_connection if conn is None: - return + return None + + driver_dialect = self._plugin_service.driver_dialect if self._plugin_service.is_in_transaction: self._plugin_service.update_in_transaction(True) try: + driver_dialect.execute("Connection.rollback", lambda: conn.rollback()) conn.rollback() except Exception: pass - driver_dialect = self._plugin_service.driver_dialect - if driver_dialect is not None and not driver_dialect.is_closed(conn): + if not driver_dialect.is_closed(conn): try: - conn.close() + return driver_dialect.execute("Connection.close", lambda: conn.close()) except Exception: pass + return None + def _invalid_invocation_on_closed_connection(self): if not self._closed_explicitly: self._is_closed = False diff --git a/aws_advanced_python_wrapper/pg_driver_dialect.py b/aws_advanced_python_wrapper/pg_driver_dialect.py index 51333a759..fbe441f39 100644 --- a/aws_advanced_python_wrapper/pg_driver_dialect.py +++ b/aws_advanced_python_wrapper/pg_driver_dialect.py @@ -48,6 +48,7 @@ class PgDriverDialect(DriverDialect): "Connection.is_read_only", "Connection.set_read_only", "Connection.rollback", + "Connection.close", "Connection.cursor", "Cursor.close", "Cursor.callproc", diff --git a/aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties b/aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties index c7be1412b..65f16806b 100644 --- a/aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties +++ b/aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties @@ -382,8 +382,6 @@ RoundRobinHostSelector.RoundRobinInvalidHostWeightPairs= [RoundRobinHostSelector WeightedRandomHostSelector.WeightedRandomInvalidHostWeightPairs= [WeightedRandomHostSelector] The provided host weight pairs have not been configured correctly. Please ensure the provided host weight pairs is a comma separated list of pairs, each pair in the format of :. Weight values must be an integer greater than or equal to the default weight value of 1. Weight pair: '{}' WeightedRandomHostSelector.WeightedRandomInvalidDefaultWeight=[WeightedRandomHostSelector] The provided default weight value is not valid. Weight values must be an integer greater than or equal to 1. -SlidingExpirationCache.CleaningUp=[SlidingExpirationCache] Cleaning up... - SqlAlchemyPooledConnectionProvider.PoolNone=[SqlAlchemyPooledConnectionProvider] Attempted to find or create a pool for '{}' but the result of the attempt evaluated to None. SqlAlchemyPooledConnectionProvider.UnableToCreateDefaultKey=[SqlAlchemyPooledConnectionProvider] Unable to create a default key for internal connection pools. By default, the user parameter is used, but the given user evaluated to None or the empty string (""). Please ensure you have passed a valid user in the connection properties. diff --git a/aws_advanced_python_wrapper/sqlalchemy_driver_dialect.py b/aws_advanced_python_wrapper/sqlalchemy_driver_dialect.py index 0919196fc..290a8f86d 100644 --- a/aws_advanced_python_wrapper/sqlalchemy_driver_dialect.py +++ b/aws_advanced_python_wrapper/sqlalchemy_driver_dialect.py @@ -67,7 +67,7 @@ def abort_connection(self, conn: Connection): if isinstance(conn, PoolProxiedConnection): conn = conn.driver_connection if conn is None: - return + return None return self._underlying_driver.abort_connection(conn) @@ -122,6 +122,6 @@ def transfer_session_state(self, from_conn: Connection, to_conn: Connection): to_driver_conn = to_conn.driver_connection if from_driver_conn is None or to_driver_conn is None: - return + return None return self._underlying_driver.transfer_session_state(from_driver_conn, to_driver_conn) diff --git a/aws_advanced_python_wrapper/states/session_state.py b/aws_advanced_python_wrapper/states/session_state.py index ad3203c93..0108c8e69 100644 --- a/aws_advanced_python_wrapper/states/session_state.py +++ b/aws_advanced_python_wrapper/states/session_state.py @@ -92,3 +92,6 @@ def copy(self): new_session_state.readonly = self.readonly.copy() return new_session_state + + def __str__(self): + return f"autocommit: {self.auto_commit}, readonly: {self.readonly}" diff --git a/aws_advanced_python_wrapper/utils/pg_exception_handler.py b/aws_advanced_python_wrapper/utils/pg_exception_handler.py index 0e5bae09e..741caac35 100644 --- a/aws_advanced_python_wrapper/utils/pg_exception_handler.py +++ b/aws_advanced_python_wrapper/utils/pg_exception_handler.py @@ -27,9 +27,19 @@ class PgExceptionHandler(ExceptionHandler): _PAM_AUTHENTICATION_FAILED_MSG = "PAM authentication failed" _CONNECTION_FAILED = "connection failed" _CONSUMING_INPUT_FAILED = "consuming input failed" + _CONNECTION_SOCKET_CLOSED = "connection socket closed" - _NETWORK_ERRORS: List[str] - _ACCESS_ERRORS: List[str] + _NETWORK_ERROR_MESSAGES: List[str] = [ + _CONNECTION_FAILED, + _CONSUMING_INPUT_FAILED, + _CONNECTION_SOCKET_CLOSED + ] + _ACCESS_ERROR_MESSAGES: List[str] = [ + _PASSWORD_AUTHENTICATION_FAILED_MSG, + _PAM_AUTHENTICATION_FAILED_MSG + ] + _NETWORK_ERROR_CODES: List[str] + _ACCESS_ERROR_CODES: List[str] def is_network_exception(self, error: Optional[Exception] = None, sql_state: Optional[str] = None) -> bool: if isinstance(error, QueryTimeoutError) or isinstance(error, ConnectionTimeout): @@ -43,7 +53,7 @@ def is_network_exception(self, error: Optional[Exception] = None, sql_state: Opt # getattr may throw an AttributeError if the error does not have a `sqlstate` attribute pass - if sql_state is not None and sql_state in self._NETWORK_ERRORS: + if sql_state is not None and sql_state in self._NETWORK_ERROR_CODES: return True if isinstance(error, OperationalError): @@ -51,7 +61,7 @@ def is_network_exception(self, error: Optional[Exception] = None, sql_state: Opt return False # Check the error message if this is a generic error error_msg: str = error.args[0] - return self._CONNECTION_FAILED in error_msg or self._CONSUMING_INPUT_FAILED in error_msg + return any(msg in error_msg for msg in self._NETWORK_ERROR_MESSAGES) return False @@ -63,7 +73,7 @@ def is_login_exception(self, error: Optional[Exception] = None, sql_state: Optio 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: + if sql_state is not None and sql_state in self._ACCESS_ERROR_CODES: return True if isinstance(error, OperationalError): @@ -72,15 +82,14 @@ def is_login_exception(self, error: Optional[Exception] = None, sql_state: Optio # Check the error message if this is a generic error error_msg: str = error.args[0] - if self._PASSWORD_AUTHENTICATION_FAILED_MSG in error_msg \ - or self._PAM_AUTHENTICATION_FAILED_MSG in error_msg: + if any(msg in error_msg for msg in self._ACCESS_ERROR_MESSAGES): return True return False class SingleAzPgExceptionHandler(PgExceptionHandler): - _NETWORK_ERRORS: List[str] = [ + _NETWORK_ERROR_CODES: List[str] = [ "53", # insufficient resources "57P01", # admin shutdown "57P02", # crash shutdown @@ -92,14 +101,14 @@ class SingleAzPgExceptionHandler(PgExceptionHandler): "XX" # internal error(backend) ] - _ACCESS_ERRORS: List[str] = [ + _ACCESS_ERROR_CODES: List[str] = [ "28000", # PAM authentication errors "28P01" ] class MultiAzPgExceptionHandler(PgExceptionHandler): - _NETWORK_ERRORS: List[str] = [ + _NETWORK_ERROR_CODES: List[str] = [ "28000", # access denied during reboot, this should be considered a temporary failure "53", # insufficient resources "57P01", # admin shutdown @@ -112,4 +121,4 @@ class MultiAzPgExceptionHandler(PgExceptionHandler): "XX" # internal error(backend) ] - _ACCESS_ERRORS: List[str] = ["28P01"] + _ACCESS_ERROR_CODES: List[str] = ["28P01"] diff --git a/aws_advanced_python_wrapper/utils/sliding_expiration_cache.py b/aws_advanced_python_wrapper/utils/sliding_expiration_cache.py index e4bbe8daa..cd3900dc7 100644 --- a/aws_advanced_python_wrapper/utils/sliding_expiration_cache.py +++ b/aws_advanced_python_wrapper/utils/sliding_expiration_cache.py @@ -126,7 +126,6 @@ def _cleanup_thread_internal(self): while True: try: sleep(self._cleanup_interval_ns / 1_000_000_000) - logger.debug("SlidingExpirationCache.CleaningUp") self._cleanup_time_ns.set(perf_counter_ns() + self._cleanup_interval_ns) keys = [key for key, _ in self._cdict.items()] for key in keys: diff --git a/aws_advanced_python_wrapper/utils/telemetry/xray_telemetry.py b/aws_advanced_python_wrapper/utils/telemetry/xray_telemetry.py index ebb81bbf8..b3a7bf530 100644 --- a/aws_advanced_python_wrapper/utils/telemetry/xray_telemetry.py +++ b/aws_advanced_python_wrapper/utils/telemetry/xray_telemetry.py @@ -78,7 +78,7 @@ def post_copy(context: XRayTelemetryContext, trace_level: TelemetryTraceLevel): return if trace_level in [TelemetryTraceLevel.FORCE_TOP_LEVEL, TelemetryTraceLevel.TOP_LEVEL]: - with ThreadPoolExecutor() as executor: + with ThreadPoolExecutor(thread_name_prefix=context.get_name()) as executor: future = executor.submit(_clone_and_close_context, context, trace_level) future.result() else: diff --git a/aws_advanced_python_wrapper/writer_failover_handler.py b/aws_advanced_python_wrapper/writer_failover_handler.py index 25b701944..7888df493 100644 --- a/aws_advanced_python_wrapper/writer_failover_handler.py +++ b/aws_advanced_python_wrapper/writer_failover_handler.py @@ -120,9 +120,9 @@ def get_result_from_future(self, current_topology: Tuple[HostInfo, ...]) -> Writ executor = ThreadPoolExecutor(thread_name_prefix="WriterFailoverHandlerExecutor") try: + futures = [executor.submit(self.reconnect_to_writer, writer_host), + executor.submit(self.wait_for_new_writer, current_topology, writer_host)] try: - futures = [executor.submit(self.reconnect_to_writer, writer_host), - executor.submit(self.wait_for_new_writer, current_topology, writer_host)] for future in as_completed(futures, timeout=self._max_failover_timeout_sec): result = future.result() if result.is_connected: @@ -132,9 +132,10 @@ def get_result_from_future(self, current_topology: Tuple[HostInfo, ...]) -> Writ return result except TimeoutError: self._timeout_event.set() - finally: - self._timeout_event.set() + for future in futures: + future.cancel() finally: + self._timeout_event.set() executor.shutdown(wait=False) return WriterFailoverHandlerImpl.failed_writer_failover_result diff --git a/tests/integration/container/conftest.py b/tests/integration/container/conftest.py index c438561fa..2e23eeba3 100644 --- a/tests/integration/container/conftest.py +++ b/tests/integration/container/conftest.py @@ -28,6 +28,8 @@ DriverDialectManager from aws_advanced_python_wrapper.exception_handling import ExceptionManager from aws_advanced_python_wrapper.host_list_provider import RdsHostListProvider +from aws_advanced_python_wrapper.host_monitoring_plugin import \ + MonitoringThreadContainer from aws_advanced_python_wrapper.plugin_service import PluginServiceImpl from aws_advanced_python_wrapper.utils.log import Logger from aws_advanced_python_wrapper.utils.rdsutils import RdsUtils @@ -67,6 +69,8 @@ def pytest_runtest_setup(item): else: TestEnvironment.get_current().set_current_driver(None) + logger.info("Starting test preparation for: " + test_name) + segment: Optional[Segment] = None if TestEnvironmentFeatures.TELEMETRY_TRACES_ENABLED in TestEnvironment.get_current().get_features(): segment = xray_recorder.begin_segment("test: setup") @@ -92,7 +96,7 @@ def pytest_runtest_setup(item): # Wait up to 5min instances: List[str] = list() start_time = timeit.default_timer() - while (len(instances) != request.get_num_of_instances() + while (len(instances) < request.get_num_of_instances() or len(instances) == 0 or not rds_utility.is_db_instance_writer(instances[0])) and ( timeit.default_timer() - start_time) < 300: # 5 min @@ -135,6 +139,7 @@ def pytest_runtest_setup(item): DatabaseDialectManager._known_endpoint_dialects.clear() CustomEndpointPlugin._monitors.clear() CustomEndpointMonitor._custom_endpoint_info_cache.clear() + MonitoringThreadContainer.clean_up() ConnectionProviderManager.reset_provider() DatabaseDialectManager.reset_custom_dialect() diff --git a/tests/integration/container/test_aurora_failover.py b/tests/integration/container/test_aurora_failover.py index db8dfbe9e..fa6bb2e0c 100644 --- a/tests/integration/container/test_aurora_failover.py +++ b/tests/integration/container/test_aurora_failover.py @@ -39,6 +39,8 @@ from .utils.test_environment import TestEnvironment from .utils.test_environment_features import TestEnvironmentFeatures +logger = Logger(__name__) + @enable_on_num_instances(min_instances=2) @enable_on_deployments([DatabaseEngineDeployment.AURORA, DatabaseEngineDeployment.RDS_MULTI_AZ_CLUSTER]) @@ -49,6 +51,12 @@ class TestAuroraFailover: IDLE_CONNECTIONS_NUM: int = 5 logger = Logger(__name__) + @pytest.fixture(autouse=True) + def setup_method(self, request): + self.logger.info(f"Starting test: {request.node.name}") + yield + self.logger.info(f"Ending test: {request.node.name}") + @pytest.fixture(scope='class') def aurora_utility(self): region: str = TestEnvironment.get_current().get_info().get_region() @@ -56,7 +64,15 @@ def aurora_utility(self): @pytest.fixture(scope='class') def props(self): - p: Properties = Properties({"plugins": "failover", "connect_timeout": 60, "topology_refresh_ms": 10, "autocommit": True}) + p: Properties = Properties({ + "plugins": "failover", + "socket_timeout": 10, + "connect_timeout": 10, + "monitoring-connect_timeout": 5, + "monitoring-socket_timeout": 5, + "topology_refresh_ms": 10, + "autocommit": True + }) features = TestEnvironment.get_current().get_features() if TestEnvironmentFeatures.TELEMETRY_TRACES_ENABLED in features \ @@ -116,7 +132,6 @@ def test_fail_from_writer_to_new_writer_fail_on_connection_bound_object_invocati assert aurora_utility.is_db_instance_writer(current_connection_id) is True assert current_connection_id != initial_writer_id - @pytest.mark.parametrize("plugins", ["failover,host_monitoring", "failover,host_monitoring_v2"]) @enable_on_features([TestEnvironmentFeatures.NETWORK_OUTAGES_ENABLED, TestEnvironmentFeatures.ABORT_CONNECTION_SUPPORTED]) def test_fail_from_reader_to_writer( @@ -125,13 +140,12 @@ def test_fail_from_reader_to_writer( test_driver: TestDriver, conn_utils, proxied_props, - aurora_utility, - plugins): + aurora_utility): target_driver_connect = DriverHelper.get_connect_func(test_driver) reader: TestInstanceInfo = test_environment.get_proxy_instances()[1] writer_id: str = test_environment.get_proxy_writer().get_instance_id() - proxied_props["plugins"] = plugins + proxied_props["plugins"] = "failover,host_monitoring" with AwsWrapperConnection.connect( target_driver_connect, **conn_utils.get_proxy_connect_params(reader.get_host()), @@ -323,28 +337,3 @@ def test_writer_failover_in_idle_connections( # Ensure that all idle connections are closed. for idle_connection in idle_connections: assert idle_connection.is_closed is True - - @enable_on_features([TestEnvironmentFeatures.NETWORK_OUTAGES_ENABLED]) - def test_failover__socket_timeout( - self, - test_driver: TestDriver, - test_environment: TestEnvironment, - proxied_props, - conn_utils, - aurora_utility): - target_driver_connect = DriverHelper.get_connect_func(test_driver) - reader: TestInstanceInfo = test_environment.get_proxy_instances()[1] - writer_id: str = test_environment.get_proxy_writer().get_instance_id() - - WrapperProperties.PLUGINS.set(proxied_props, "failover") - WrapperProperties.SOCKET_TIMEOUT_SEC.set(proxied_props, 3) - with AwsWrapperConnection.connect( - target_driver_connect, - **conn_utils.get_proxy_connect_params(reader.get_host()), - **proxied_props) as aws_conn: - ProxyHelper.disable_connectivity(reader.get_instance_id()) - aurora_utility.assert_first_query_throws(aws_conn, FailoverSuccessError) - - current_connection_id = aurora_utility.query_instance_id(aws_conn) - assert writer_id == current_connection_id - assert aurora_utility.is_db_instance_writer(current_connection_id) is True diff --git a/tests/integration/container/test_basic_connectivity.py b/tests/integration/container/test_basic_connectivity.py index aa70e1080..e0c5504fc 100644 --- a/tests/integration/container/test_basic_connectivity.py +++ b/tests/integration/container/test_basic_connectivity.py @@ -126,16 +126,19 @@ def test_proxied_wrapper_connection_failed( # That is expected exception. Test pass. assert True - @pytest.mark.parametrize("plugins", ["host_monitoring", "host_monitoring_v2"]) @enable_on_num_instances(min_instances=2) @enable_on_deployments([DatabaseEngineDeployment.AURORA, DatabaseEngineDeployment.RDS_MULTI_AZ_CLUSTER]) @enable_on_features([TestEnvironmentFeatures.ABORT_CONNECTION_SUPPORTED]) def test_wrapper_connection_reader_cluster_with_efm_enabled(self, test_driver: TestDriver, conn_utils, plugins): + props: Properties = Properties({ + WrapperProperties.PLUGINS.name: "host_monitoring", + "socket_timeout": 5, + "connect_timeout": 5, + "monitoring-connect_timeout": 3, + "monitoring-socket_timeout": 3, + "autocommit": True}) target_driver_connect = DriverHelper.get_connect_func(test_driver) - conn = AwsWrapperConnection.connect( - target_driver_connect, - **conn_utils.get_connect_params(conn_utils.reader_cluster_host), - plugins=plugins, connect_timeout=10) + conn = AwsWrapperConnection.connect(target_driver_connect, **conn_utils.get_connect_params(conn_utils.reader_cluster_host), **props) cursor = conn.cursor() cursor.execute("SELECT 1") result = cursor.fetchone() diff --git a/tests/integration/container/test_host_monitoring_v2.py b/tests/integration/container/test_host_monitoring_v2.py index 6b797c692..ee9ff4b68 100644 --- a/tests/integration/container/test_host_monitoring_v2.py +++ b/tests/integration/container/test_host_monitoring_v2.py @@ -22,7 +22,8 @@ from aws_advanced_python_wrapper.utils.properties import (Properties, WrapperProperties) from tests.integration.container.utils.conditions import ( - disable_on_features, enable_on_deployments) + disable_on_engines, disable_on_features, enable_on_deployments) +from tests.integration.container.utils.database_engine import DatabaseEngine from tests.integration.container.utils.database_engine_deployment import \ DatabaseEngineDeployment from tests.integration.container.utils.driver_helper import DriverHelper @@ -44,6 +45,7 @@ @disable_on_features([TestEnvironmentFeatures.PERFORMANCE, TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY, TestEnvironmentFeatures.BLUE_GREEN_DEPLOYMENT]) +@disable_on_engines([DatabaseEngine.MYSQL]) class TestHostMonitoringV2: @pytest.fixture(scope='class') def rds_utils(self): @@ -55,6 +57,8 @@ def props(self): p: Properties = Properties({"plugins": "host_monitoring_v2", "socket_timeout": 30, "connect_timeout": 10, + "monitoring-connect_timeout": 5, + "monitoring-socket_timeout": 5, "failure_detection_time_ms": 5_000, "failure_detection_interval_ms": 5_000, "failure_detection_count": 1, diff --git a/tests/integration/container/test_read_write_splitting.py b/tests/integration/container/test_read_write_splitting.py index c17664185..d73dabb67 100644 --- a/tests/integration/container/test_read_write_splitting.py +++ b/tests/integration/container/test_read_write_splitting.py @@ -26,6 +26,7 @@ from aws_advanced_python_wrapper.host_list_provider import RdsHostListProvider from aws_advanced_python_wrapper.sql_alchemy_connection_provider import \ SqlAlchemyPooledConnectionProvider +from aws_advanced_python_wrapper.utils.log import Logger from aws_advanced_python_wrapper.utils.properties import (Properties, WrapperProperties) from tests.integration.container.utils.conditions import ( @@ -51,6 +52,15 @@ TestEnvironmentFeatures.BLUE_GREEN_DEPLOYMENT, TestEnvironmentFeatures.PERFORMANCE]) class TestReadWriteSplitting: + + logger = Logger(__name__) + + @pytest.fixture(autouse=True) + def setup_method(self, request): + self.logger.info(f"Starting test: {request.node.name}") + yield + self.logger.info(f"Ending test: {request.node.name}") + @pytest.fixture(scope='class') def rds_utils(self): region: str = TestEnvironment.get_current().get_info().get_region() @@ -64,7 +74,7 @@ def clear_caches(self): @pytest.fixture(scope='class') def props(self): - p: Properties = Properties({"plugins": "read_write_splitting", "connect_timeout": 30, "autocommit": True}) + p: Properties = Properties({"plugins": "read_write_splitting", "socket_timeout": 10, "connect_timeout": 10, "autocommit": True}) if TestEnvironmentFeatures.TELEMETRY_TRACES_ENABLED in TestEnvironment.get_current().get_features() \ or TestEnvironmentFeatures.TELEMETRY_METRICS_ENABLED in TestEnvironment.get_current().get_features(): @@ -82,7 +92,11 @@ def props(self): @pytest.fixture(scope='class') def failover_props(self): return { - "plugins": "read_write_splitting,failover", "connect_timeout": 10, "autocommit": True} + "plugins": "read_write_splitting,failover", + "socket_timeout": 10, + "connect_timeout": 10, + "autocommit": True + } @pytest.fixture(scope='class') def proxied_props(self, props, conn_utils): @@ -243,9 +257,6 @@ def test_set_read_only_true__all_readers_down( target_driver_connect = DriverHelper.get_connect_func(test_driver) connect_params = conn_utils.get_proxy_connect_params() - # To prevent endless waiting while executing SQL queries - WrapperProperties.SOCKET_TIMEOUT_SEC.set(connect_params, 10) - with AwsWrapperConnection.connect(target_driver_connect, **connect_params, **proxied_props) as conn: writer_id = rds_utils.query_instance_id(conn) @@ -317,9 +328,6 @@ def test_failover_to_new_writer__switch_read_only( target_driver_connect = DriverHelper.get_connect_func(test_driver) connect_params = conn_utils.get_proxy_connect_params() - # To prevent endless waiting while executing SQL queries - WrapperProperties.SOCKET_TIMEOUT_SEC.set(connect_params, 10) - with AwsWrapperConnection.connect(target_driver_connect, **connect_params, **proxied_failover_props) as conn: original_writer_id = rds_utils.query_instance_id(conn) @@ -349,15 +357,13 @@ def test_failover_to_new_writer__switch_read_only( current_id = rds_utils.query_instance_id(conn) assert new_writer_id == current_id - @pytest.mark.parametrize("plugins", ["read_write_splitting,failover,host_monitoring_v2"]) @enable_on_features([TestEnvironmentFeatures.NETWORK_OUTAGES_ENABLED, TestEnvironmentFeatures.ABORT_CONNECTION_SUPPORTED]) @enable_on_num_instances(min_instances=3) @disable_on_engines([DatabaseEngine.MYSQL]) def test_failover_to_new_reader__switch_read_only( self, test_environment: TestEnvironment, test_driver: TestDriver, - proxied_failover_props, conn_utils, rds_utils, plugins): - WrapperProperties.PLUGINS.set(proxied_failover_props, plugins) + proxied_failover_props, conn_utils, rds_utils): WrapperProperties.FAILOVER_MODE.set(proxied_failover_props, "reader-or-writer") target_driver_connect = DriverHelper.get_connect_func(test_driver) @@ -398,16 +404,13 @@ def test_failover_to_new_reader__switch_read_only( current_id = rds_utils.query_instance_id(conn) assert other_reader_id == current_id - @pytest.mark.parametrize("plugins", ["read_write_splitting,failover,host_monitoring", - "read_write_splitting,failover,host_monitoring_v2"]) @enable_on_features([TestEnvironmentFeatures.NETWORK_OUTAGES_ENABLED, TestEnvironmentFeatures.ABORT_CONNECTION_SUPPORTED]) @enable_on_num_instances(min_instances=3) @disable_on_engines([DatabaseEngine.MYSQL]) def test_failover_reader_to_writer__switch_read_only( self, test_environment: TestEnvironment, test_driver: TestDriver, - proxied_failover_props, conn_utils, rds_utils, plugins): - WrapperProperties.PLUGINS.set(proxied_failover_props, plugins) + proxied_failover_props, conn_utils, rds_utils): target_driver_connect = DriverHelper.get_connect_func(test_driver) with AwsWrapperConnection.connect( target_driver_connect, **conn_utils.get_proxy_connect_params(), **proxied_failover_props) as conn: @@ -519,19 +522,16 @@ def test_pooled_connection__cluster_url_failover( new_driver_conn = conn.target_connection assert initial_driver_conn is not new_driver_conn - @pytest.mark.parametrize("plugins", ["read_write_splitting,failover,host_monitoring", - "read_write_splitting,failover,host_monitoring_v2"]) @enable_on_features([TestEnvironmentFeatures.FAILOVER_SUPPORTED, TestEnvironmentFeatures.NETWORK_OUTAGES_ENABLED, TestEnvironmentFeatures.ABORT_CONNECTION_SUPPORTED]) @disable_on_engines([DatabaseEngine.MYSQL]) def test_pooled_connection__failover_failed( self, test_environment: TestEnvironment, test_driver: TestDriver, - rds_utils, conn_utils, proxied_failover_props, plugins): + rds_utils, conn_utils, proxied_failover_props): writer_host = test_environment.get_writer().get_host() provider = SqlAlchemyPooledConnectionProvider(lambda _, __: {"pool_size": 1}, None, lambda host_info, props: writer_host in host_info.host) ConnectionProviderManager.set_connection_provider(provider) - WrapperProperties.PLUGINS.set(proxied_failover_props, plugins) WrapperProperties.FAILOVER_TIMEOUT_SEC.set(proxied_failover_props, "1") WrapperProperties.FAILURE_DETECTION_TIME_MS.set(proxied_failover_props, "1000") WrapperProperties.FAILURE_DETECTION_COUNT.set(proxied_failover_props, "1") diff --git a/tests/integration/container/utils/rds_test_utility.py b/tests/integration/container/utils/rds_test_utility.py index 7b44bca7f..9fc069bbd 100644 --- a/tests/integration/container/utils/rds_test_utility.py +++ b/tests/integration/container/utils/rds_test_utility.py @@ -132,10 +132,13 @@ def wait_until_instance_has_desired_status( "RdsTestUtility.InstanceDescriptionTimeout", instance_id, desired_status, wait_time_mins)) def wait_until_cluster_has_desired_status(self, cluster_id: str, desired_status: str) -> None: + stop_time = datetime.now() + timedelta(minutes=10) cluster_info = self.get_db_cluster(cluster_id) status = cluster_info.get("Status") while status != desired_status: - sleep(1) + if datetime.now() > stop_time: + raise TimeoutError(f"Cluster {cluster_id} did not reach status '{desired_status}' within 10 minutes.") + sleep(10) cluster_info = self.get_db_cluster(cluster_id) status = cluster_info.get("Status") diff --git a/tests/integration/host/build.gradle.kts b/tests/integration/host/build.gradle.kts index b746eb642..6ede2d6ce 100644 --- a/tests/integration/host/build.gradle.kts +++ b/tests/integration/host/build.gradle.kts @@ -72,6 +72,8 @@ tasks.register("test-python-3.11-mysql") { systemProperty("exclude-multi-az-cluster", "true") systemProperty("exclude-multi-az-instance", "true") systemProperty("exclude-bg", "true") + systemProperty("exclude-traces-telemetry", "true") + systemProperty("exclude-metrics-telemetry", "true") systemProperty("exclude-pg-driver", "true") systemProperty("exclude-pg-engine", "true") } @@ -86,6 +88,8 @@ tasks.register("test-python-3.8-mysql") { systemProperty("exclude-multi-az-cluster", "true") systemProperty("exclude-multi-az-instance", "true") systemProperty("exclude-bg", "true") + systemProperty("exclude-traces-telemetry", "true") + systemProperty("exclude-metrics-telemetry", "true") systemProperty("exclude-pg-driver", "true") systemProperty("exclude-pg-engine", "true") } diff --git a/tests/integration/host/src/test/java/integration/host/TestEnvironmentConfiguration.java b/tests/integration/host/src/test/java/integration/host/TestEnvironmentConfiguration.java index f36f8af8c..3e7e56773 100644 --- a/tests/integration/host/src/test/java/integration/host/TestEnvironmentConfiguration.java +++ b/tests/integration/host/src/test/java/integration/host/TestEnvironmentConfiguration.java @@ -18,6 +18,10 @@ import integration.DebugEnv; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; + public class TestEnvironmentConfiguration { public boolean excludeDocker = @@ -99,5 +103,8 @@ public class TestEnvironmentConfiguration { public String iamUser = System.getenv("IAM_USER"); + final String numInstancesVar = System.getenv("NUM_INSTANCES"); + final Integer numInstances = numInstancesVar == null ? null : Integer.parseInt(numInstancesVar); + public DebugEnv debugEnv = DebugEnv.fromEnv(); } diff --git a/tests/integration/host/src/test/java/integration/host/TestEnvironmentProvider.java b/tests/integration/host/src/test/java/integration/host/TestEnvironmentProvider.java index 15011003b..52d126faa 100644 --- a/tests/integration/host/src/test/java/integration/host/TestEnvironmentProvider.java +++ b/tests/integration/host/src/test/java/integration/host/TestEnvironmentProvider.java @@ -26,7 +26,9 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.concurrent.Future; import java.util.logging.Logger; import java.util.stream.Stream; @@ -54,6 +56,23 @@ public Stream provideTestTemplateInvocationContex TestEnvironmentConfiguration config = new TestEnvironmentConfiguration(); + final List validNumInstances = Arrays.asList(1, 2, 3, 5); + final List instancesToTest; + if (config.numInstances != null) { + System.out.printf( + "The NUM_INSTANCES environment variable was set to %d. All test configurations for different cluster sizes will be skipped.%n", + config.numInstances); + if (!validNumInstances.contains(config.numInstances)) { + throw new RuntimeException( + String.format( + "The NUM_INSTANCES environment variable was set to an invalid value: %d. Valid values are: %s.", + config.numInstances, validNumInstances)); + } + instancesToTest = Arrays.asList(config.numInstances); + } else { + instancesToTest = validNumInstances; + } + for (DatabaseEngineDeployment deployment : DatabaseEngineDeployment.values()) { if (deployment == DatabaseEngineDeployment.DOCKER && config.excludeDocker) { continue; @@ -86,7 +105,7 @@ public Stream provideTestTemplateInvocationContex continue; } - for (int numOfInstances : Arrays.asList(1, 2, 3, 5)) { + for (int numOfInstances : instancesToTest) { if (instances == DatabaseInstances.SINGLE_INSTANCE && numOfInstances > 1) { continue; } diff --git a/tests/unit/test_failover_plugin.py b/tests/unit/test_failover_plugin.py index a9e0c856c..3c6f4f0ec 100644 --- a/tests/unit/test_failover_plugin.py +++ b/tests/unit/test_failover_plugin.py @@ -410,7 +410,7 @@ def test_invalidate_current_connection_with_open_connection(plugin_service_mock, plugin = FailoverPlugin(plugin_service_mock, Properties()) - with mock.patch.object(conn_mock, "close") as close_mock: + with mock.patch.object(driver_dialect_mock, "execute") as close_mock: with mock.patch.object(driver_dialect_mock, "is_closed") as is_closed_mock: driver_dialect_mock.is_closed.return_value = False