chore: topology utils refactoring (#615)

Author: karenc-bq

common/lib/database_dialect/topology_aware_database_dialect.ts

@@ -14,20 +14,21 @@
  * limitations under the License.
  */
 
-import { HostInfo } from "../host_info";
-import { HostListProvider } from "../host_list_provider/host_list_provider";
 import { HostRole } from "../host_role";
 import { ClientWrapper } from "../client_wrapper";
+import { TopologyQueryResult } from "../host_list_provider/topology_utils";
 
 export interface TopologyAwareDatabaseDialect {
-  queryForTopology(client: ClientWrapper, hostListProvider: HostListProvider): Promise<HostInfo[]>;
+  queryForTopology(client: ClientWrapper): Promise<TopologyQueryResult[]>;
 
   identifyConnection(targetClient: ClientWrapper): Promise<string>;
 
   getHostRole(client: ClientWrapper): Promise<HostRole>;
 
   // Returns the host id of the targetClient if it is connected to a writer, null otherwise.
   getWriterId(targetClient: ClientWrapper): Promise<string | null>;
+
+  getInstanceId(targetClient: ClientWrapper): Promise<[string, string]>;
 }
 
 export interface GlobalAuroraTopologyDialect extends TopologyAwareDatabaseDialect {

common/lib/host_list_provider/global_topology_utils.ts

@@ -0,0 +1,88 @@
+/*
+ * 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";
+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");
+  }
+
+  async queryForTopologyWithRegion(
+    targetClient: ClientWrapper,
+    dialect: DatabaseDialect,
+    initialHost: HostInfo,
+    instanceTemplateByRegion: Map<string, HostInfo>
+  ): Promise<HostInfo[]> {
+    if (!isDialectTopologyAware(dialect)) {
+      throw new TypeError(Messages.get("RdsHostListProvider.incorrectDialect"));
+    }
+
+    return await dialect
+      .queryForTopology(targetClient)
+      .then((res: TopologyQueryResult[]) => this.verifyWriter(this.createHostsWithTemplateMap(res, initialHost, instanceTemplateByRegion)));
+  }
+
+  private createHostsWithTemplateMap(
+    topologyQueryResults: TopologyQueryResult[],
+    initialHost: HostInfo,
+    instanceTemplateByRegion: Map<string, HostInfo>
+  ): HostInfo[] {
+    const hostsMap = new Map<string, HostInfo>();
+    topologyQueryResults.forEach((row) => {
+      if (!row.awsRegion) {
+        throw new AwsWrapperError(Messages.get("GlobalTopologyUtils.missingRegion", row.host));
+      }
+      const clusterInstanceTemplate = instanceTemplateByRegion.get(row.awsRegion);
+
+      if (!clusterInstanceTemplate) {
+        throw new AwsWrapperError(Messages.get("GlobalTopologyUtils.missingTemplateForRegion", row.awsRegion, row.host));
+      }
+
+      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/host_list_provider.ts

@@ -40,9 +40,7 @@ export interface HostListProvider {
 
   getHostRole(client: ClientWrapper, dialect: DatabaseDialect): Promise<HostRole>;
 
-  identifyConnection(targetClient: ClientWrapper, dialect: DatabaseDialect): Promise<HostInfo | null>;
-
-  createHost(host: string, isWriter: boolean, weight: number, lastUpdateTime: number, port?: number): HostInfo;
+  identifyConnection(targetClient: ClientWrapper): Promise<HostInfo | null>;
 
   getHostProviderType(): string;
 

common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts

@@ -15,7 +15,6 @@
 */
 
 import { HostInfo } from "../../host_info";
-import { CacheMap } from "../../utils/cache_map";
 import { PluginService } from "../../plugin_service";
 import { HostAvailability } from "../../host_availability/host_availability";
 import { logTopology, sleep } from "../../utils/utils";
@@ -27,7 +26,9 @@ import { MonitoringRdsHostListProvider } from "./monitoring_host_list_provider";
 import { Messages } from "../../utils/messages";
 import { CoreServicesContainer } from "../../utils/core_services_container";
 import { Topology } from "../topology";
-import { StorageService, StorageServiceImpl } from "../../utils/storage/storage_service";
+import { StorageService } from "../../utils/storage/storage_service";
+import { TopologyUtils } from "../topology_utils";
+import { RdsUtils } from "../../utils/rds_utils";
 
 export interface ClusterTopologyMonitor {
   forceRefresh(client: ClientWrapper, timeoutMs: number): Promise<HostInfo[]>;
@@ -49,6 +50,9 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
   private readonly refreshRateMs: number;
   private readonly highRefreshRateMs: number;
   private readonly storageService: StorageService;
+  private readonly topologyUtils: TopologyUtils;
+  private readonly rdsUtils: RdsUtils = new RdsUtils();
+  private readonly instanceTemplate: HostInfo;
 
   private writerHostInfo: HostInfo = null;
   private isVerifiedWriterConnection: boolean = false;
@@ -74,17 +78,21 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
   private requestToUpdateTopology: boolean = false;
 
   constructor(
+    topologyUtils: TopologyUtils,
     clusterId: string,
     initialHostInfo: HostInfo,
     props: Map<string, any>,
+    instanceTemplate: HostInfo,
     pluginService: PluginService,
     hostListProvider: MonitoringRdsHostListProvider,
     refreshRateMs: number,
     highRefreshRateMs: number
   ) {
+    this.topologyUtils = topologyUtils;
     this.clusterId = clusterId;
     this.storageService = CoreServicesContainer.getInstance().getStorageService(); // TODO: store serviceContainer instead
     this.initialHostInfo = initialHostInfo;
+    this.instanceTemplate = instanceTemplate;
     this._pluginService = pluginService;
     this._hostListProvider = hostListProvider;
     this.refreshRateMs = refreshRateMs;
@@ -212,12 +220,19 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
         this.monitoringClient = client;
         logger.debug(Messages.get("ClusterTopologyMonitor.openedMonitoringConnection", this.initialHostInfo.host));
         try {
-          const writerId = await this.getWriterHostIdIfConnected(this.monitoringClient, this.initialHostInfo.hostId);
-          if (writerId) {
+          if (await this.topologyUtils.isWriterInstance(this.monitoringClient)) {
             this.isVerifiedWriterConnection = true;
-            this.writerHostInfo = this.initialHostInfo;
-            logger.info(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.initialHostInfo.host));
-            writerVerifiedByThisTask = true;
+
+            if (this.rdsUtils.isRdsInstance(this.initialHostInfo.host)) {
+              this.writerHostInfo = this.initialHostInfo;
+              logger.info(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.writerHostInfo.host));
+              writerVerifiedByThisTask = true;
+            } else {
+              const pair: [string, string] = await this.topologyUtils.getInstanceId(this.monitoringClient);
+              const instanceTemplate: HostInfo = await this.getInstanceTemplate(pair[1], this.monitoringClient);
+              this.writerHostInfo = this.topologyUtils.createHost(pair[0], pair[1], true, 0, Date.now(), this.initialHostInfo, instanceTemplate);
+              logger.debug(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.writerHostInfo.host));
+            }
           }
         } catch (error) {
           // Do nothing.
@@ -245,6 +260,10 @@ export class ClusterTopologyMonitorImpl implements ClusterTopologyMonitor {
     return hosts;
   }
 
+  protected getInstanceTemplate(hostId: string, targetClient: ClientWrapper): Promise<HostInfo> {
+    return Promise.resolve(this.instanceTemplate);
+  }
+
   updateTopologyCache(hosts: HostInfo[]): void {
     this.storageService.set(this.clusterId, new Topology(hosts));
     this.requestToUpdateTopology = false;
@@ -513,7 +532,7 @@ export class HostMonitor {
     let hosts: HostInfo[];
     try {
       hosts = await this.monitor.hostListProvider.sqlQueryForTopology(client);
-      if (hosts === null || hosts.length === 0) {
+      if (hosts === null) {
         return;
       }
       this.monitor.hostMonitorsLatestTopology = hosts;

common/lib/host_list_provider/monitoring/monitoring_host_list_provider.ts

@@ -27,6 +27,7 @@ import { BlockingHostListProvider } from "../host_list_provider";
 import { logger } from "../../../logutils";
 import { SlidingExpirationCacheWithCleanupTask } from "../../utils/sliding_expiration_cache_with_cleanup_task";
 import { isDialectTopologyAware } from "../../utils/utils";
+import { TopologyUtils } from "../topology_utils";
 
 export class MonitoringRdsHostListProvider extends RdsHostListProvider implements BlockingHostListProvider {
   static readonly CACHE_CLEANUP_NANOS: bigint = BigInt(60_000_000_000); // 1 minute.
@@ -48,8 +49,14 @@ export class MonitoringRdsHostListProvider extends RdsHostListProvider implement
 
   private readonly pluginService: PluginService;
 
-  constructor(properties: Map<string, any>, originalUrl: string, hostListProviderService: HostListProviderService, pluginService: PluginService) {
-    super(properties, originalUrl, hostListProviderService);
+  constructor(
+    properties: Map<string, any>,
+    originalUrl: string,
+    topologyUtils: TopologyUtils,
+    hostListProviderService: HostListProviderService,
+    pluginService: PluginService
+  ) {
+    super(properties, originalUrl, topologyUtils, hostListProviderService);
     this.pluginService = pluginService;
   }
 
@@ -58,7 +65,7 @@ export class MonitoringRdsHostListProvider extends RdsHostListProvider implement
     await MonitoringRdsHostListProvider.monitors.clear();
   }
 
-  async queryForTopology(targetClient: ClientWrapper, dialect: DatabaseDialect): Promise<HostInfo[]> {
+  async getCurrentTopology(targetClient: ClientWrapper, dialect: DatabaseDialect): Promise<HostInfo[]> {
     const monitor: ClusterTopologyMonitor = this.initMonitor();
 
     try {
@@ -70,11 +77,7 @@ export class MonitoringRdsHostListProvider extends RdsHostListProvider implement
   }
 
   async sqlQueryForTopology(targetClient: ClientWrapper): Promise<HostInfo[]> {
-    const dialect: DatabaseDialect = this.hostListProviderService.getDialect();
-    if (!isDialectTopologyAware(dialect)) {
-      throw new TypeError(Messages.get("RdsHostListProvider.incorrectDialect"));
-    }
-    return await dialect.queryForTopology(targetClient, this).then((res: any) => this.processQueryResults(res));
+    return await this.topologyUtils.queryForTopology(targetClient, this.pluginService.getDialect(), this.initialHost, this.clusterInstanceTemplate);
   }
 
   async forceMonitoringRefresh(shouldVerifyWriter: boolean, timeoutMs: number): Promise<HostInfo[]> {
@@ -88,9 +91,11 @@ export class MonitoringRdsHostListProvider extends RdsHostListProvider implement
       this.clusterId,
       () =>
         new ClusterTopologyMonitorImpl(
+          this.topologyUtils,
           this.clusterId,
           this.initialHost,
           this.properties,
+          this.clusterInstanceTemplate,
           this.pluginService,
           this,
           WrapperProperties.CLUSTER_TOPOLOGY_REFRESH_RATE_MS.get(this.properties),