Multi-AZ integration tests (#328)

Author: congoamz

.github/workflows/autoscaling_tests.yml

@@ -53,7 +53,7 @@ jobs:
         run: |
           ./gradlew --no-parallel --no-daemon test-autoscaling --info
         env:
-          AURORA_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
+          RDS_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
           AURORA_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
           AWS_ACCESS_KEY_ID: ${{ env.TEMP_AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ env.TEMP_AWS_SECRET_ACCESS_KEY }}

.github/workflows/integration_tests.yml

@@ -56,8 +56,9 @@ jobs:
         run: |
           ./gradlew --no-parallel --no-daemon test-all-environments --info
         env:
-          AURORA_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
+          RDS_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
           AURORA_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
+          MULTI_AZ_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
           AWS_ACCESS_KEY_ID: ${{ env.TEMP_AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ env.TEMP_AWS_SECRET_ACCESS_KEY }}
           AWS_SESSION_TOKEN: ${{ env.TEMP_AWS_SESSION_TOKEN }}

.github/workflows/mysql_performance_tests.yml

@@ -53,7 +53,7 @@ jobs:
         run: |
           ./gradlew --no-parallel --no-daemon test-mysql-aurora-performance --info
         env:
-          AURORA_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
+          RDS_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
           AURORA_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
           AWS_ACCESS_KEY_ID: ${{ env.TEMP_AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ env.TEMP_AWS_SECRET_ACCESS_KEY }}

.github/workflows/pg_performance_tests.yml

@@ -53,7 +53,7 @@ jobs:
         run: |
           ./gradlew --no-parallel --no-daemon test-pg-aurora-performance --info
         env:
-          AURORA_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
+          RDS_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
           AURORA_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
           AWS_ACCESS_KEY_ID: ${{ env.TEMP_AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ env.TEMP_AWS_SECRET_ACCESS_KEY }}

aws_advanced_python_wrapper/database_dialect.py

@@ -32,7 +32,7 @@
 from aws_advanced_python_wrapper.errors import (AwsWrapperError,
                                                 QueryTimeoutError)
 from aws_advanced_python_wrapper.host_list_provider import (
-    ConnectionStringHostListProvider, MultiAzRdsHostListProvider,
+    ConnectionStringHostListProvider, MultiAzHostListProvider,
     RdsHostListProvider)
 from aws_advanced_python_wrapper.hostinfo import HostInfo
 from aws_advanced_python_wrapper.utils.decorators import \
@@ -253,7 +253,7 @@ def prepare_conn_props(self, props: Properties):
 
 
 class RdsMysqlDialect(MysqlDatabaseDialect):
-    _DIALECT_UPDATE_CANDIDATES = (DialectCode.AURORA_MYSQL,)
+    _DIALECT_UPDATE_CANDIDATES = (DialectCode.AURORA_MYSQL, DialectCode.MULTI_AZ_MYSQL)
 
     def is_dialect(self, conn: Connection) -> bool:
         try:
@@ -278,7 +278,7 @@ class RdsPgDialect(PgDatabaseDialect):
                          "(setting LIKE '%aurora_stat_utils%') AS aurora_stat_utils "
                          "FROM pg_settings "
                          "WHERE name='rds.extensions'")
-    _DIALECT_UPDATE_CANDIDATES = (DialectCode.AURORA_PG,)
+    _DIALECT_UPDATE_CANDIDATES = (DialectCode.AURORA_PG, DialectCode.MULTI_AZ_PG)
 
     def is_dialect(self, conn: Connection) -> bool:
         if not super().is_dialect(conn):
@@ -337,6 +337,8 @@ def get_host_list_provider_supplier(self) -> Callable:
 
 
 class AuroraPgDialect(PgDatabaseDialect, TopologyAwareDatabaseDialect):
+    _DIALECT_UPDATE_CANDIDATES: Tuple[DialectCode, ...] = (DialectCode.MULTI_AZ_PG,)
+
     _EXTENSIONS_QUERY = "SELECT (setting LIKE '%aurora_stat_utils%') AS aurora_stat_utils " \
                         "FROM pg_settings WHERE name='rds.extensions'"
 
@@ -352,6 +354,10 @@ class AuroraPgDialect(PgDatabaseDialect, TopologyAwareDatabaseDialect):
     _HOST_ID_QUERY = "SELECT aurora_db_instance_identifier()"
     _IS_READER_QUERY = "SELECT pg_is_in_recovery()"
 
+    @property
+    def dialect_update_candidates(self) -> Optional[Tuple[DialectCode, ...]]:
+        return AuroraPgDialect._DIALECT_UPDATE_CANDIDATES
+
     def is_dialect(self, conn: Connection) -> bool:
         if not super().is_dialect(conn):
             return False
@@ -385,7 +391,7 @@ def get_host_list_provider_supplier(self) -> Callable:
         return lambda provider_service, props: RdsHostListProvider(provider_service, props)
 
 
-class MultiAzMysqlDialect(MysqlDatabaseDialect):
+class MultiAzMysqlDialect(MysqlDatabaseDialect, TopologyAwareDatabaseDialect):
     _TOPOLOGY_QUERY = "SELECT id, endpoint, port FROM mysql.rds_topology"
     _WRITER_HOST_QUERY = "SHOW REPLICA STATUS"
     _WRITER_HOST_COLUMN_INDEX = 39
@@ -400,15 +406,16 @@ def is_dialect(self, conn: Connection) -> bool:
         try:
             with closing(conn.cursor()) as cursor:
                 cursor.execute(MultiAzMysqlDialect._TOPOLOGY_QUERY)
-                if cursor.fetchone() is not None:
+                records = cursor.fetchall()
+                if records is not None and len(records) > 0:
                     return True
         except Exception:
             pass
 
         return False
 
     def get_host_list_provider_supplier(self) -> Callable:
-        return lambda provider_service, props: MultiAzRdsHostListProvider(
+        return lambda provider_service, props: MultiAzHostListProvider(
             provider_service,
             props,
             self._TOPOLOGY_QUERY,
@@ -430,7 +437,7 @@ def prepare_conn_props(self, props: Properties):
             props["conn_attrs"].update(extra_conn_attrs)
 
 
-class MultiAzPgDialect(PgDatabaseDialect):
+class MultiAzPgDialect(PgDatabaseDialect, TopologyAwareDatabaseDialect):
     # The driver name passed to show_topology is used for RDS metrics purposes.
     # It is not required for functional correctness.
     _TOPOLOGY_QUERY = \
@@ -464,7 +471,7 @@ def is_dialect(self, conn: Connection) -> bool:
         return False
 
     def get_host_list_provider_supplier(self) -> Callable:
-        return lambda provider_service, props: MultiAzRdsHostListProvider(
+        return lambda provider_service, props: MultiAzHostListProvider(
             provider_service,
             props,
             self._TOPOLOGY_QUERY,

aws_advanced_python_wrapper/host_list_provider.py

@@ -185,11 +185,16 @@ def _initialize(self):
             self._initial_hosts: Tuple[HostInfo, ...] = (self._initial_host_info,)
             self._host_list_provider_service.initial_connection_host_info = self._initial_host_info
 
-            self._cluster_instance_template: HostInfo
             host_pattern = WrapperProperties.CLUSTER_INSTANCE_HOST_PATTERN.get(self._props)
             if host_pattern:
+                if host_pattern.find(":") > -1:
+                    host_pattern, port = host_pattern.split(":")
+                else:
+                    port = HostInfo.NO_PORT
+
                 self._cluster_instance_template = HostInfo(
-                    host=WrapperProperties.CLUSTER_INSTANCE_HOST_PATTERN.get(self._props),
+                    host=host_pattern,
+                    port=port,
                     host_availability_strategy=host_availability_strategy)
             else:
                 self._cluster_instance_template = HostInfo(
@@ -503,7 +508,7 @@ class FetchTopologyResult:
         is_cached_data: bool
 
 
-class MultiAzRdsHostListProvider(RdsHostListProvider):
+class MultiAzHostListProvider(RdsHostListProvider):
     def __init__(
             self,
             provider_service: HostListProviderService,
@@ -560,13 +565,23 @@ def _process_multi_az_query_results(self, cursor: Cursor, writer_id: str) -> Tup
         return tuple(hosts)
 
     def _create_multi_az_host(self, record: Tuple, writer_id: str) -> HostInfo:
-        host_id = record[0]
+        id = record[0]  # The ID will look something like '0123456789' (MySQL) or 'db-ABC1DE2FGHI' (Postgres)
         host = record[1]
         port = record[2]
-        role = HostRole.WRITER if host_id == writer_id else HostRole.READER
+        role = HostRole.WRITER if id == writer_id else HostRole.READER
+
+        host_pattern = WrapperProperties.CLUSTER_INSTANCE_HOST_PATTERN.get(self._props)
+        if host_pattern:
+            instance_name = self._rds_utils.get_instance_id(host)  # e.g. 'postgres-instance-1'
+            if instance_name is None:
+                raise AwsWrapperError(Messages.get("MultiAzHostListProvider.UnableToParseInstanceName"))
+
+            host = host_pattern.replace("?", instance_name)
+            if host.find(":") > -1:
+                host, port = host.split(":")
 
         host_info = HostInfo(
-            host=host, port=port, role=role, availability=HostAvailability.AVAILABLE, weight=0, host_id=host_id)
+            host=host, port=port, role=role, availability=HostAvailability.AVAILABLE, weight=0, host_id=id)
         host_info.add_alias(host)
         return host_info
 

aws_advanced_python_wrapper/plugin_service.py

@@ -269,7 +269,7 @@ def __init__(
         self._target_func = target_func
         self._driver_dialect_manager = driver_dialect_manager
         self._driver_dialect = driver_dialect
-        self._dialect = self._dialect_provider.get_dialect(driver_dialect.dialect_code, props)
+        self._database_dialect = self._dialect_provider.get_dialect(driver_dialect.dialect_code, props)
 
     @property
     def hosts(self) -> Tuple[HostInfo, ...]:
@@ -331,7 +331,7 @@ def driver_dialect(self) -> DriverDialect:
 
     @property
     def database_dialect(self) -> DatabaseDialect:
-        return self._dialect
+        return self._database_dialect
 
     @property
     def network_bound_methods(self) -> Set[str]:
@@ -362,16 +362,16 @@ def update_dialect(self, connection: Optional[Connection] = None):
         if connection is None:
             raise AwsWrapperError(Messages.get("PluginServiceImpl.UpdateDialectConnectionNone"))
 
-        original_dialect = self._dialect
-        self._dialect = \
+        original_dialect = self._database_dialect
+        self._database_dialect = \
             self._dialect_provider.query_for_dialect(
                 self._original_url,
                 self._initial_connection_host_info,
                 connection,
                 self.driver_dialect)
 
-        if original_dialect != self._dialect:
-            host_list_provider_init = self._dialect.get_host_list_provider_supplier()
+        if original_dialect != self._database_dialect:
+            host_list_provider_init = self._database_dialect.get_host_list_provider_supplier()
             self.host_list_provider = host_list_provider_init(self, self._props)
 
     def update_driver_dialect(self, connection_provider: ConnectionProvider):

aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties

@@ -19,13 +19,6 @@ AuroraHostListPlugin.ProviderAlreadySet=[AuroraHostListPlugin]Another dynamic ho
 AuroraPgDialect.HasExtensionsTrue=[AuroraPgDialect] has_extensions: True
 AuroraPgDialect.HasTopologyTrue=[AuroraPgDialect] has_topology: True
 
-AuroraTestUtility.ClusterMemberNotFound=[AuroraTestUtility] Cannot find cluster member whose db instance identifier is '{}'.
-AuroraTestUtility.CreateDBInstanceFailed=[AuroraTestUtility] Could not create database instance `{}`.
-AuroraTestUtility.FailoverRequestNotSuccessful=[AuroraTestUtility] Failover cluster request was not successful.
-AuroraTestUtility.InstanceDescriptionTimeout=[AuroraTestUtility] Instance description timeout for {}. The instance did not reach status '{}' within {} minutes.
-AuroraTestUtility.InvalidDatabaseEngine=[AuroraTestUtility] The detected database engine is not valid: {}
-AuroraTestUtility.WriterInstanceNotFound=[AuroraTestUtility] Cannot find writer instance for cluster '{}'.
-
 AwsSdk.UnsupportedRegion=[AwsSdk] Unsupported AWS region {}. For supported regions please read https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Concepts.RegionsAndAvailabilityZones.html
 
 AwsSecretsManagerPlugin.ConnectException=[AwsSecretsManagerPlugin] Error occurred while opening a connection: {}
@@ -136,6 +129,8 @@ MonitoringThreadContainer.SupplierMonitorNone=[MonitorThreadContainer] The monit
 MonitorService.EmptyAliasSet=[MonitorService] Empty alias set passed for '{}'. The alias set should not be empty.
 MonitorService.ErrorPopulatingAliases=[MonitorService] An error occurred while populating aliases: '{}'.
 
+MultiAzHostListProvider.UnableToParseInstanceName=[MultiAzHostListProvider] The MultiAzHostListProvider was unable to parse the instance name from the endpoint returned by the topology query.
+
 OpenedConnectionTracker.OpenedConnectionsTracked=[OpenedConnectionTracker] Opened Connections Tracked: {}
 OpenedConnectionTracker.InvalidatingConnections=[OpenedConnectionTracker] Invalidating opened connections to host: {}
 OpenedConnectionTracker.UnableToPopulateOpenedConnectionSet=[OpenedConnectionTracker] The driver is unable to track this opened connection because the instance endpoint is unknown.
@@ -181,6 +176,14 @@ RdsHostListProvider.UninitializedInitialHostInfo=[RdsHostListProvider] The drive
 
 RdsPgDialect.RdsToolsAuroraUtils=[RdsPgDialect] rds_tools: {}, aurora_utils: {}
 
+RdsTestUtility.ClusterMemberNotFound=[RdsTestUtility] Cannot find cluster member whose db instance identifier is '{}'.
+RdsTestUtility.CreateDBInstanceFailed=[RdsTestUtility] Could not create database instance `{}`.
+RdsTestUtility.FailoverRequestNotSuccessful=[RdsTestUtility] Failover cluster request was not successful.
+RdsTestUtility.InstanceDescriptionTimeout=[RdsTestUtility] Instance description timeout for {}. The instance did not reach status '{}' within {} minutes.
+RdsTestUtility.InvalidDatabaseEngine=[RdsTestUtility] The detected database engine is not valid: {}
+RdsTestUtility.MethodNotSupportedForDeployment=[RdsTestUtility] Method '{}' is not supported for the current database engine deployment: '{}'
+RdsTestUtility.WriterInstanceNotFound=[RdsTestUtility] Cannot find writer instance for cluster '{}'.
+
 ReaderFailoverHandler.AttemptingReaderConnection=[ReaderFailoverHandler] Trying to connect to reader: '{}', with properties '{}'
 ReaderFailoverHandler.FailedReaderConnection=[ReaderFailoverHandler] Failed to connect to reader: '{}'
 ReaderFailoverHandler.InvalidTopology=[ReaderFailoverHandler] '{}' was called with an invalid (None or empty) topology.

aws_advanced_python_wrapper/utils/rdsutils.py

@@ -59,43 +59,43 @@ class RdsUtils:
     """
 
     AURORA_DNS_PATTERN = r"(?P<instance>.+)\." \
-                         r"(?P<dns>proxy-|cluster-|cluster-ro-|cluster-custom-)" \
-                         r"?(?P<domain>[a-zA-Z0-9]+\." \
-                         r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)"
+                         r"(?P<dns>proxy-|cluster-|cluster-ro-|cluster-custom-)?" \
+                         r"(?P<domain>[a-zA-Z0-9]+\." \
+                         r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)(?!\.cn$)"
     AURORA_INSTANCE_PATTERN = r"(?P<instance>.+)\." \
                               r"(?P<domain>[a-zA-Z0-9]+\." \
-                              r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)"
+                              r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)(?!\.cn$)"
     AURORA_CLUSTER_PATTERN = r"(?P<instance>.+)\." \
                              r"(?P<dns>cluster-|cluster-ro-)+" \
                              r"(?P<domain>[a-zA-Z0-9]+\." \
-                             r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)"
+                             r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)(?!\.cn$)"
     AURORA_CUSTOM_CLUSTER_PATTERN = r"(?P<instance>.+)\." \
                                     r"(?P<dns>cluster-custom-)+" \
                                     r"(?P<domain>[a-zA-Z0-9]+\." \
-                                    r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)"
+                                    r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com)(?!\.cn$)"
     AURORA_PROXY_DNS_PATTERN = r"(?P<instance>.+)\." \
                                r"(?P<dns>proxy-)+" \
                                r"(?P<domain>[a-zA-Z0-9]+\." \
-                               r"(?P<region>[a-zA-Z0-9\\-]+)\.rds\.amazonaws\.com)"
+                               r"(?P<region>[a-zA-Z0-9\\-]+)\.rds\.amazonaws\.com)(?!\.cn$)"
     AURORA_CHINA_DNS_PATTERN = r"(?P<instance>.+)\." \
                                r"(?P<dns>proxy-|cluster-|cluster-ro-|cluster-custom-)?" \
-                               r"(?P<domain>[a-zA-Z0-9]+\.rds\." \
-                               r"(?P<region>[a-zA-Z0-9\-]+)\.amazonaws\.com\.cn)"
+                               r"(?P<domain>[a-zA-Z0-9]+\." \
+                               r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com\.cn)"
     AURORA_CHINA_INSTANCE_PATTERN = r"(?P<instance>.+)\." \
                                     r"(?P<domain>[a-zA-Z0-9]+\." \
                                     r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com\.cn)"
     AURORA_CHINA_CLUSTER_PATTERN = r"(?P<instance>.+)\." \
                                    r"(?P<dns>cluster-|cluster-ro-)+" \
-                                   r"(?P<domain>[a-zA-Z0-9]+\.rds\." \
-                                   r"(?P<region>[a-zA-Z0-9\-]+)\.amazonaws\.com\.cn)"
+                                   r"(?P<domain>[a-zA-Z0-9]+\." \
+                                   r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com\.cn)"
     AURORA_CHINA_CUSTOM_CLUSTER_PATTERN = r"(?P<instance>.+)\." \
                                           r"(?P<dns>cluster-custom-)+" \
-                                          r"(?P<domain>[a-zA-Z0-9]+\.rds\." \
-                                          r"(?P<region>[a-zA-Z0-9\-]+)\.amazonaws\.com\.cn)"
+                                          r"(?P<domain>[a-zA-Z0-9]+\." \
+                                          r"(?P<region>[a-zA-Z0-9\-]+)\.rds\.amazonaws\.com\.cn)"
     AURORA_CHINA_PROXY_DNS_PATTERN = r"(?P<instance>.+)\." \
                                      r"(?P<dns>proxy-)+" \
-                                     r"(?P<domain>[a-zA-Z0-9]+\.rds\." \
-                                     r"(?P<region>[a-zA-Z0-9\-])+\.amazonaws\.com\.cn)"
+                                     r"(?P<domain>[a-zA-Z0-9]+\." \
+                                     r"(?P<region>[a-zA-Z0-9\-])+\.rds\.amazonaws\.com\.cn)"
 
     IP_V4 = r"^(([1-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){1}" \
             r"(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){2}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$"
@@ -104,6 +104,7 @@ class RdsUtils:
 
     DNS_GROUP = "dns"
     DOMAIN_GROUP = "domain"
+    INSTANCE_GROUP = "instance"
     REGION_GROUP = "region"
 
     def is_rds_cluster_dns(self, host: str) -> bool:
@@ -170,6 +171,16 @@ def get_rds_cluster_host_url(self, host: str):
 
         return None
 
+    def get_instance_id(self, host: str) -> Optional[str]:
+        if not host or not host.strip():
+            return None
+
+        match = self._find(host, [self.AURORA_INSTANCE_PATTERN, self.AURORA_CHINA_INSTANCE_PATTERN])
+        if match:
+            return match.group(self.INSTANCE_GROUP)
+
+        return None
+
     def is_ipv4(self, host: str) -> bool:
         return self._contains(host, [self.IP_V4])