From 30e67166ca3bb3f62e97fa492d9f6f49c31bed02 Mon Sep 17 00:00:00 2001 From: Juan Lee Date: Fri, 13 Nov 2026 02:31:29 -0700 Subject: [PATCH] test: assert Blue/Green switchover calls resume and are routed The Blue/Green switchover tests now check what #1352 broke, inside the one switchover each lane runs. Probes start on the phase the plugin publishes. An execute on an open blue connection and a new connect to a green instance, both started in IN_PROGRESS, must return within 60 s, after the phase moved on, and use at most 1.0 s of their thread's CPU. The 60 s is twice bg_connect_timeout_ms, which the tests pin at its 30 s default. A new connection to the blue endpoint opened in POST (and in PREPARATION, when sampled) must reach the substitute host the plugin's status lists for it. A call still in flight after 60 s is reported as a spin or as blocked from a thread CPU sample, so a hung wrapper fails the test instead of hanging the suite. The plugin's logger is held at INFO during the test, and after it while any test thread is still running, so a spinning call can't flood the report with debug records. The switchover metrics are printed before the verdict, and an error while printing them can't hide it. The old test asserted nothing about the wrapper's calls or where new connections went, left hung threads running, and could not finish under the harness's 600 s pytest timeout: it waited 6 minutes and then slept 12. It now waits until the switchover completes, bounded from the request, then for the RDS API to report completion. Every thread is a daemon and the joins share one deadline. The services container is built on the main thread before the monitor threads start: conftest releases it and its lazy re-initialisation isn't locked, so their first connects could find no monitor service. If the switchover is never requested, the failure names the wait result and any unhandled exceptions. The direct topology monitors now poll with autocommit on, as the plugin's own status connections do: without it MySQL's REPEATABLE READ kept returning the first snapshot, so they never saw the switchover complete. The async twin gets the same changes, unwraps AsyncBlueGreenPlugin instead of the sync class, and adds an execute probe on the blue cluster endpoint, through the connection that owns the status provider (the async status doesn't list instance endpoints). That probe takes the provider out of the shutdown hooks every async worker drains when it ends, and stops it itself, and its summary line shows the role and execute routing the plugin had for its host. The harness waits for an in-flight switchover to settle before it picks the cluster to delete, and also deletes a deployment whose switchover failed. It asks RDS for the cluster parameter group family of the engine version instead of assuming aurora-postgresql16, so the PostgreSQL task works on newer default versions, and it deletes that group when the cluster create fails. Covers #1352. --- .../container/test_blue_green_deployment.py | 409 +++++++++++-- .../test_blue_green_deployment_async.py | 288 +++++++-- .../container/utils/blue_green_checks.py | 548 ++++++++++++++++++ .../integration/host/TestEnvironment.java | 19 + .../integration/util/AuroraTestUtility.java | 29 +- 5 files changed, 1212 insertions(+), 81 deletions(-) create mode 100644 tests/integration/container/utils/blue_green_checks.py 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()