feat: gdb failover support

Author: karenc-bq

.github/workflows/integration_tests.yml

@@ -5,7 +5,6 @@ on:
   push:
     branches:
       - main
-      - refactor/monitor-service
     paths-ignore:
       - "**/*.md"
       - "**/*.jpg"

common/lib/connection_plugin_chain_builder.ts

@@ -43,6 +43,7 @@ import { CustomEndpointPluginFactory } from "./plugins/custom_endpoint/custom_en
 import { ConfigurationProfile } from "./profile/configuration_profile";
 import { HostMonitoring2PluginFactory } from "./plugins/efm2/host_monitoring2_plugin_factory";
 import { BlueGreenPluginFactory } from "./plugins/bluegreen/blue_green_plugin_factory";
+import { GlobalDbFailoverPluginFactory } from "./plugins/gdb_failover/global_db_failover_plugin_factory";
 import { FullServicesContainer } from "./utils/full_services_container";
 
 /*
@@ -66,6 +67,7 @@ export class ConnectionPluginChainBuilder {
     ["readWriteSplitting", { factory: ReadWriteSplittingPluginFactory, weight: 600 }],
     ["failover", { factory: FailoverPluginFactory, weight: 700 }],
     ["failover2", { factory: Failover2PluginFactory, weight: 710 }],
+    ["gdbFailover", { factory: GlobalDbFailoverPluginFactory, weight: 720 }],
     ["efm", { factory: HostMonitoringPluginFactory, weight: 800 }],
     ["efm2", { factory: HostMonitoring2PluginFactory, weight: 810 }],
     ["fastestResponseStrategy", { factory: FastestResponseStrategyPluginFactory, weight: 900 }],
@@ -87,6 +89,7 @@ export class ConnectionPluginChainBuilder {
     [ReadWriteSplittingPluginFactory, 600],
     [FailoverPluginFactory, 700],
     [Failover2PluginFactory, 710],
+    [GlobalDbFailoverPluginFactory, 720],
     [HostMonitoringPluginFactory, 800],
     [HostMonitoring2PluginFactory, 810],
     [LimitlessConnectionPluginFactory, 950],

common/lib/database_dialect/database_dialect_codes.ts

@@ -15,11 +15,13 @@
 */
 
 export class DatabaseDialectCodes {
+  static readonly GLOBAL_AURORA_MYSQL: string = "global-aurora-mysql";
   static readonly AURORA_MYSQL: string = "aurora-mysql";
   static readonly RDS_MYSQL: string = "rds-mysql";
   static readonly MYSQL: string = "mysql";
   // https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/multi-az-db-clusters-concepts.html
   static readonly RDS_MULTI_AZ_MYSQL: string = "rds-multi-az-mysql";
+  static readonly GLOBAL_AURORA_PG: string = "global-aurora-pg";
   static readonly AURORA_PG: string = "aurora-pg";
   static readonly RDS_PG: string = "rds-pg";
   // https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/multi-az-db-clusters-concepts.html

common/lib/database_dialect/database_dialect_manager.ts

@@ -95,6 +95,14 @@ export class DatabaseDialectManager implements DatabaseDialectProvider {
 
     if (this.dbType === DatabaseType.MYSQL) {
       const type = this.rdsHelper.identifyRdsType(host);
+      if (type == RdsUrlType.RDS_GLOBAL_WRITER_CLUSTER) {
+        this.canUpdate = false;
+        this.dialectCode = DatabaseDialectCodes.GLOBAL_AURORA_MYSQL;
+        this.dialect = <DatabaseDialect>this.knownDialectsByCode.get(DatabaseDialectCodes.GLOBAL_AURORA_MYSQL);
+        this.logCurrentDialect();
+        return this.dialect;
+      }
+
       if (type.isRdsCluster) {
         this.canUpdate = true;
         this.dialectCode = DatabaseDialectCodes.AURORA_MYSQL;
@@ -128,6 +136,14 @@ export class DatabaseDialectManager implements DatabaseDialectProvider {
         return this.dialect;
       }
 
+      if (type == RdsUrlType.RDS_GLOBAL_WRITER_CLUSTER) {
+        this.canUpdate = false;
+        this.dialectCode = DatabaseDialectCodes.GLOBAL_AURORA_PG;
+        this.dialect = <DatabaseDialect>this.knownDialectsByCode.get(DatabaseDialectCodes.GLOBAL_AURORA_PG);
+        this.logCurrentDialect();
+        return this.dialect;
+      }
+
       if (type.isRdsCluster) {
         this.canUpdate = true;
         this.dialectCode = DatabaseDialectCodes.AURORA_PG;

common/lib/host_list_provider/connection_string_host_list_provider.ts

@@ -112,4 +112,8 @@ export class ConnectionStringHostListProvider implements StaticHostListProvider
   getClusterId(): string {
     throw new AwsWrapperError("ConnectionStringHostListProvider does not support getClusterId.");
   }
+
+  forceMonitoringRefresh(shouldVerifyWriter: boolean, timeoutMs: number): Promise<HostInfo[]> {
+    throw new AwsWrapperError("ConnectionStringHostListProvider does not support forceMonitoringRefresh.");
+  }
 }

common/lib/host_list_provider/global_aurora_host_list_provider.ts

@@ -0,0 +1,74 @@
+/*
+  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.
+*/
+
+import { RdsHostListProvider } from "./rds_host_list_provider";
+import { GlobalTopologyUtils } from "./global_topology_utils";
+import { FullServicesContainer } from "../utils/full_services_container";
+import { HostInfo } from "../host_info";
+import { WrapperProperties } from "../wrapper_property";
+import { ClusterTopologyMonitor, ClusterTopologyMonitorImpl } from "./monitoring/cluster_topology_monitor";
+import { GlobalAuroraTopologyMonitor } from "./monitoring/global_aurora_topology_monitor";
+import { MonitorInitializer } from "../utils/monitoring/monitor";
+import { ClientWrapper } from "../client_wrapper";
+import { DatabaseDialect } from "../database_dialect/database_dialect";
+
+export class GlobalAuroraHostListProvider extends RdsHostListProvider {
+  protected readonly globalTopologyUtils: GlobalTopologyUtils;
+  protected instanceTemplatesByRegion: Map<string, HostInfo>;
+
+  constructor(properties: Map<string, any>, originalUrl: string, topologyUtils: GlobalTopologyUtils, servicesContainers: FullServicesContainer) {
+    super(properties, originalUrl, topologyUtils, servicesContainers);
+    this.globalTopologyUtils = topologyUtils;
+  }
+
+  protected override initSettings(): void {
+    super.initSettings();
+
+    const instanceTemplates = WrapperProperties.GLOBAL_CLUSTER_INSTANCE_HOST_PATTERNS.get(this.properties);
+    this.instanceTemplatesByRegion = this.globalTopologyUtils.parseInstanceTemplates(
+      instanceTemplates,
+      (hostPattern: string) => this.validateHostPatternSetting(hostPattern),
+      () => this.hostListProviderService.getHostInfoBuilder()
+    );
+  }
+
+  protected override async getOrCreateMonitor(): Promise<ClusterTopologyMonitor> {
+    const initializer: MonitorInitializer = {
+      createMonitor: (servicesContainer: FullServicesContainer): ClusterTopologyMonitor => {
+        return new GlobalAuroraTopologyMonitor(
+          servicesContainer,
+          this.globalTopologyUtils,
+          this.clusterId,
+          this.initialHost,
+          this.properties,
+          this.clusterInstanceTemplate,
+          this.refreshRateNano,
+          this.highRefreshRateNano,
+          this.instanceTemplatesByRegion
+        );
+      }
+    };
+
+    return await this.servicesContainers
+      .getMonitorService()
+      .runIfAbsent(ClusterTopologyMonitorImpl, this.clusterId, this.servicesContainers, this.properties, initializer);
+  }
+
+  override async getCurrentTopology(targetClient: ClientWrapper, dialect: DatabaseDialect): Promise<HostInfo[]> {
+    this.init();
+    return await this.globalTopologyUtils.queryForTopologyWithRegion(targetClient, dialect, this.initialHost, this.instanceTemplatesByRegion);
+  }
+}

common/lib/host_list_provider/global_topology_utils.ts

@@ -21,15 +21,18 @@ import { HostInfo } from "../host_info";
 import { isDialectTopologyAware } from "../utils/utils";
 import { Messages } from "../utils/messages";
 import { AwsWrapperError } from "../utils/errors";
+import { HostInfoBuilder } from "../host_info_builder";
+import { logger } from "../../logutils";
 
 export class GlobalTopologyUtils extends TopologyUtils {
-  async queryForTopology(
+  override async queryForTopology(
     targetClient: ClientWrapper,
     dialect: DatabaseDialect,
     initialHost: HostInfo,
     clusterInstanceTemplate: HostInfo
   ): Promise<HostInfo[]> {
-    throw new AwsWrapperError("Not implemented");
+    // This method should not be called on this class.
+    throw new AwsWrapperError("GlobalTopologyUtils.queryForTopology should not be called directly. Use queryForTopologyWithRegion instead.");
   }
 
   async queryForTopologyWithRegion(
@@ -39,14 +42,61 @@ export class GlobalTopologyUtils extends TopologyUtils {
     instanceTemplateByRegion: Map<string, HostInfo>
   ): Promise<HostInfo[]> {
     if (!isDialectTopologyAware(dialect)) {
-      throw new TypeError(Messages.get("RdsHostListProvider.incorrectDialect"));
+      throw new AwsWrapperError(Messages.get("RdsHostListProvider.incorrectDialect"));
     }
 
     return await dialect
       .queryForTopology(targetClient)
       .then((res: TopologyQueryResult[]) => this.verifyWriter(this.createHostsWithTemplateMap(res, initialHost, instanceTemplateByRegion)));
   }
 
+  async getRegion(instanceId: string, targetClient: ClientWrapper, dialect: DatabaseDialect): Promise<string | null> {
+    if (!isDialectTopologyAware(dialect)) {
+      throw new AwsWrapperError(Messages.get("RdsHostListProvider.incorrectDialect"));
+    }
+
+    const results = await dialect.queryForTopology(targetClient);
+    const match = results.find((row) => row.id === instanceId);
+    return match?.awsRegion ?? null;
+  }
+
+  parseInstanceTemplates(
+    instanceTemplatesString: string | null,
+    hostValidator: (hostPattern: string) => void,
+    hostInfoBuilderFunc: () => HostInfoBuilder
+  ): Map<string, HostInfo> {
+    if (!instanceTemplatesString) {
+      throw new AwsWrapperError(Messages.get("GlobalTopologyUtils.globalClusterInstanceHostPatternsRequired"));
+    }
+
+    const instanceTemplates = new Map<string, HostInfo>();
+    const patterns = instanceTemplatesString.split(",");
+
+    for (const pattern of patterns) {
+      const trimmedPattern = pattern.trim();
+      const colonIndex = trimmedPattern.indexOf(":");
+      if (colonIndex === -1) {
+        throw new AwsWrapperError(Messages.get("GlobalTopologyUtils.invalidPatternFormat", trimmedPattern));
+      }
+
+      const region = trimmedPattern.substring(0, colonIndex).trim();
+      const hostPattern = trimmedPattern.substring(colonIndex + 1).trim();
+
+      if (!region || !hostPattern) {
+        throw new AwsWrapperError(Messages.get("GlobalTopologyUtils.invalidPatternFormat", trimmedPattern));
+      }
+
+      hostValidator(hostPattern);
+
+      const hostInfo = hostInfoBuilderFunc().withHost(hostPattern).build();
+      instanceTemplates.set(region, hostInfo);
+    }
+
+    logger.debug(`Detected Global Database patterns: ${JSON.stringify(Array.from(instanceTemplates.entries()))}`);
+
+    return instanceTemplates;
+  }
+
   private createHostsWithTemplateMap(
     topologyQueryResults: TopologyQueryResult[],
     initialHost: HostInfo,

common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts

@@ -160,8 +160,8 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust
     this.stopMonitoring = true;
     this.hostMonitorsStop = true;
     this.requestToUpdateTopology = true;
-    // await Promise.all(this.untrackedPromises);
-    // await Promise.all(this.submittedHosts.values());
+    await Promise.all(this.untrackedPromises);
+    await Promise.all(this.submittedHosts.values());
 
     const monitoringClientToClose = this.monitoringClient;
     const hostMonitorsWriterClientToClose = this.hostMonitorsWriterClient;
@@ -258,7 +258,7 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust
       let client: ClientWrapper;
       try {
         client = await this.servicesContainer.getPluginService().forceConnect(this.initialHostInfo, this._monitoringProperties);
-      } catch {
+      } catch (connectError) {
         // Unable to connect to host;
         return null;
       }
@@ -374,7 +374,6 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust
 
     this.untrackedPromises = [];
     this.submittedHosts.clear();
-    this.hostMonitors.clear();
 
     return super.stop();
   }
@@ -417,8 +416,8 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust
                 );
                 await minimalServiceContainer.getPluginManager().init();
                 const hostMonitor = new HostMonitor(minimalServiceContainer, this, hostInfo, this.writerHostInfo);
-                this.submittedHosts.set(hostInfo.host, hostMonitor.run());
-                this.hostMonitors.set(hostInfo.host, hostMonitor);
+                const promise = hostMonitor.run();
+                this.submittedHosts.set(hostInfo.host, promise);
               }
             });
 
@@ -451,7 +450,8 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust
                 hosts.forEach((hostInfo) => {
                   if (!this.submittedHosts.get(hostInfo.host)) {
                     const hostMonitor = new HostMonitor(this.servicesContainer, this, hostInfo, this.writerHostInfo);
-                    this.submittedHosts.set(hostInfo.host, hostMonitor.run());
+                    const promise = hostMonitor.run();
+                    this.submittedHosts.set(hostInfo.host, promise);
                   }
                 });
               }
@@ -697,7 +697,8 @@ export class HostMonitor {
             } else {
               // It might be some transient error. Let's try again.
               // If the error repeats, we will try again after a longer delay.
-              await sleep(this.calculateBackoffWithJitter(this.connectionAttempts++));
+              const backoff = this.calculateBackoffWithJitter(this.connectionAttempts++);
+              await sleep(backoff);
               this.monitor.completedOneCycle.set(this.hostInfo.hostId, true);
               this.monitor.readerTopologiesById.delete(this.hostInfo.hostId);
               continue;
@@ -718,7 +719,8 @@ export class HostMonitor {
           if (isWriter) {
             try {
               // First connection after failover may be stale.
-              if ((await this.monitor.pluginService.getHostRole(this.client)) !== HostRole.WRITER) {
+              const hostRole = await this.monitor.pluginService.getHostRole(this.client);
+              if (hostRole !== HostRole.WRITER) {
                 isWriter = false;
               }
             } catch (error: any) {

common/lib/host_list_provider/monitoring/global_aurora_topology_monitor.ts

@@ -0,0 +1,65 @@
+/*
+  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.
+*/
+
+import { ClusterTopologyMonitorImpl } from "./cluster_topology_monitor";
+import { GlobalTopologyUtils } from "../global_topology_utils";
+import { FullServicesContainer } from "../../utils/full_services_container";
+import { HostInfo } from "../../host_info";
+import { ClientWrapper } from "../../client_wrapper";
+import { AwsWrapperError } from "../../utils/errors";
+import { Messages } from "../../utils/messages";
+
+export class GlobalAuroraTopologyMonitor extends ClusterTopologyMonitorImpl {
+  protected readonly instanceTemplatesByRegion: Map<string, HostInfo>;
+  declare public readonly topologyUtils: GlobalTopologyUtils;
+
+  constructor(
+    servicesContainer: FullServicesContainer,
+    topologyUtils: GlobalTopologyUtils,
+    clusterId: string,
+    initialHostInfo: HostInfo,
+    properties: Map<string, any>,
+    instanceTemplate: HostInfo,
+    refreshRateNano: number,
+    highRefreshRateNano: number,
+    instanceTemplatesByRegion: Map<string, HostInfo>
+  ) {
+    super(servicesContainer, topologyUtils, clusterId, initialHostInfo, properties, instanceTemplate, refreshRateNano, highRefreshRateNano);
+
+    this.instanceTemplatesByRegion = instanceTemplatesByRegion;
+    this.topologyUtils = topologyUtils;
+  }
+
+  protected override async getInstanceTemplate(hostId: string, targetClient: ClientWrapper): Promise<HostInfo> {
+    const dialect = this.hostListProviderService.getDialect();
+    const region = await this.topologyUtils.getRegion(hostId, targetClient, dialect);
+
+    if (region) {
+      const instanceTemplate = this.instanceTemplatesByRegion.get(region);
+      if (!instanceTemplate) {
+        throw new AwsWrapperError(Messages.get("GlobalAuroraTopologyMonitor.cannotFindRegionTemplate", region));
+      }
+      return instanceTemplate;
+    }
+
+    return this.instanceTemplate;
+  }
+
+  override async queryForTopology(client: ClientWrapper): Promise<HostInfo[]> {
+    const dialect = this.hostListProviderService.getDialect();
+    return await this.topologyUtils.queryForTopologyWithRegion(client, dialect, this.initialHostInfo, this.instanceTemplatesByRegion);
+  }
+}

common/lib/host_selector.ts

@@ -18,5 +18,5 @@ import { HostInfo } from "./host_info";
 import { HostRole } from "./host_role";
 
 export interface HostSelector {
-  getHost(hosts: HostInfo[], role: HostRole, props?: Map<string, any>): HostInfo;
+  getHost(hosts: HostInfo[], role: HostRole | null, props?: Map<string, any>): HostInfo;
 }