feat: GDB Failover

.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/aurora_topology_utils.ts

@@ -0,0 +1,68 @@
+/*
+  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 { TopologyQueryResult, TopologyUtils } from "./topology_utils";
+import { ClientWrapper } from "../client_wrapper";
+import { DatabaseDialect } from "../database_dialect/database_dialect";
+import { HostInfo } from "../host_info";
+import { isDialectTopologyAware } from "../utils/utils";
+import { Messages } from "../utils/messages";
+
+/**
+ * TopologyUtils implementation for Aurora clusters using a single HostInfo template.
+ */
+export class AuroraTopologyUtils extends TopologyUtils {
+  async queryForTopology(
+    targetClient: ClientWrapper,
+    dialect: DatabaseDialect,
+    initialHost: HostInfo,
+    clusterInstanceTemplate: HostInfo
+  ): Promise<HostInfo[]> {
+    if (!isDialectTopologyAware(dialect)) {
+      throw new TypeError(Messages.get("RdsHostListProvider.incorrectDialect"));
+    }
+
+    return await dialect
+      .queryForTopology(targetClient)
+      .then((res: TopologyQueryResult[]) => this.verifyWriter(this.createHosts(res, initialHost, clusterInstanceTemplate)));
+  }
+
+  public createHosts(topologyQueryResults: TopologyQueryResult[], initialHost: HostInfo, clusterInstanceTemplate: HostInfo): HostInfo[] {
+    const hostsMap = new Map<string, HostInfo>();
+    topologyQueryResults.forEach((row) => {
+      const lastUpdateTime = row.lastUpdateTime ?? Date.now();
+
+      const host = this.createHost(
+        row.id,
+        row.host,
+        row.isWriter,
+        row.weight,
+        lastUpdateTime,
+        initialHost,
+        clusterInstanceTemplate,
+        row.endpoint,
+        row.port
+      );
+
+      const existing = hostsMap.get(host.host);
+      if (!existing || existing.lastUpdateTime < host.lastUpdateTime) {
+        hostsMap.set(host.host, host);
+      }
+    });
+
+    return Array.from(hostsMap.values());
+  }
+}

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,67 @@
+/*
+  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 { 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";
+import { parseInstanceTemplates } from "../utils/utils";
+
+export class GlobalAuroraHostListProvider extends RdsHostListProvider {
+  protected instanceTemplatesByRegion: Map<string, HostInfo>;
+  protected override initSettings(): void {
+    super.initSettings();
+
+    const instanceTemplates = WrapperProperties.GLOBAL_CLUSTER_INSTANCE_HOST_PATTERNS.get(this.properties);
+    this.instanceTemplatesByRegion = 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.topologyUtils,
+          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.topologyUtils.queryForTopology(targetClient, dialect, this.initialHost, this.instanceTemplatesByRegion);
+  }
+}

common/lib/host_list_provider/global_topology_utils.ts

@@ -22,31 +22,36 @@ import { isDialectTopologyAware } from "../utils/utils";
 import { Messages } from "../utils/messages";
 import { AwsWrapperError } from "../utils/errors";
 
-export class GlobalTopologyUtils extends TopologyUtils {
-  async queryForTopology(
-    targetClient: ClientWrapper,
-    dialect: DatabaseDialect,
-    initialHost: HostInfo,
-    clusterInstanceTemplate: HostInfo
-  ): Promise<HostInfo[]> {
-    throw new AwsWrapperError("Not implemented");
-  }
+export interface GdbTopologyUtils {
+  getRegion(instanceId: string, targetClient: ClientWrapper, dialect: DatabaseDialect): Promise<string | null>;
+}
 
-  async queryForTopologyWithRegion(
+export class GlobalTopologyUtils extends TopologyUtils implements GdbTopologyUtils {
+  async queryForTopology(
     targetClient: ClientWrapper,
     dialect: DatabaseDialect,
     initialHost: HostInfo,
     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;
+  }
+
   private createHostsWithTemplateMap(
     topologyQueryResults: TopologyQueryResult[],
     initialHost: HostInfo,

common/lib/host_list_provider/monitoring/global_aurora_topology_monitor.ts

@@ -0,0 +1,69 @@
+/*
+  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 { GdbTopologyUtils, 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";
+import { TopologyUtils } from "../topology_utils";
+
+function isGdbTopologyUtils(utils: TopologyUtils): utils is TopologyUtils & GdbTopologyUtils {
+  return "getRegion" in utils && typeof (utils as unknown as GdbTopologyUtils).getRegion === "function";
+}
+
+export class GlobalAuroraTopologyMonitor extends ClusterTopologyMonitorImpl {
+  protected readonly instanceTemplatesByRegion: Map<string, HostInfo>;
+  declare public readonly topologyUtils: TopologyUtils;
+
+  constructor(
+    servicesContainer: FullServicesContainer,
+    topologyUtils: TopologyUtils,
+    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> {
+    if (!isGdbTopologyUtils(this.topologyUtils)) {
+      throw new AwsWrapperError(Messages.get("GlobalAuroraTopologyMonitor.invalidTopologyUtils"));
+    }
+
+    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;
+  }
+}

common/lib/host_list_provider/rds_host_list_provider.ts

@@ -257,14 +257,9 @@ export class RdsHostListProvider implements DynamicHostListProvider {
     return topology == null ? null : topology.hosts;
   }
 
-  static clearAll(): void {
-    // No-op
-    // TODO: remove if still not used after full service container refactoring
-  }
-
   clear(): void {
     if (this.clusterId) {
-      CoreServicesContainer.getInstance().getStorageService().remove(Topology, this.clusterId);
+      this.servicesContainers.getStorageService().remove(Topology, this.clusterId);
     }
   }