diff --git a/tests/integration/container/test_blue_green_deployment.py b/tests/integration/container/test_blue_green_deployment.py index aab38443..41893ba6 100644 --- a/tests/integration/container/test_blue_green_deployment.py +++ b/tests/integration/container/test_blue_green_deployment.py @@ -29,8 +29,10 @@ import math import socket +import sys from collections import deque from dataclasses import dataclass, field +from logging import INFO, getLogger from threading import Event, Thread from time import perf_counter_ns, sleep @@ -39,9 +41,11 @@ from aws_advanced_python_wrapper import AwsWrapperConnection from aws_advanced_python_wrapper.blue_green_plugin import (BlueGreenPlugin, - BlueGreenRole) + BlueGreenRole, + BlueGreenStatus) from aws_advanced_python_wrapper.database_dialect import DialectCode from aws_advanced_python_wrapper.driver_info import DriverInfo +from aws_advanced_python_wrapper.utils import services_container from aws_advanced_python_wrapper.utils.atomic import AtomicInt from aws_advanced_python_wrapper.utils.concurrent import (ConcurrentDict, CountDownLatch) @@ -49,6 +53,20 @@ from aws_advanced_python_wrapper.utils.properties import (Properties, WrapperProperties) from aws_advanced_python_wrapper.utils.rds_utils import RdsUtils +from .utils.blue_green_checks import (BG_CONNECT_TIMEOUT_MS, BG_ID, + PLUGIN_STATUS_POLL_SEC, + PROBE_CONNECT_TIMEOUT_SEC, + ROUTING_PROBE_ATTEMPTS, + SWITCHOVER_TEST_TIMEOUT_SEC, + THREAD_JOIN_SEC, PluginStatusObservation, + ProbeResult, RoutingAttempt, + connect_suspend_listed, describe_error, + evaluate_routing, + evaluate_suspended_call, get_probe_host, + join_all, listed_substitute, + record_plugin_phase, settle_switchover, + summary_lines, wait_for_phase, + wait_for_switchover) from .utils.conditions import enable_on_deployments, enable_on_features from .utils.database_engine import DatabaseEngine from .utils.database_engine_deployment import DatabaseEngineDeployment @@ -88,7 +106,11 @@ def test_utility(self): def rds_utils(self): return RdsUtils() + # Worst case on any tree: setup + 150 s (latest request) + 300 s (API wait from the request) + 5 s (trigger join) + # + 2 s (CPU sample) + 20 s (joins), so setup + 477 s. The harness's --timeout=600 stays for every other test. + @pytest.mark.timeout(SWITCHOVER_TEST_TIMEOUT_SEC) def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: TestEnvironment, test_driver): + test_start_ns = perf_counter_ns() self.results.clear() self.unhandled_exceptions.clear() @@ -119,6 +141,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: if rds_utils.is_not_green_or_old_instance(host): threads.append(Thread( + daemon=True, target=self.direct_topology_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -126,6 +149,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.direct_blue_connectivity_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -133,6 +157,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.direct_blue_idle_connectivity_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -140,6 +165,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_blue_idle_connectivity_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -147,6 +173,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_blue_executing_connectivity_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -154,6 +181,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_blue_new_connection_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -161,6 +189,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.blue_dns_monitor, args=(host_id, host, start_latch, stop, finish_latch, bg_results))) thread_count += 1 @@ -168,6 +197,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: if rds_utils.is_green_instance(host): threads.append(Thread( + daemon=True, target=self.direct_topology_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -175,6 +205,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_green_connectivity_monitor, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -182,6 +213,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.green_dns_monitor, args=(host_id, host, start_latch, stop, finish_latch, bg_results))) thread_count += 1 @@ -191,6 +223,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: rds_client = test_utility.get_rds_client() threads.append(Thread( + daemon=True, target=self.green_iam_connectivity_monitor, args=(test_driver, conn_utils, rds_client, host_id, "BlueHostToken", rds_utils.remove_green_instance_prefix(host), host, test_instance.get_port(), @@ -200,6 +233,7 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.green_iam_connectivity_monitor, args=(test_driver, conn_utils, rds_client, host_id, "GreenHostToken", host, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, @@ -208,52 +242,150 @@ def test_switchover(self, conn_utils, test_utility, rds_utils, test_environment: thread_count += 1 thread_finish_count += 1 - threads.append(Thread( + switchover_requested = Event() + switchover_requested_ns = AtomicInt(0) + trigger_thread = Thread( + daemon=True, target=self.bg_switchover_trigger, - args=(test_utility, info.get_bg_deployment_id(), start_latch, finish_latch, self.results))) + args=(test_utility, info.get_bg_deployment_id(), start_latch, finish_latch, self.results, stop, + switchover_requested, switchover_requested_ns)) + threads.append(trigger_thread) thread_count += 1 thread_finish_count += 1 + # #1352 probes. Not counted in the latches: each one starts on the plugin phase it needs. + port = test_instance.get_port() + blue_endpoint = test_environment.get_database_info().get_cluster_endpoint() \ + if test_environment.get_deployment() == DatabaseEngineDeployment.AURORA else test_instance.get_host() + green_endpoint = self.get_green_endpoint(test_environment, test_utility, info.get_bg_deployment_id()) + writer_host = test_instance.get_host() + blue_writer_ip = self.resolve_ip(blue_endpoint) + green_writer_ip = "unresolved" if green_endpoint is None else self.resolve_ip(green_endpoint) + green_host = next((host for host in topology_instances if rds_utils.is_green_instance(host)), None) + observation = PluginStatusObservation() + execute_probe = ProbeResult("WrapperBlueSuspendedExecute") + connect_probe = ProbeResult("WrapperGreenSuspendedConnect", probe_host=green_host or "") + post_probe = ProbeResult("WrapperBlueRouting POST") + preparation_probe = ProbeResult("WrapperBlueRouting PREPARATION") + probes = [execute_probe, connect_probe, post_probe, preparation_probe] + threads.append(Thread( + daemon=True, + target=self.plugin_status_monitor, + args=(observation, switchover_requested, stop))) + threads.append(Thread( + daemon=True, + target=self.wrapper_blue_suspended_execute_probe, + args=(test_driver, conn_utils, blue_endpoint, writer_host, port, db_name, observation, switchover_requested, + stop, execute_probe))) + threads.append(Thread( + daemon=True, + target=self.wrapper_green_suspended_connect_probe, + args=(test_driver, conn_utils, green_host, port, db_name, observation, stop, connect_probe))) + threads.append(Thread( + daemon=True, + target=self.wrapper_blue_routing_probe, + args=(test_driver, conn_utils, blue_endpoint, writer_host, port, db_name, observation, switchover_requested, + stop, preparation_probe, post_probe))) + start_latch.set_count(thread_count) finish_latch.set_count(thread_finish_count) for result in self.results.values(): result.start_time_ns.set(start_time_ns) - for thread in threads: - thread.start() - - self.logger.debug("All threads started.") - - finish_latch.wait_sec(6 * 60) - self.logger.debug("All threads completed.") - - sleep(12 * 60) - - self.logger.debug("Stopping all threads...") - stop.set() - - for thread in threads: - thread.join(timeout=30) - if thread.is_alive(): - self.logger.debug("Timed out waiting for a thread to stop running...") + # An unfixed wrapper's suspended calls log two debug records per pass of their loop; keep them out of the report. + plugin_logger = getLogger(BlueGreenPlugin.__module__) + plugin_log_level = plugin_logger.level + plugin_logger.setLevel(INFO) + switch_interval = sys.getswitchinterval() + threads_start_ns = perf_counter_ns() + wait_result = "not run" + try: + try: + # conftest released the services container and its lazy re-initialisation isn't locked: build it here, + # before the monitors' first connects race on it. + services_container.get_monitor_service() + for thread in threads: + thread.start() + + self.logger.debug("All threads started.") + wait_result = wait_for_switchover( + probes, observation, switchover_requested, switchover_requested_ns, threads_start_ns, + self.is_switchover_completed_seen) + self.logger.debug(f"Done waiting for the switchover: {wait_result}.") + finally: + settle_switchover( + test_utility, info.get_bg_deployment_id(), trigger_thread, stop, switchover_requested, + switchover_requested_ns) + self.logger.debug("Stop set for all threads.") + + alive_count = join_all(threads, THREAD_JOIN_SEC) + self.logger.debug(f"Done waiting for threads to stop, {alive_count} still running.") + + writer_ips = f"green writer IP {green_writer_ip}, blue writer IP {blue_writer_ip}" + not_sampled = [] if preparation_probe.phase_at_start == "PREPARATION" else [f"{preparation_probe.name}: not sampled"] + for line in summary_lines( + test_start_ns, threads_start_ns, switchover_requested_ns.get(), wait_result, observation, probes, + [writer_ips] + not_sampled): + print(line) + + for host_id, result in self.results.items(): + assert result.bg_trigger_time_ns.get() > 0, \ + (f"bg_trigger_time for {host_id} was {result.bg_trigger_time_ns.get()}, wait result: {wait_result}, " + f"unhandled exceptions: {', '.join(describe_error(e) for e in self.unhandled_exceptions) or 'none'}") + + self.logger.debug("Test is over.") + now_ns = perf_counter_ns() + post_first_seen_ns = observation.plugin_phase_first_seen_ns.get("POST") + failures = ( + evaluate_suspended_call("execute", execute_probe, now_ns, post_first_seen_ns) + + evaluate_suspended_call("connect", connect_probe, now_ns, post_first_seen_ns) + + evaluate_routing(post_probe, "POST", required=True, writer_ips=writer_ips) + + evaluate_routing(preparation_probe, "PREPARATION", required=False, writer_ips=writer_ips)) + # A monitor still appending to its results can make print_metrics raise; it must not hide the verdict. + try: + self.print_metrics(rds_utils) + except Exception as e: + print(f"print_metrics failed: {describe_error(e)}") + if failures: + pytest.fail("\n".join(failures)) - self.logger.debug("Done waiting for threads to stop.") + if len(self.unhandled_exceptions) > 0: + self.log_unhandled_exceptions() + pytest.fail("There were unhandled exceptions.") - for host_id, result in self.results.items(): - assert result.bg_trigger_time_ns.get() > 0, \ - f"bg_trigger_time for {host_id} was {result.bg_trigger_time_ns.get()}" + self.assert_test() - self.logger.debug("Test is over.") - self.print_metrics(rds_utils) + self.logger.debug("Completed") + finally: + # Any thread still alive may be an unfixed call spinning (the inherited monitors as well as the probes). + # Restoring the interval or the DEBUG level under it floods the output until the harness can't keep up. + # A lane runs one test. + if not any(probe.is_in_flight() for probe in probes) and not any(thread.is_alive() for thread in threads): + sys.setswitchinterval(switch_interval) + plugin_logger.setLevel(plugin_log_level) - if len(self.unhandled_exceptions) > 0: - self.log_unhandled_exceptions() - pytest.fail("There were unhandled exceptions.") + def is_switchover_completed_seen(self) -> bool: + return any(result.green_status_time.get("SWITCHOVER_COMPLETED", 0) for result in self.results.values()) - self.assert_test() + def resolve_ip(self, host: str) -> str: + try: + return socket.gethostbyname(host) + except Exception: + return "unresolved" - self.logger.debug("Completed") + def get_green_endpoint(self, test_env: TestEnvironment, test_utility: RdsTestUtility, bg_id: str) -> Optional[str]: + try: + bg_deployment = test_utility.get_blue_green_deployment(bg_id) + if bg_deployment is None: + return None + if test_env.get_deployment() == DatabaseEngineDeployment.RDS_MULTI_AZ_INSTANCE: + green_instance = test_utility.get_rds_instance_info_by_arn(bg_deployment["Target"]) + return None if green_instance is None else green_instance["Endpoint"]["Address"] + green_cluster = test_utility.get_cluster_by_arn(bg_deployment["Target"]) + return None if green_cluster is None else green_cluster["Endpoint"] + except Exception: + return None def get_bg_endpoints( self, @@ -348,10 +480,12 @@ def direct_topology_monitor( else: pytest.fail(f"Unsupported database engine: {engine}") + # Each poll must read the status table as it is now: without autocommit, MySQL's REPEATABLE READ + # repeats the first read's snapshot (the plugin sets autocommit on its status connections for this reason). + connect_params = {**conn_utils.get_connect_params(host=host, port=port, dbname=db), "autocommit": True} + try: - conn = self.get_direct_connection_with_retry( - test_driver, - **conn_utils.get_connect_params(host=host, port=port, dbname=db)) + conn = self.get_direct_connection_with_retry(test_driver, **connect_params) self.logger.debug(f"[DirectTopology] @ {host_id}] Connection opened.") sleep(1) @@ -366,8 +500,7 @@ def direct_topology_monitor( end_time_ns = perf_counter_ns() + 15 * 60 * 1_000_000_000 # 15 minutes while not stop.is_set() and perf_counter_ns() < end_time_ns: if conn is None: - conn = self.get_direct_connection_with_retry( - test_driver, **conn_utils.get_connect_params(host=host, port=port, dbname=db)) + conn = self.get_direct_connection_with_retry(test_driver, **connect_params) self.logger.debug(f"[DirectTopology] @ {host_id}] Connection re-opened.") try: @@ -605,6 +738,7 @@ def get_wrapper_connect_params(self, conn_utils: ConnectionUtils, host: str, por params = conn_utils.get_connect_params(host=host, port=port, dbname=db) params = {**params, **self.get_telemetry_params()} params[WrapperProperties.CLUSTER_ID.name] = self.TEST_CLUSTER_ID + params[WrapperProperties.BG_CONNECT_TIMEOUT_MS.name] = BG_CONNECT_TIMEOUT_MS test_env = TestEnvironment.get_current() engine = test_env.get_engine() db_deployment = test_env.get_deployment() @@ -787,6 +921,198 @@ def wrapper_blue_new_connection_monitor( finish_latch.count_down() self.logger.debug(f"[WrapperBlueNewConnection @ {host_id}] Thread is completed.") + # #1352 observer: the first time each plugin phase is published. Sync status storage is process-wide. + def plugin_status_monitor(self, observation: PluginStatusObservation, switchover_requested: Event, stop: Event) -> None: + while not stop.is_set(): + try: + status = self.read_plugin_status() + if status is not None: + record_plugin_phase(observation.plugin_phase_first_seen_ns, observation.phase_events, status) + if not switchover_requested.is_set(): + observation.pre_trigger_status = status + observation.status_seen.set() + except Exception as e: + observation.error = describe_error(e) + sleep(PLUGIN_STATUS_POLL_SEC) + + def read_plugin_status(self) -> Optional[BlueGreenStatus]: + return services_container.get_storage_service().get(BlueGreenStatus, BG_ID, register_data_access=False) + + def read_plugin_phase(self) -> Optional[str]: + try: + status = self.read_plugin_status() + except Exception: + return None + return None if status is None else status.phase.name + + # #1352: an execute that starts in IN_PROGRESS must return once the phase moves on, without a busy loop. + def wrapper_blue_suspended_execute_probe( + self, + test_driver: TestDriver, + conn_utils: ConnectionUtils, + blue_endpoint: str, + writer_host: str, + port: int, + db: str, + observation: PluginStatusObservation, + switchover_requested: Event, + stop: Event, + result: ProbeResult) -> None: + conn = None + try: + result.probe_host = get_probe_host(observation, blue_endpoint, writer_host, switchover_requested, stop) + target_driver_connect = DriverHelper.get_connect_func(test_driver) + conn = AwsWrapperConnection.connect( + target_driver_connect, **self.get_wrapper_connect_params(conn_utils, result.probe_host, port, db)) + bg_plugin: Optional[BlueGreenPlugin] = conn._unwrap(BlueGreenPlugin) + assert bg_plugin is not None, f"Unable to find blue/green plugin in wrapper connection for {result.probe_host}." + # cursor() goes through the plugin, so a cursor made inside the window would take the hold. + cursor = conn.cursor() + if not wait_for_phase(observation, "IN_PROGRESS", stop): + return + + status = self.read_plugin_status() + result.phase_at_start = None if status is None else status.phase.name + error: Optional[Exception] = None + result.begin() + try: + cursor.execute("SELECT 1") + except Exception as e: + error = e + result.hold_ns = bg_plugin.get_hold_time_ns() + result.phase_at_end = self.read_plugin_phase() + result.end(error) + try: + cursor.fetchall() + except Exception: + pass # the old blue may drop the connection at POST + except Exception as e: + result.outcome = result.outcome or describe_error(e) + finally: + result.done = True + self.close_connection(conn) + + # #1352: a connect to a green host that starts in IN_PROGRESS must end once the phase moves on, without a busy loop. + def wrapper_green_suspended_connect_probe( + self, + test_driver: TestDriver, + conn_utils: ConnectionUtils, + green_host: Optional[str], + port: int, + db: str, + observation: PluginStatusObservation, + stop: Event, + result: ProbeResult) -> None: + conn = None + try: + if green_host is None: + result.outcome = "no green instance in the topology" + return + params = self.get_wrapper_connect_params(conn_utils, green_host, port, db) + params[WrapperProperties.CONNECT_TIMEOUT_SEC.name] = PROBE_CONNECT_TIMEOUT_SEC + target_driver_connect = DriverHelper.get_connect_func(test_driver) + if not wait_for_phase(observation, "IN_PROGRESS", stop): + return + + status = self.read_plugin_status() + result.phase_at_start = None if status is None else status.phase.name + result.suspend_listed, result.suspend_detail = connect_suspend_listed(status, green_host) + error: Optional[Exception] = None + result.begin() + try: + conn = AwsWrapperConnection.connect(target_driver_connect, **params) + except Exception as e: + error = e + result.phase_at_end = self.read_plugin_phase() + result.end(error) + except Exception as e: + result.outcome = result.outcome or describe_error(e) + finally: + result.done = True + self.close_connection(conn) + + # #1352: a new connection to the probe host opened in PREPARATION, then in POST, + # must reach the substitute host the plugin's status lists for it. + def wrapper_blue_routing_probe( + self, + test_driver: TestDriver, + conn_utils: ConnectionUtils, + blue_endpoint: str, + writer_host: str, + port: int, + db: str, + observation: PluginStatusObservation, + switchover_requested: Event, + stop: Event, + preparation_result: ProbeResult, + post_result: ProbeResult) -> None: + try: + probe_host = get_probe_host(observation, blue_endpoint, writer_host, switchover_requested, stop) + preparation_result.probe_host = probe_host + post_result.probe_host = probe_host + params = self.get_wrapper_connect_params(conn_utils, probe_host, port, db) + params[WrapperProperties.CONNECT_TIMEOUT_SEC.name] = PROBE_CONNECT_TIMEOUT_SEC + target_driver_connect = DriverHelper.get_connect_func(test_driver) + if wait_for_phase(observation, "PREPARATION", stop): + self.run_routing_attempts(target_driver_connect, params, port, "PREPARATION", 1, preparation_result) + preparation_result.done = True + if wait_for_phase(observation, "POST", stop): + self.run_routing_attempts(target_driver_connect, params, port, "POST", ROUTING_PROBE_ATTEMPTS, post_result) + except Exception as e: + post_result.outcome = post_result.outcome or describe_error(e) + finally: + preparation_result.done = True + post_result.done = True + + def run_routing_attempts( + self, + target_driver_connect: Any, + params: Dict[str, Any], + port: int, + phase: str, + max_attempts: int, + result: ProbeResult) -> None: + status = self.read_plugin_status() + result.phase_at_start = None if status is None else status.phase.name + if result.phase_at_start != phase: + return + + result.start_ns = perf_counter_ns() + for _ in range(max_attempts): + # Re-read before every attempt: late POST drops the substitutes and connects by name. + status = self.read_plugin_status() + substitute = listed_substitute(status, result.probe_host, port) + attempt = RoutingAttempt("none" if status is None else status.phase.name, substitute) + result.attempts.append(attempt) + if substitute is None: + break + + conn = None + try: + conn = AwsWrapperConnection.connect(target_driver_connect, **params) + if listed_substitute(self.read_plugin_status(), result.probe_host, port) != substitute: + attempt.outcome = "status changed during the attempt" + else: + attempt.driver_host = self.get_driver_host(conn) + attempt.outcome = "ok" + except Exception as e: + attempt.outcome = describe_error(e) + finally: + self.close_connection(conn) + if attempt.outcome == "ok": + break + result.end_ns = perf_counter_ns() + + def get_driver_host(self, conn: AwsWrapperConnection) -> Optional[str]: + """The host the driver connected to (never hostaddr).""" + try: + target = conn.target_connection + if TestEnvironment.get_current().get_engine() == DatabaseEngine.MYSQL: + return str(target.server_host) + return str(target.info.host) + except Exception: + return None + def is_timeout_exception(self, exception: Exception) -> bool: error_message = str(exception).lower() timeout_keywords = [ @@ -1055,7 +1381,10 @@ def bg_switchover_trigger( bg_id: str, start_latch: CountDownLatch, finish_latch: CountDownLatch, - results: Dict[str, BlueGreenResults]): + results: Dict[str, BlueGreenResults], + stop: Event, + switchover_requested: Event, + switchover_requested_ns: AtomicInt): try: start_latch.count_down() @@ -1067,6 +1396,12 @@ def bg_switchover_trigger( result.threads_sync_time.set(sync_time_ns) sleep(30) + # The main thread sets stop when it gives up waiting for the request; it then waits for the API only if + # switchover_requested is set, so the check and the set stay adjacent. + if stop.is_set(): + return + switchover_requested_ns.set(perf_counter_ns()) + switchover_requested.set() test_utility.switchover_blue_green_deployment(bg_id) bg_trigger_time_ns = perf_counter_ns() diff --git a/tests/integration/container/test_blue_green_deployment_async.py b/tests/integration/container/test_blue_green_deployment_async.py index 6ec996d8..6e5edb41 100644 --- a/tests/integration/container/test_blue_green_deployment_async.py +++ b/tests/integration/container/test_blue_green_deployment_async.py @@ -35,7 +35,7 @@ - ``conn.close()`` → ``await conn.close()``. - ``is_closed`` for psycopg AsyncConnection uses the sync ``.closed`` property; for aiomysql uses ``not .open``. See ``_is_conn_closed_async``. -- ``conn._unwrap(BlueGreenPlugin)`` → ``_unwrap_plugin_async(conn, BlueGreenPlugin)`` +- ``conn._unwrap(BlueGreenPlugin)`` → ``_unwrap_plugin_async(conn, AsyncBlueGreenPlugin)`` (searches ``conn._plugin_manager._plugins`` directly because ``AsyncAwsWrapperConnection`` does not expose ``_unwrap``). - ``DriverHelper.get_connect_func`` returns the async connect callable for @@ -52,8 +52,10 @@ import asyncio import math import socket +import sys from collections import deque from dataclasses import dataclass, field +from logging import INFO, getLogger from threading import Event, Thread from time import perf_counter_ns, sleep from typing import (TYPE_CHECKING, Any, Callable, Deque, Dict, List, Optional, @@ -63,11 +65,16 @@ import pytest from tabulate import tabulate # type: ignore[import-untyped] +from aws_advanced_python_wrapper.aio.blue_green_plugin import \ + AsyncBlueGreenPlugin +from aws_advanced_python_wrapper.aio.blue_green_plugin import \ + BlueGreenStatus as AsyncBlueGreenStatus +from aws_advanced_python_wrapper.aio.cleanup import unregister_shutdown_hook from aws_advanced_python_wrapper.aio.wrapper import AsyncAwsWrapperConnection -from aws_advanced_python_wrapper.blue_green_plugin import (BlueGreenPlugin, - BlueGreenRole) +from aws_advanced_python_wrapper.blue_green_plugin import BlueGreenRole from aws_advanced_python_wrapper.database_dialect import DialectCode from aws_advanced_python_wrapper.driver_info import DriverInfo +from aws_advanced_python_wrapper.utils import services_container from aws_advanced_python_wrapper.utils.atomic import AtomicInt from aws_advanced_python_wrapper.utils.concurrent import (ConcurrentDict, CountDownLatch) @@ -76,6 +83,13 @@ from aws_advanced_python_wrapper.utils.rds_utils import RdsUtils from tests.integration.container.utils.async_connection_helpers import \ cleanup_async +from tests.integration.container.utils.blue_green_checks import ( + ASYNC_CONNECT_WAIT_SEC, ASYNC_EXECUTE_TIMEOUT_SEC, BG_CONNECT_TIMEOUT_MS, + MSG_ASYNC_NOT_CONNECTED, PLUGIN_STATUS_POLL_SEC, + SWITCHOVER_TEST_TIMEOUT_SEC, THREAD_JOIN_SEC, PluginStatusObservation, + ProbeResult, choose_probe_host, describe_error, evaluate_suspended_call, + execute_routing_detail, failure_message, join_all, record_plugin_phase, + settle_switchover, summary_lines, wait_for_switchover) from tests.integration.container.utils.conditions import ( enable_on_deployments, enable_on_features) from tests.integration.container.utils.database_engine import DatabaseEngine @@ -88,6 +102,8 @@ TestEnvironmentFeatures if TYPE_CHECKING: + from aws_advanced_python_wrapper.aio.blue_green_plugin import \ + AsyncBlueGreenStatusProvider from tests.integration.container.utils.connection_utils import \ ConnectionUtils from tests.integration.container.utils.test_driver import TestDriver @@ -261,10 +277,15 @@ def test_utility(self): def rds_utils(self): return RdsUtils() + # Worst case on any tree: setup + 30 s (async probe connect) + 150 s (latest request) + 300 s (API wait from the + # request) + 5 s (trigger join) + 2 s (CPU sample) + 20 s (joins), so setup + 507 s. The harness's --timeout=600 + # stays for every other test. + @pytest.mark.timeout(SWITCHOVER_TEST_TIMEOUT_SEC) def test_switchover_async( self, conn_utils: ConnectionUtils, test_utility: RdsTestUtility, rds_utils: RdsUtils, test_environment: TestEnvironment, test_driver: TestDriver): + test_start_ns = perf_counter_ns() self.results.clear() self.unhandled_exceptions.clear() @@ -295,6 +316,7 @@ def test_switchover_async( if rds_utils.is_not_green_or_old_instance(host): threads.append(Thread( + daemon=True, target=self.direct_topology_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -302,6 +324,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.direct_blue_connectivity_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -309,6 +332,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.direct_blue_idle_connectivity_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -316,6 +340,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_blue_idle_connectivity_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -323,6 +348,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_blue_executing_connectivity_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -330,6 +356,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_blue_new_connection_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -337,6 +364,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.blue_dns_monitor, args=(host_id, host, start_latch, stop, finish_latch, bg_results))) thread_count += 1 @@ -344,6 +372,7 @@ def test_switchover_async( if rds_utils.is_green_instance(host): threads.append(Thread( + daemon=True, target=self.direct_topology_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -351,6 +380,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.wrapper_green_connectivity_monitor_async, args=(test_driver, conn_utils, host_id, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, bg_results))) @@ -358,6 +388,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.green_dns_monitor, args=(host_id, host, start_latch, stop, finish_latch, bg_results))) thread_count += 1 @@ -367,6 +398,7 @@ def test_switchover_async( rds_client = test_utility.get_rds_client() threads.append(Thread( + daemon=True, target=self.green_iam_connectivity_monitor_async, args=(test_driver, conn_utils, rds_client, host_id, "BlueHostToken", rds_utils.remove_green_instance_prefix(host), host, test_instance.get_port(), @@ -376,6 +408,7 @@ def test_switchover_async( thread_finish_count += 1 threads.append(Thread( + daemon=True, target=self.green_iam_connectivity_monitor_async, args=(test_driver, conn_utils, rds_client, host_id, "GreenHostToken", host, host, test_instance.get_port(), db_name, start_latch, stop, finish_latch, @@ -384,52 +417,119 @@ def test_switchover_async( thread_count += 1 thread_finish_count += 1 - threads.append(Thread( + switchover_requested = Event() + switchover_requested_ns = AtomicInt(0) + trigger_thread = Thread( + daemon=True, target=self.bg_switchover_trigger, - args=(test_utility, info.get_bg_deployment_id(), start_latch, finish_latch, self.results))) + args=(test_utility, info.get_bg_deployment_id(), start_latch, finish_latch, self.results, stop, + switchover_requested, switchover_requested_ns)) + threads.append(trigger_thread) thread_count += 1 thread_finish_count += 1 + # #1352 probe. Each AsyncAwsWrapperConnection keeps its own Blue/Green status store, so only the + # connection that owns the status provider sees the phases: the probe connects first and is also the observer. + blue_endpoint = test_environment.get_database_info().get_cluster_endpoint() \ + if test_environment.get_deployment() == DatabaseEngineDeployment.AURORA else test_instance.get_host() + probe_host, probe_host_choice = choose_probe_host(None, blue_endpoint, test_instance.get_host()) + observation = PluginStatusObservation() + execute_probe = ProbeResult("AsyncWrapperBlueSuspendedExecute") + probes = [execute_probe] + probe_connected = Event() + probe_thread = Thread( + daemon=True, + target=self.wrapper_blue_suspended_execute_probe_async, + args=(test_driver, conn_utils, probe_host, test_instance.get_port(), db_name, observation, + stop, probe_connected, execute_probe)) + start_latch.set_count(thread_count) finish_latch.set_count(thread_finish_count) for result in self.results.values(): result.start_time_ns.set(start_time_ns) - for thread in threads: - thread.start() - - self.logger.debug("All threads started.") - - finish_latch.wait_sec(6 * 60) - self.logger.debug("All threads completed.") - - sleep(12 * 60) - - self.logger.debug("Stopping all threads...") - stop.set() - - for thread in threads: - thread.join(timeout=30) - if thread.is_alive(): - self.logger.debug("Timed out waiting for a thread to stop running...") - - self.logger.debug("Done waiting for threads to stop.") + # An unfixed wrapper's suspended calls log two debug records per pass of their loop; keep them out of the report. + plugin_logger = getLogger(AsyncBlueGreenPlugin.__module__) + plugin_log_level = plugin_logger.level + plugin_logger.setLevel(INFO) + switch_interval = sys.getswitchinterval() + threads_start_ns = perf_counter_ns() + wait_result = "not run" + try: + # conftest released the services container and its lazy re-initialisation isn't locked: build it here, + # before the monitors' first connects race on it. + services_container.get_monitor_service() + probe_thread.start() + if not probe_connected.wait(ASYNC_CONNECT_WAIT_SEC) or not probe_thread.is_alive(): + stop.set() + for line in summary_lines( + test_start_ns, threads_start_ns, 0, "async probe not connected", observation, probes): + print(line) + pytest.fail(failure_message( + MSG_ASYNC_NOT_CONNECTED, + f"connected {probe_connected.is_set()}, alive {probe_thread.is_alive()}, outcome {execute_probe.outcome or '-'}")) + threads.append(probe_thread) - for host_id, result in self.results.items(): - assert result.bg_trigger_time_ns.get() > 0, \ - f"bg_trigger_time for {host_id} was {result.bg_trigger_time_ns.get()}" + try: + threads_start_ns = perf_counter_ns() + for thread in threads: + if thread is not probe_thread: + thread.start() + + self.logger.debug("All threads started.") + observation.probe_host = probe_host + observation.probe_host_choice = f"{probe_host_choice}, the async probe connects before any status" + wait_result = wait_for_switchover( + probes, observation, switchover_requested, switchover_requested_ns, threads_start_ns, + self.is_switchover_completed_seen) + self.logger.debug(f"Done waiting for the switchover: {wait_result}.") + finally: + settle_switchover( + test_utility, info.get_bg_deployment_id(), trigger_thread, stop, switchover_requested, + switchover_requested_ns) + self.logger.debug("Stop set for all threads.") + + alive_count = join_all(threads, THREAD_JOIN_SEC) + self.logger.debug(f"Done waiting for threads to stop, {alive_count} still running.") + + for line in summary_lines( + test_start_ns, threads_start_ns, switchover_requested_ns.get(), wait_result, observation, probes): + print(line) + + for host_id, result in self.results.items(): + assert result.bg_trigger_time_ns.get() > 0, \ + (f"bg_trigger_time for {host_id} was {result.bg_trigger_time_ns.get()}, wait result: {wait_result}, " + f"unhandled exceptions: {', '.join(describe_error(e) for e in self.unhandled_exceptions) or 'none'}") + + self.logger.debug("Test is over.") + failures = evaluate_suspended_call( + "async execute", execute_probe, perf_counter_ns(), observation.plugin_phase_first_seen_ns.get("POST")) + # A monitor still appending to its results can make print_metrics raise; it must not hide the verdict. + try: + self.print_metrics(rds_utils) + except Exception as e: + print(f"print_metrics failed: {describe_error(e)}") + if failures: + pytest.fail("\n".join(failures)) - self.logger.debug("Test is over.") - self.print_metrics(rds_utils) + if len(self.unhandled_exceptions) > 0: + self.log_unhandled_exceptions() + pytest.fail("There were unhandled exceptions.") - if len(self.unhandled_exceptions) > 0: - self.log_unhandled_exceptions() - pytest.fail("There were unhandled exceptions.") + self.assert_test() - self.assert_test() + self.logger.debug("Completed") + finally: + # Any thread still alive may be an unfixed call spinning (the inherited monitors as well as the probes). + # Restoring the interval or the DEBUG level under it floods the output until the harness can't keep up. + # A lane runs one test. + if not any(probe.is_in_flight() for probe in probes) and not any(thread.is_alive() for thread in threads): + sys.setswitchinterval(switch_interval) + plugin_logger.setLevel(plugin_log_level) - self.logger.debug("Completed") + def is_switchover_completed_seen(self) -> bool: + return any(result.green_status_time.get("SWITCHOVER_COMPLETED", 0) for result in self.results.values()) def get_bg_endpoints( self, @@ -571,6 +671,7 @@ def get_wrapper_connect_params( params = conn_utils.get_connect_params(host=host, port=port, dbname=db) params = {**params, **self.get_telemetry_params()} params[WrapperProperties.CLUSTER_ID.name] = self.TEST_CLUSTER_ID + params[WrapperProperties.BG_CONNECT_TIMEOUT_MS.name] = BG_CONNECT_TIMEOUT_MS test_env = TestEnvironment.get_current() engine = test_env.get_engine() db_deployment = test_env.get_deployment() @@ -668,11 +769,14 @@ def _log_and_return_time(_: Any) -> int: else: results.blue_status_time.compute_if_absent(status, _log_and_return_time) + # Each poll must read the status table as it is now: without autocommit, MySQL's REPEATABLE READ + # repeats the first read's snapshot (the plugin sets autocommit on its status connections for this reason). + connect_params = {**conn_utils.get_connect_params(host=host, port=port, dbname=db), "autocommit": True} + def _body(loop: asyncio.AbstractEventLoop) -> None: conn: Any = None try: - conn = loop.run_until_complete(_open_direct_conn_async( - test_driver, conn_utils.get_connect_params(host=host, port=port, dbname=db))) + conn = loop.run_until_complete(_open_direct_conn_async(test_driver, connect_params)) self.logger.debug(f"[AsyncDirectTopology @ {host_id}] Connection opened.") sleep(1) @@ -684,8 +788,7 @@ def _body(loop: asyncio.AbstractEventLoop) -> None: while not stop.is_set() and perf_counter_ns() < end_time_ns: if conn is None: try: - conn = loop.run_until_complete(_open_direct_conn_async( - test_driver, conn_utils.get_connect_params(host=host, port=port, dbname=db))) + conn = loop.run_until_complete(_open_direct_conn_async(test_driver, connect_params)) self.logger.debug(f"[AsyncDirectTopology @ {host_id}] Connection re-opened.") except Exception: sleep(0.1) @@ -912,7 +1015,7 @@ def _body(loop: asyncio.AbstractEventLoop) -> None: conn: Optional[AsyncAwsWrapperConnection] = None try: conn = loop.run_until_complete(_open()) - bg_plugin: Optional[BlueGreenPlugin] = _unwrap_plugin_async(conn, BlueGreenPlugin) + bg_plugin: Optional[AsyncBlueGreenPlugin] = _unwrap_plugin_async(conn, AsyncBlueGreenPlugin) assert bg_plugin is not None, \ f"Unable to find blue/green plugin in wrapper connection for {host}." self.logger.debug(f"[AsyncWrapperBlueExecute @ {host_id}] Connection opened.") @@ -982,7 +1085,7 @@ def _body(loop: asyncio.AbstractEventLoop) -> None: try: conn = loop.run_until_complete(_open()) end_time_ns = perf_counter_ns() - bg_plugin: Optional[BlueGreenPlugin] = _unwrap_plugin_async(conn, BlueGreenPlugin) + bg_plugin: Optional[AsyncBlueGreenPlugin] = _unwrap_plugin_async(conn, AsyncBlueGreenPlugin) assert bg_plugin is not None, \ f"Unable to find blue/green plugin in wrapper connection for {host}." @@ -998,7 +1101,7 @@ def _body(loop: asyncio.AbstractEventLoop) -> None: end_time_ns = perf_counter_ns() if conn is not None: - bg_plugin = _unwrap_plugin_async(conn, BlueGreenPlugin) + bg_plugin = _unwrap_plugin_async(conn, AsyncBlueGreenPlugin) assert bg_plugin is not None, \ f"Unable to find blue/green plugin in wrapper connection for {host}." results.blue_wrapper_connect_times.append( @@ -1025,6 +1128,96 @@ def _body(loop: asyncio.AbstractEventLoop) -> None: finish_latch.count_down() self.logger.debug(f"[AsyncWrapperBlueNewConnection @ {host_id}] Thread is completed.") + # #1352: an async execute that starts in IN_PROGRESS must return once the phase moves on. One coroutine runs + # for the thread's whole life, so the status provider's tasks keep running on this loop. + def wrapper_blue_suspended_execute_probe_async( + self, + test_driver: TestDriver, + conn_utils: ConnectionUtils, + host: str, + port: int, + db: str, + observation: PluginStatusObservation, + stop: Event, + connected: Event, + result: ProbeResult) -> None: + result.probe_host = host + connect_params = self.get_wrapper_connect_params(conn_utils, host, port, db) + + def _read_phase(bg_plugin: AsyncBlueGreenPlugin) -> Optional[str]: + try: + status = bg_plugin._plugin_service.get_status(AsyncBlueGreenStatus, bg_plugin._bg_id) + except Exception: + return None + return None if status is None else status.phase.name + + async def _execute_in_progress(cursor: Any, bg_plugin: AsyncBlueGreenPlugin) -> None: + result.phase_at_start = "IN_PROGRESS" + error: Optional[Exception] = None + result.begin() + try: + # A blocking regression would never yield to this bound. + await asyncio.wait_for(cursor.execute("SELECT 1"), ASYNC_EXECUTE_TIMEOUT_SEC) + except Exception as e: + error = e + result.hold_ns = bg_plugin.get_hold_time_ns() + result.phase_at_end = _read_phase(bg_plugin) + result.end(error) + try: + await cursor.fetchall() + except Exception: + pass # the old blue may drop the connection at POST + + async def _main() -> None: + conn = await _open_wrapper_conn_async(test_driver, connect_params) + provider: Optional[AsyncBlueGreenStatusProvider] = None + try: + bg_plugin: Optional[AsyncBlueGreenPlugin] = _unwrap_plugin_async(conn, AsyncBlueGreenPlugin) + assert bg_plugin is not None, f"Unable to find blue/green plugin in wrapper connection for {host}." + # cursor() is sync and lazy. The warm-up resolves the driver cursor, so the execute in the window is a + # single plugin pass and its hold time isn't reset. + cursor = conn.cursor() + await cursor.execute("SELECT 1") + await cursor.fetchall() + # Every async worker drains the process-wide shutdown hooks when it ends, and several end on their + # first lost connection, mid-switchover. Take this connection's status provider out of that drain + # before the other threads start, and stop it here instead. + provider = AsyncBlueGreenPlugin._status_providers.get(bg_plugin._bg_id) + if provider is not None: + unregister_shutdown_hook(provider.stop) + connected.set() + + executed = False + while not stop.is_set(): + status = bg_plugin._plugin_service.get_status(AsyncBlueGreenStatus, bg_plugin._bg_id) + phase = record_plugin_phase(observation.plugin_phase_first_seen_ns, observation.phase_events, status) + if phase == "IN_PROGRESS" and not executed: + executed = True + # Diagnostic only: shows which branch of the plugin's execute a call that isn't held took. + result.routing_detail = execute_routing_detail(status, bg_plugin._plugin_service) + await _execute_in_progress(cursor, bg_plugin) + elif phase in ("POST", "COMPLETED"): + executed = True # IN_PROGRESS was missed; the check reports not run + await asyncio.sleep(PLUGIN_STATUS_POLL_SEC) + finally: + if provider is not None: + observation.provider_monitors_running = any(monitor.is_running() for monitor in provider._monitors) + try: + await provider.stop() + except Exception: + pass + await _close_conn_async(conn) + + def _body(loop: asyncio.AbstractEventLoop) -> None: + loop.run_until_complete(_main()) + + try: + _run_worker_with_loop(_body) + except Exception as e: + result.outcome = result.outcome or describe_error(e) + finally: + result.done = True + # Blue / Green DNS monitors are purely sync (socket calls); reuse verbatim. def blue_dns_monitor( @@ -1092,7 +1285,7 @@ def _body(loop: asyncio.AbstractEventLoop) -> None: conn = loop.run_until_complete(_open()) self.logger.debug(f"[AsyncWrapperGreenConnectivity @ {host_id}] Connection opened.") - bg_plugin: Optional[BlueGreenPlugin] = _unwrap_plugin_async(conn, BlueGreenPlugin) + bg_plugin: Optional[AsyncBlueGreenPlugin] = _unwrap_plugin_async(conn, AsyncBlueGreenPlugin) assert bg_plugin is not None, \ f"Unable to find blue/green plugin in wrapper connection for {host}." @@ -1273,7 +1466,10 @@ def bg_switchover_trigger( bg_id: str, start_latch: CountDownLatch, finish_latch: CountDownLatch, - results: Any) -> None: + results: Any, + stop: Event, + switchover_requested: Event, + switchover_requested_ns: AtomicInt) -> None: """Trigger BG switchover using RDS API (synchronous boto3 calls).""" try: start_latch.count_down() @@ -1284,6 +1480,12 @@ def bg_switchover_trigger( result.threads_sync_time.set(sync_time_ns) sleep(30) + # The main thread sets stop when it gives up waiting for the request; it then waits for the API only if + # switchover_requested is set, so the check and the set stay adjacent. + if stop.is_set(): + return + switchover_requested_ns.set(perf_counter_ns()) + switchover_requested.set() test_utility.switchover_blue_green_deployment(bg_id) bg_trigger_time_ns = perf_counter_ns() diff --git a/tests/integration/container/utils/blue_green_checks.py b/tests/integration/container/utils/blue_green_checks.py new file mode 100644 index 00000000..be3ede75 --- /dev/null +++ b/tests/integration/container/utils/blue_green_checks.py @@ -0,0 +1,548 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"). +# You may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Checks shared by the sync and async Blue/Green switchover tests. + +The switchover tests run probes that hit the plugin's PREPARATION, IN_PROGRESS +and POST phases and then judge them here. A suspended call must return once the +phase moves on, without a busy loop, and a new connection to a blue host must go +to the substitute host the plugin's status lists for it. This module holds the +pure logic, so it imports no database driver. +""" + +from __future__ import annotations + +import os +import sys +from dataclasses import dataclass, field +from threading import Event, Lock, get_ident, get_native_id +from time import perf_counter_ns, sleep, thread_time +from typing import (TYPE_CHECKING, Any, Callable, Dict, List, Optional, + Sequence, Tuple) + +from aws_advanced_python_wrapper.blue_green_plugin import ( + BlueGreenRole, SubstituteConnectRouting, SuspendConnectRouting) +from aws_advanced_python_wrapper.hostinfo import HostInfo +from aws_advanced_python_wrapper.utils.properties import WrapperProperties + +if TYPE_CHECKING: + from threading import Thread + + from aws_advanced_python_wrapper.utils.atomic import AtomicInt + +# No bg_id is set in the test's connect params, so the plugin keys its status by the property default, normalized as +# BlueGreenPlugin.__init__ does. +BG_ID = str(WrapperProperties.BG_ID.default_value).strip().lower() +# Pinned in the switchover tests' connect params, at its default. +BG_CONNECT_TIMEOUT_MS = 30_000 +# Twice bg_connect_timeout_ms (pinned to 30 s in the connect params), which caps a correct suspend. Same 60 s as JDBC's +# MAX_TRAFFIC_SUSPENSION_MS (bgSwitchoverTimeoutMs / 3), derived differently. +MAX_TRAFFIC_SUSPENSION_NS = 2 * BG_CONNECT_TIMEOUT_MS * 1_000_000 +# A correct suspend waits on the status condition in 50 ms steps; the unfixed loop burned about a whole core. +MAX_SUSPENDED_CALL_CPU_SEC = 1.0 +# Thread CPU sample taken on a call that is still in flight past MAX_TRAFFIC_SUSPENSION_NS. +SPIN_SAMPLE_SEC = 2.0 +# Past the main thread's sample of a stuck call: the 60 s bound, a 1 s wait tick and the 2 s CPU sample, with headroom. +ASYNC_EXECUTE_TIMEOUT_SEC = MAX_TRAFFIC_SUSPENSION_NS // 1_000_000_000 + 15 +# A spinning thread shares the GIL with the other spinners and still gets far more than this in 2 s; a blocked read gets close to 0. +SPIN_MIN_CPU_SEC = 0.05 +# The plugin publishes each phase for at least 1.7 s (measured), so a 20 ms poll sees every phase. +PLUGIN_STATUS_POLL_SEC = 0.02 +# Poll step for probes waiting on a phase, and the main thread's wait tick. +PROBE_WAIT_STEP_SEC = 0.5 +WAIT_TICK_SEC = 1.0 +# Connect timeout of each routing and connect probe attempt; POST lasts 23 to 26 s (measured). +PROBE_CONNECT_TIMEOUT_SEC = 10 +ROUTING_PROBE_ATTEMPTS = 3 +# The async probe must connect, and own the status provider, before any other thread starts. +ASYNC_CONNECT_WAIT_SEC = 30 +# From thread start: the trigger waits for the start latch and then 30 s before it requests the switchover. +TRIGGER_WAIT_SEC = 150 +# From the switchover request: the plugin reached COMPLETED 41 to 45 s after it (measured). +SWITCHOVER_WAIT_SEC = 240 +# From the switchover request: the RDS API reported SWITCHOVER_COMPLETED after 68 to 73 s (measured). +BG_API_WAIT_SEC = 300 +BG_API_POLL_SEC = 5.0 +TRIGGER_JOIN_SEC = 5 +# One deadline shared by every join. +THREAD_JOIN_SEC = 20 +# See the worst case on the switchover tests' marker. +SWITCHOVER_TEST_TIMEOUT_SEC = 900 +# Lets the main thread get the GIL often while stuck probes spin. +MIN_SWITCH_INTERVAL_SEC = 0.0005 + +PHASE_ORDER = ("PREPARATION", "IN_PROGRESS", "POST", "COMPLETED") + +MSG_SPIN_EXECUTE = "#1352 spin - sync execute suspended in IN_PROGRESS did not return within 60 s" +MSG_SPIN_CONNECT = "#1352 spin - sync connect suspended in IN_PROGRESS did not return within 60 s" +MSG_SPIN_ASYNC_EXECUTE = "#1352 spin - async execute suspended in IN_PROGRESS did not return within 60 s" +MSG_BUSY_LOOP = "#1352 busy loop - a suspended call used more than 1.0 s of its thread CPU" +MSG_ROUTING_POST = "#1352 routing - connection opened in POST went to the dialled host, not the listed substitute" +MSG_ROUTING_PREPARATION = "#1352 routing - connection opened in PREPARATION went to the dialled host, not the listed substitute" +MSG_BLOCKED = "#1352 check inconclusive - a suspended probe is blocked, not spinning" +MSG_UNEXPECTED_HOST = "#1352 check inconclusive - a probe connection went to an unexpected host" +MSG_EXECUTE_NOT_IN_PROGRESS = "#1352 check not run - the execute probe did not start in IN_PROGRESS" +MSG_EXECUTE_NOT_HELD = "#1352 check not run - the execute probe call was not held by the plugin" +MSG_ASYNC_EXECUTE_NOT_IN_PROGRESS = "#1352 check not run - the async execute probe did not start in IN_PROGRESS" +MSG_ASYNC_EXECUTE_NOT_HELD = "#1352 check not run - the async execute probe call was not held by the plugin" +MSG_CONNECT_NOT_IN_PROGRESS = "#1352 check not run - the connect probe did not start in IN_PROGRESS" +MSG_CONNECT_NOT_SUSPENDED = "#1352 check not run - the connect probe was not suspended by the plugin" +MSG_ROUTING_NO_SUBSTITUTE = "#1352 check not run - the routing probe saw no POST status with a substitute routing for its host" +MSG_ROUTING_NO_CONNECTION = "#1352 check not run - the routing probe opened no connection while a substitute was listed" +MSG_DRIVER_HOST_UNREADABLE = "#1352 check not run - the driver host of a probe connection could not be read" +MSG_ASYNC_NOT_CONNECTED = "#1352 check not run - the async execute probe did not connect before the other threads" + +# label -> (spin, not started in IN_PROGRESS, not held); connect has no hold. +_SUSPENDED_CALL_MESSAGES: Dict[str, Tuple[str, str, Optional[str]]] = { + "execute": (MSG_SPIN_EXECUTE, MSG_EXECUTE_NOT_IN_PROGRESS, MSG_EXECUTE_NOT_HELD), + "connect": (MSG_SPIN_CONNECT, MSG_CONNECT_NOT_IN_PROGRESS, None), + "async execute": (MSG_SPIN_ASYNC_EXECUTE, MSG_ASYNC_EXECUTE_NOT_IN_PROGRESS, MSG_ASYNC_EXECUTE_NOT_HELD), +} +_ROUTING_MESSAGES: Dict[str, str] = {"POST": MSG_ROUTING_POST, "PREPARATION": MSG_ROUTING_PREPARATION} + + +@dataclass +class RoutingAttempt: + phase: str + substitute_host: Optional[str] # None when the status listed no substitute for the probe host + driver_host: Optional[str] = None + outcome: str = "" + + +@dataclass +class ProbeResult: + name: str + probe_host: str = "" + phase_at_start: Optional[str] = None + # The plugin phase read right after a suspended call returned. + phase_at_end: Optional[str] = None + start_ns: int = 0 + end_ns: int = 0 # 0 while the call is in flight + hold_ns: int = 0 + cpu_sec: float = 0.0 + outcome: str = "" # "ok", or the exception class and the start of its message + suspend_listed: bool = False + suspend_detail: str = "" + # For the summary only: how the plugin's execute would route this host under the status that started the call. + routing_detail: str = "" + attempts: List[RoutingAttempt] = field(default_factory=list) + native_id: Optional[int] = None + ident: Optional[int] = None + done: bool = False + # Filled by the main thread when the call is still in flight. + spinning: Optional[bool] = None + spin_cpu_delta_sec: Optional[float] = None + frames: Tuple[str, str] = ("none", "none") + _cpu_start_sec: float = field(default=0.0, repr=False) + + def begin(self) -> None: + """Called on the probe's own thread right before the call.""" + self.native_id = get_native_id() + self.ident = get_ident() + self._cpu_start_sec = thread_time() + self.start_ns = perf_counter_ns() + + def end(self, error: Optional[BaseException]) -> None: + """Called on the probe's own thread right after the call.""" + self.cpu_sec = thread_time() - self._cpu_start_sec + self.outcome = "ok" if error is None else describe_error(error) + self.end_ns = perf_counter_ns() + + def in_flight_ns(self, now_ns: int) -> int: + if self.start_ns == 0: + return 0 + return (self.end_ns if self.end_ns > 0 else now_ns) - self.start_ns + + def is_in_flight(self) -> bool: + return self.start_ns > 0 and self.end_ns == 0 + + def is_stuck(self, now_ns: int) -> bool: + return self.is_in_flight() and now_ns - self.start_ns > MAX_TRAFFIC_SUSPENSION_NS + + +@dataclass +class PluginStatusObservation: + """What the plugin published, as the test's observer saw it.""" + plugin_phase_first_seen_ns: Dict[str, int] = field(default_factory=dict) + phase_events: Dict[str, Event] = field(default_factory=lambda: {phase: Event() for phase in PHASE_ORDER}) + status_seen: Event = field(default_factory=Event) + pre_trigger_status: Any = None + error: str = "" + probe_host: Optional[str] = None + probe_host_choice: str = "" + # Async only: whether the status provider's monitors were still running at the end of the window. + provider_monitors_running: Optional[bool] = None + lock: Lock = field(default_factory=Lock) + + +def describe_error(error: BaseException) -> str: + return f"{type(error).__name__}: {str(error)[:200]}" + + +def _clean(text: str) -> str: + return text.replace(";", ",").replace("|", "/").replace("\n", " ") + + +def failure_message(text: str, detail: str) -> str: + return f"{text} ({_clean(detail)})" + + +def _sec(ns: int) -> str: + return f"{ns / 1_000_000_000:.2f} s" + + +def record_plugin_phase( + phase_first_seen_ns: Dict[str, int], + phase_events: Dict[str, Event], + status: Any) -> Optional[str]: + """Record the first time a phase is seen; works for the sync and the aio status classes.""" + if status is None: + return None + name: str = status.phase.name + if name not in phase_first_seen_ns: + phase_first_seen_ns[name] = perf_counter_ns() + event = phase_events.get(name) + if event is not None: + event.set() + return name + + +def wait_for_phase(observation: PluginStatusObservation, phase: str, stop: Event) -> bool: + """Wait for the phase; False on stop or when a later phase was seen without it.""" + later = PHASE_ORDER[PHASE_ORDER.index(phase) + 1:] + while not stop.is_set(): + if observation.phase_events[phase].is_set(): + return True + if any(observation.phase_events[p].is_set() for p in later): + return False + observation.phase_events[phase].wait(PROBE_WAIT_STEP_SEC) + return False + + +def choose_probe_host(status: Any, cluster_endpoint: str, writer_host: str) -> Tuple[str, str]: + if status is None: + return cluster_endpoint, "no pre-trigger status" + if status.get_role(HostInfo(host=cluster_endpoint)) is not None: + return cluster_endpoint, "cluster endpoint" + return writer_host, "writer instance" + + +def get_probe_host( + observation: PluginStatusObservation, + cluster_endpoint: str, + writer_host: str, + switchover_requested: Event, + stop: Event) -> str: + """Wait for the first published status (or the request), then choose the probe host once for every probe.""" + while not (observation.status_seen.is_set() or switchover_requested.is_set() or stop.is_set()): + observation.status_seen.wait(PROBE_WAIT_STEP_SEC) + with observation.lock: + if observation.probe_host is None: + observation.probe_host, observation.probe_host_choice = \ + choose_probe_host(observation.pre_trigger_status, cluster_endpoint, writer_host) + return observation.probe_host + + +def listed_substitute(status: Any, probe_host: str, port: int) -> Optional[str]: + """The substitute host the status lists for the probe host. Reads the routings; never calls is_match, the code under test.""" + if status is None: + return None + host = probe_host.lower() + endpoints = (host, f"{host}:{port}") + for routing in status.connect_routings: + if isinstance(routing, SubstituteConnectRouting) and routing._endpoint is not None \ + and routing._endpoint.lower() in endpoints: + return routing._substitute_host_info.host + return None + + +def connect_suspend_listed(status: Any, host: str) -> Tuple[bool, str]: + if status is None: + return False, "no status" + role = status.get_role(HostInfo(host=host)) + if role != BlueGreenRole.TARGET: + return False, f"role of the connect probe host was {role}" + for routing in status.connect_routings: + if isinstance(routing, SuspendConnectRouting) and routing._endpoint is None \ + and routing._bg_role == BlueGreenRole.TARGET: + return True, "SuspendConnectRouting(None, TARGET) listed" + return False, "no SuspendConnectRouting(None, TARGET) listed" + + +def execute_routing_detail(status: Any, plugin_service: Any) -> str: + """The current host's role and first matching execute routing, as the plugin's execute picks them. Summary only.""" + try: + if status is None: + return "routing: no status" + host_info = plugin_service.current_host_info + role = status.get_role(host_info) + routing = None if role is None else next( + (r for r in status.execute_routings if r.is_match(host_info, role)), None) + host = None if host_info is None else host_info.get_host_and_port() + return (f"current host {host}, role {role}, first matching execute routing {routing}, " + f"roles by endpoint {sorted(status.roles_by_endpoint.keys())}") + except Exception as e: + return f"routing: {describe_error(e)}" + + +def thread_cpu_sec(native_id: Optional[int]) -> Optional[float]: + """utime + stime of one thread of this process (Linux); None where /proc is missing.""" + if native_id is None: + return None + try: + with open(f"/proc/self/task/{native_id}/stat") as stat_file: + stat = stat_file.read() + fields = stat[stat.rindex(")") + 2:].split() + # Fields after the command name start at field 3 (state); utime and stime are fields 14 and 15. + return (int(fields[11]) + int(fields[12])) / os.sysconf("SC_CLK_TCK") + except (OSError, ValueError, IndexError): + return None + + +def stuck_frames(ident: Optional[int]) -> Tuple[str, str]: + """The innermost frame of a thread and its innermost frame in the wrapper, as file:function (never arguments).""" + frame = None if ident is None else sys._current_frames().get(ident) + if frame is None: + return "none", "none" + innermost = f"{os.path.basename(frame.f_code.co_filename)}:{frame.f_code.co_name}" + wrapper_frame = "none" + marker = f"aws_advanced_python_wrapper{os.sep}" + current: Any = frame + while current is not None: + filename = current.f_code.co_filename + if marker in filename: + wrapper_frame = f"{filename[filename.rindex(marker):]}:{current.f_code.co_name}" + break + current = current.f_back + return innermost, wrapper_frame + + +def is_spinning(cpu_before: Optional[float], cpu_after: Optional[float]) -> bool: + if cpu_before is None or cpu_after is None: + return True + return cpu_after - cpu_before >= SPIN_MIN_CPU_SEC + + +def sample_in_flight_probes(probes: Sequence[ProbeResult]) -> None: + """Sample the thread CPU of in-flight probes once over SPIN_SAMPLE_SEC, and lower the switch interval first.""" + if not probes: + return + sys.setswitchinterval(MIN_SWITCH_INTERVAL_SEC) + before = [thread_cpu_sec(probe.native_id) for probe in probes] + sleep(SPIN_SAMPLE_SEC) + for probe, cpu_before in zip(probes, before): + cpu_after = thread_cpu_sec(probe.native_id) + probe.spinning = is_spinning(cpu_before, cpu_after) + probe.spin_cpu_delta_sec = None if cpu_before is None or cpu_after is None else cpu_after - cpu_before + probe.frames = stuck_frames(probe.ident) + + +def _is_settled(probe: ProbeResult, now_ns: int, completed: bool) -> bool: + return probe.done or probe.end_ns > 0 or probe.is_stuck(now_ns) or (probe.start_ns == 0 and completed) + + +def wait_for_switchover( + probes: Sequence[ProbeResult], + observation: PluginStatusObservation, + switchover_requested: Event, + switchover_requested_ns: AtomicInt, + threads_start_ns: int, + completed_seen: Callable[[], bool]) -> str: + """Wait until every probe settled and the switchover completed (or a probe is stuck), within the stated bounds.""" + sampled: List[str] = [] + while True: + now_ns = perf_counter_ns() + requested = switchover_requested.is_set() + if requested and now_ns >= switchover_requested_ns.get() + SWITCHOVER_WAIT_SEC * 1_000_000_000: + reason = f"bound reached: switchover request + {SWITCHOVER_WAIT_SEC} s" + break + if not requested and now_ns >= threads_start_ns + TRIGGER_WAIT_SEC * 1_000_000_000: + reason = f"bound reached: no switchover request {TRIGGER_WAIT_SEC} s after thread start" + break + + newly_stuck = [p for p in probes if p.is_stuck(now_ns) and p.name not in sampled] + if newly_stuck: + sample_in_flight_probes(newly_stuck) + sampled.extend(p.name for p in newly_stuck) + continue + + completed = observation.phase_events["COMPLETED"].is_set() or completed_seen() + if requested and all(_is_settled(p, now_ns, completed) for p in probes) \ + and (completed or any(p.is_stuck(now_ns) for p in probes)): + return "probes settled" + (", switchover completed" if completed else ", a probe is stuck") + sleep(WAIT_TICK_SEC) + + sample_in_flight_probes([p for p in probes if p.is_in_flight() and p.name not in sampled]) + return reason + + +def evaluate_suspended_call( + label: str, + result: ProbeResult, + now_ns: int, + post_first_seen_ns: Optional[int]) -> List[str]: + spin_message, not_run_message, not_held_message = _SUSPENDED_CALL_MESSAGES[label] + # A probe sampled past the bound as blocked gets the inconclusive text; an unsampled or spinning one the defect's. + stuck_message = MSG_BLOCKED if result.spinning is False else spin_message + if result.phase_at_start != "IN_PROGRESS": + return [failure_message(not_run_message, f"{result.name} phase at start {result.phase_at_start}, outcome {result.outcome or '-'}")] + if label == "connect" and not result.suspend_listed: + return [failure_message(MSG_CONNECT_NOT_SUSPENDED, f"{result.name} {result.suspend_detail}")] + + if result.is_in_flight(): + cpu_delta = "unread" if result.spin_cpu_delta_sec is None else f"{result.spin_cpu_delta_sec:.3f} s" + detail = (f"{result.name} on {result.probe_host} in flight {_sec(result.in_flight_ns(now_ns))}, " + f"CPU over {SPIN_SAMPLE_SEC} s sample {cpu_delta}, " + f"innermost frame {result.frames[0]}, wrapper frame {result.frames[1]}") + return [failure_message(stuck_message, detail)] + + returned = f"{result.name} returned after {_sec(result.in_flight_ns(now_ns))} with {result.outcome}" + if result.in_flight_ns(now_ns) > MAX_TRAFFIC_SUSPENSION_NS or result.hold_ns > MAX_TRAFFIC_SUSPENSION_NS: + return [failure_message(stuck_message, f"{returned}, hold {_sec(result.hold_ns)}, over the 60 s bound")] + if result.phase_at_end == "IN_PROGRESS": + # Released by the plugin's suspend timer, not by the phase change. + return [failure_message(not_run_message, f"{returned}, plugin phase still IN_PROGRESS when it returned")] + if not_held_message is not None and result.hold_ns == 0: + return [failure_message(not_held_message, returned)] + if label == "connect" and (post_first_seen_ns is None or result.end_ns < post_first_seen_ns - 1_000_000_000): + return [failure_message(MSG_CONNECT_NOT_SUSPENDED, f"{returned}, before the plugin's POST")] + if result.cpu_sec > MAX_SUSPENDED_CALL_CPU_SEC: + return [failure_message(MSG_BUSY_LOOP, f"{returned}, thread CPU {result.cpu_sec:.3f} s")] + return [] + + +def evaluate_routing(result: ProbeResult, phase_label: str, required: bool, writer_ips: str) -> List[str]: + listed = [a for a in result.attempts if a.substitute_host is not None and a.phase == phase_label] + if result.phase_at_start != phase_label or not listed: + if not required: + return [] + detail = f"{result.name} phase at start {result.phase_at_start}, {len(result.attempts)} attempts, none with a substitute" + return [failure_message(MSG_ROUTING_NO_SUBSTITUTE, detail)] + + succeeded = [a for a in listed if a.outcome == "ok"] + if not succeeded: + return [failure_message(MSG_ROUTING_NO_CONNECTION, f"{result.name} in {phase_label}, last error {listed[-1].outcome}")] + + attempt = succeeded[0] + if attempt.driver_host is None: + return [failure_message(MSG_DRIVER_HOST_UNREADABLE, f"{result.name} in {phase_label}")] + + driver_host = attempt.driver_host.lower() + substitute = (attempt.substitute_host or "").lower() + detail = (f"{result.name} in {phase_label} dialled {result.probe_host}, listed substitute {attempt.substitute_host}, " + f"driver host {attempt.driver_host}, {writer_ips}") + if driver_host == substitute: + return [] + if driver_host == result.probe_host.lower(): + return [failure_message(_ROUTING_MESSAGES[phase_label], detail)] + return [failure_message(MSG_UNEXPECTED_HOST, detail)] + + +def join_all(threads: Sequence[Thread], sec: float) -> int: + """Join every thread within one shared deadline; returns how many are still alive.""" + deadline_ns = perf_counter_ns() + int(sec * 1_000_000_000) + for thread in threads: + remaining = (deadline_ns - perf_counter_ns()) / 1_000_000_000 + if remaining <= 0: + break + thread.join(remaining) + return sum(1 for thread in threads if thread.is_alive()) + + +def wait_for_bg_deployment_status( + test_utility: Any, + bg_id: str, + deadline_ns: int, + poll_sec: float = BG_API_POLL_SEC) -> Optional[str]: + """Poll the RDS API until the deployment reports a finished switchover or the deadline passes. Never raises.""" + status: Optional[str] = None + while True: + try: + deployment = test_utility.get_blue_green_deployment(bg_id) + status = None if deployment is None else deployment.get("Status") + except Exception as e: + print(f"Blue/Green deployment status read failed: {describe_error(e)}") + status = None + if status in ("SWITCHOVER_COMPLETED", "SWITCHOVER_FAILED"): + return status + remaining = (deadline_ns - perf_counter_ns()) / 1_000_000_000 + if remaining <= 0: + print(f"Blue/Green deployment status still {status} at the API wait deadline.") + return status + sleep(min(poll_sec, remaining)) + + +def settle_switchover( + test_utility: Any, + bg_id: str, + trigger_thread: Thread, + stop: Event, + switchover_requested: Event, + switchover_requested_ns: AtomicInt) -> None: + """Stop the trigger if it has not requested the switchover, else wait for the RDS API to finish it. Always sets stop.""" + try: + if not switchover_requested.is_set(): + stop.set() + trigger_thread.join(TRIGGER_JOIN_SEC) + if switchover_requested.is_set(): + wait_for_bg_deployment_status( + test_utility, bg_id, switchover_requested_ns.get() + BG_API_WAIT_SEC * 1_000_000_000) + finally: + stop.set() + + +def summary_lines( + test_start_ns: int, + threads_start_ns: int, + switchover_requested_ns: int, + wait_result: str, + observation: PluginStatusObservation, + probes: Sequence[ProbeResult], + extra: Sequence[str] = ()) -> List[str]: + zero_ns = switchover_requested_ns if switchover_requested_ns > 0 else threads_start_ns + zero_name = "switchover request" if switchover_requested_ns > 0 else "thread start" + lines = [ + "#1352 checks summary", + f" test entry to thread start: {_sec(threads_start_ns - test_start_ns)}", + f" probe host: {observation.probe_host} ({observation.probe_host_choice or 'not chosen'})", + f" wait: {wait_result}", + f" observer error: {observation.error or '-'}", + ] + if observation.provider_monitors_running is not None: + lines.append(f" status provider monitors running at the end of the window: {observation.provider_monitors_running}") + lines.extend(f" {line}" for line in extra) + for phase in PHASE_ORDER: + seen_ns = observation.plugin_phase_first_seen_ns.get(phase) + offset = "not seen" if seen_ns is None else f"{_sec(seen_ns - zero_ns)} after {zero_name}" + lines.append(f" plugin phase {phase}: {offset}") + for probe in probes: + timing = "not started" if probe.start_ns == 0 else f"start {_sec(probe.start_ns - zero_ns)}, " + ( + "in flight" if probe.end_ns == 0 else f"end {_sec(probe.end_ns - zero_ns)}") + line = f" {probe.name}: host {probe.probe_host or '-'}, phase at start {probe.phase_at_start}, " + if probe.phase_at_end is not None: + line += f"phase at end {probe.phase_at_end}, " + line += (f"{timing}, hold {_sec(probe.hold_ns)}, thread CPU {probe.cpu_sec:.3f} s, " + f"outcome {probe.outcome or '-'}") + if probe.suspend_detail: + line += f", suspend listed {probe.suspend_listed} {probe.suspend_detail}" + if probe.routing_detail: + line += f", {probe.routing_detail}" + if probe.spinning is not None: + line += (f", spinning {probe.spinning}, sample CPU {probe.spin_cpu_delta_sec}, " + f"frames {probe.frames[0]} / {probe.frames[1]}") + lines.append(line) + for index, attempt in enumerate(probe.attempts, 1): + lines.append(f" attempt {index} in {attempt.phase}: substitute {attempt.substitute_host}, " + f"driver host {attempt.driver_host}, outcome {attempt.outcome or '-'}") + return lines diff --git a/tests/integration/host/src/test/java/integration/host/TestEnvironment.java b/tests/integration/host/src/test/java/integration/host/TestEnvironment.java index fcd32a3a..44d3fe4e 100644 --- a/tests/integration/host/src/test/java/integration/host/TestEnvironment.java +++ b/tests/integration/host/src/test/java/integration/host/TestEnvironment.java @@ -591,6 +591,11 @@ private static void createDbCluster(TestEnvironment env, int numOfInstances) { env.auroraUtil.deleteCluster(env.rdsDbName, env.info.getRequest().getDatabaseEngineDeployment(), false); LOGGER.finer("Deleted cluster " + env.rdsDbName); + // Teardown never runs after a failed create, so the custom parameter group made for it goes here. + if (!StringUtils.isNullOrEmpty(env.info.getClusterParameterGroupName())) { + env.deleteCustomClusterParameterGroup(env.info.getClusterParameterGroupName()); + } + throw new RuntimeException(e); } } @@ -1355,6 +1360,20 @@ private void deleteBlueGreenDeployment() throws InterruptedException { return; } + // Settle an in-flight switchover: the status read below picks the cluster to delete. A deployment still + // switching after 15 minutes still stops the clean-up in deleteBlueGreenDeployment, as in JDBC. + try { + auroraUtil.waitUntilBlueGreenDeploymentHasRightState( + this.info.getBlueGreenDeploymentId(), "available", "switchover_completed", "switchover_failed"); + } catch (RuntimeException e) { + LOGGER.finest("BGD did not settle: " + e.getMessage()); + } + final BlueGreenDeployment settledDeployment = + auroraUtil.getBlueGreenDeployment(this.info.getBlueGreenDeploymentId()); + if (settledDeployment != null) { + blueGreenDeployment = settledDeployment; + } + auroraUtil.deleteBlueGreenDeployment(this.info.getBlueGreenDeploymentId(), true); // Remove extra DB cluster diff --git a/tests/integration/host/src/test/java/integration/util/AuroraTestUtility.java b/tests/integration/host/src/test/java/integration/util/AuroraTestUtility.java index 1b8684ed..2bba3d41 100644 --- a/tests/integration/host/src/test/java/integration/util/AuroraTestUtility.java +++ b/tests/integration/host/src/test/java/integration/util/AuroraTestUtility.java @@ -912,6 +912,10 @@ public DatabaseEngine getRdsInstanceEngine(final DBInstance instance) { } public String getAuroraParameterGroupFamily(String engine, String engineVersion) { + final String family = queryParameterGroupFamily(engine, engineVersion); + if (!StringUtils.isNullOrEmpty(family)) { + return family; + } switch (engine) { case "aurora-postgresql": return "aurora-postgresql16"; @@ -925,6 +929,29 @@ public String getAuroraParameterGroupFamily(String engine, String engineVersion) } } + /** + * Asks RDS for the parameter group family of an engine version, or returns null when the version is + * empty or the family cannot be determined. + */ + private @Nullable String queryParameterGroupFamily(String engine, String engineVersion) { + if (StringUtils.isNullOrEmpty(engineVersion)) { + return null; + } + try { + final DescribeDbEngineVersionsResponse response = rdsClient.describeDBEngineVersions( + DescribeDbEngineVersionsRequest.builder().engine(engine).engineVersion(engineVersion).build()); + return response.dbEngineVersions().stream() + .map(DBEngineVersion::dbParameterGroupFamily) + .filter(family -> !StringUtils.isNullOrEmpty(family)) + .findFirst() + .orElse(null); + } catch (Exception ex) { + LOGGER.finest(String.format( + "Could not query the parameter group family of %s %s: %s", engine, engineVersion, ex.getMessage())); + return null; + } + } + public List getTestInstancesInfo(final String clusterId) { List dbInstances = getDBInstances(clusterId); List instancesInfo = new ArrayList<>(); @@ -1247,7 +1274,7 @@ public void deleteBlueGreenDeployment(String blueGreenId, boolean waitForComplet return; } - waitUntilBlueGreenDeploymentHasRightState(blueGreenId, "available", "switchover_completed"); + waitUntilBlueGreenDeploymentHasRightState(blueGreenId, "available", "switchover_completed", "switchover_failed"); DeleteBlueGreenDeploymentResponse response = rdsClient.deleteBlueGreenDeployment( DeleteBlueGreenDeploymentRequest.builder()