diff --git a/common/lib/database_dialect/database_dialect.ts b/common/lib/database_dialect/database_dialect.ts index 2077934a..3a2c8337 100644 --- a/common/lib/database_dialect/database_dialect.ts +++ b/common/lib/database_dialect/database_dialect.ts @@ -15,13 +15,13 @@ */ import { HostListProvider } from "../host_list_provider/host_list_provider"; -import { HostListProviderService } from "../host_list_provider_service"; import { ClientWrapper } from "../client_wrapper"; import { FailoverRestriction } from "../plugins/failover/failover_restriction"; import { ErrorHandler } from "../error_handler"; import { TransactionIsolationLevel } from "../utils/transaction_isolation_level"; import { HostRole } from "../host_role"; import { FullServicesContainer } from "../utils/full_services_container"; +import { HostInfo } from "../host_info"; export enum DatabaseType { MYSQL, @@ -41,6 +41,14 @@ export interface DatabaseDialect { getDialectUpdateCandidates(): string[]; getErrorHandler(): ErrorHandler; getHostRole(targetClient: ClientWrapper): Promise; + /** + * Filters the given hosts down to those reachable from the configured accessible regions. + * + * Optional: dialects that are not region-aware (i.e. non-Global Aurora dialects) may omit this + * method, in which case callers should treat all hosts as available. Implementing this as an + * optional member avoids breaking out-of-tree custom dialects when the method is introduced. + */ + filterAvailableHosts?(hosts: HostInfo[], accessibleRegions: string[]): Promise; isDialect(targetClient: ClientWrapper): Promise; getHostListProvider(props: Map, originalUrl: string, servicesContainer: FullServicesContainer): HostListProvider; isClientValid(targetClient: ClientWrapper): Promise; diff --git a/common/lib/database_dialect/database_dialect_manager.ts b/common/lib/database_dialect/database_dialect_manager.ts index b462e60b..d75a2119 100644 --- a/common/lib/database_dialect/database_dialect_manager.ts +++ b/common/lib/database_dialect/database_dialect_manager.ts @@ -205,6 +205,10 @@ export class DatabaseDialectManager implements DatabaseDialectProvider { return this.dialect; } + isConfirmedDialect(): boolean { + return !this.canUpdate; + } + logCurrentDialect() { logger.debug(`Current dialect: ${this.dialectCode}, ${this.dialect.getDialectName()}, canUpdate: ${this.canUpdate}`); } diff --git a/common/lib/database_dialect/database_dialect_provider.ts b/common/lib/database_dialect/database_dialect_provider.ts index bc6c2931..78170525 100644 --- a/common/lib/database_dialect/database_dialect_provider.ts +++ b/common/lib/database_dialect/database_dialect_provider.ts @@ -20,4 +20,5 @@ import { ClientWrapper } from "../client_wrapper"; export interface DatabaseDialectProvider { getDialect(props: Map): DatabaseDialect; getDialectForUpdate(targetClient: ClientWrapper, originalHost: string, newHost: string): Promise; + isConfirmedDialect(): boolean; } diff --git a/common/lib/host_list_provider/monitoring/aurora_monitoring_connection_handler.ts b/common/lib/host_list_provider/monitoring/aurora_monitoring_connection_handler.ts new file mode 100644 index 00000000..3273568e --- /dev/null +++ b/common/lib/host_list_provider/monitoring/aurora_monitoring_connection_handler.ts @@ -0,0 +1,91 @@ +/* + 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 { HostInfo } from "../../host_info"; +import { HostRole } from "../../host_role"; +import { PluginService } from "../../plugin_service"; +import { WrapperProperties } from "../../wrapper_property"; +import { ClientWrapper } from "../../client_wrapper"; +import { logger } from "../../../logutils"; +import { Messages } from "../../utils/messages"; +import { + AbstractMonitoringConnectionHandler, + MonitoringConnectionPriority, + parseMonitoringConnectionPriorities +} from "./monitoring_connection_handler"; + +export class AuroraMonitoringConnectionHandler extends AbstractMonitoringConnectionHandler { + private readonly writerPriorityIndex: number; + private readonly readerPriorityIndex: number; + + constructor( + pluginService: PluginService, + monitoringProperties: Map, + getMonitoringClient: () => ClientWrapper | null, + setMonitoringClient: (client: ClientWrapper | null) => void + ) { + const priorities = parseMonitoringConnectionPriorities(WrapperProperties.MONITORING_CONNECTION_PRIORITY.get(monitoringProperties)); + super(pluginService, monitoringProperties, priorities, getMonitoringClient, setMonitoringClient); + this.writerPriorityIndex = this.computeIndex(true); + this.readerPriorityIndex = this.computeIndex(false); + logger.debug(Messages.get("AuroraMonitoringConnectionHandler.initialized", this.priorities.join(","))); + } + + private computeIndex(isWriter: boolean): number { + for (let i = 0; i < this.priorities.length; i++) { + if (this.isSatisfiedBy(this.priorities[i], isWriter)) { + return i; + } + } + return this.priorities.length; + } + + private isSatisfiedBy(priority: MonitoringConnectionPriority, isWriter: boolean): boolean { + switch (priority) { + case MonitoringConnectionPriority.STRICT_WRITER: + return isWriter; + case MonitoringConnectionPriority.STRICT_READER: + return !isWriter; + case MonitoringConnectionPriority.WRITER_OR_READER: + return true; + default: + return false; + } + } + + protected getPriorityIndex(_hostInfo: HostInfo, isWriter: boolean): number { + return isWriter ? this.writerPriorityIndex : this.readerPriorityIndex; + } + + protected findHostsForPriority(priorityIndex: number, candidates: HostInfo[]): HostInfo[] { + const priority = this.priorities[priorityIndex]; + if (!priority) { + return []; + } + switch (priority) { + case MonitoringConnectionPriority.STRICT_WRITER: + return candidates.filter((h) => h.role === HostRole.WRITER); + case MonitoringConnectionPriority.STRICT_READER: + return candidates.filter((h) => h.role === HostRole.READER); + case MonitoringConnectionPriority.WRITER_OR_READER: { + const writers = candidates.filter((h) => h.role === HostRole.WRITER); + return writers.length > 0 ? writers : candidates.filter((h) => h.role === HostRole.READER); + } + default: + return []; + } + } +} diff --git a/common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts b/common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts index 46241678..c3ed0d05 100644 --- a/common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts +++ b/common/lib/host_list_provider/monitoring/cluster_topology_monitor.ts @@ -34,6 +34,8 @@ import { Event, EventSubscriber } from "../../utils/events/event"; import { MonitorResetEvent } from "../../utils/events/monitor_reset_event"; import { ServiceUtils } from "../../utils/service_utils"; import { WrapperProperties } from "../../wrapper_property"; +import { MonitoringConnectionHandler } from "./monitoring_connection_handler"; +import { AuroraMonitoringConnectionHandler } from "./aurora_monitoring_connection_handler"; export interface ClusterTopologyMonitor extends Monitor, EventSubscriber { forceRefresh(client: ClientWrapper, timeoutMs: number): Promise; @@ -73,10 +75,11 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust private readonly storageService: StorageService; private readonly rdsUtils: RdsUtils = new RdsUtils(); protected readonly instanceTemplate: HostInfo; + protected connectionHandler: MonitoringConnectionHandler | null = null; - private writerHostInfo: HostInfo | null = null; - private isVerifiedWriterConnection: boolean = false; - private monitoringClient: ClientWrapper | null = null; + protected writerHostInfo: HostInfo | null = null; + protected lastKnownWriterHostInfo: HostInfo | null = null; + protected monitoringClient: ClientWrapper | null = null; private highRefreshRateEndTimeNs: bigint = BigInt(0); public readonly topologyUtils: TopologyUtils; @@ -89,12 +92,26 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust }; // Tracking of the host monitors. - private hostMonitors: Map = new Map(); public hostMonitorsWriterClient = null; public hostMonitorsWriterInfo: HostInfo = null; public hostMonitorsReaderClient = null; public hostMonitorsLatestTopology: HostInfo[] = []; + // Connections harvested from host monitors as they stop, keyed by host. When the writer resides in + // an inaccessible region (someRegionsInaccessible), no host monitor can obtain a verified writer + // connection, so the main loop adopts one of these (reader) connections as the monitoring connection + // to exit panic mode. Populated by HostMonitor.run()'s finally block via harvestConnection(). + public hostMonitorsHarvestedConnections: Map = new Map(); + // True when the most recently submitted set of host monitors excluded one or more hosts because they + // fell outside the accessible regions. Gates reader-consensus panic exit so the standard writer + // detection path is left untouched when all regions are accessible. + public hostMonitorsSomeRegionsInaccessible: boolean = false; + // Set by a HostMonitor when, with some regions inaccessible, a reader observes that the writer has + // changed. No host monitor can connect to the new writer to verify it, so a reader-observed change is + // the only fast signal to exit panic mode. Prompts the main loop to adopt a harvested reader connection + // without waiting for the full stable-topology window. + public hostMonitorsReaderConsensusRequested: boolean = false; + // Controls for stopping asynchronous monitoring tasks. public hostMonitorsStop: boolean = false; @@ -142,6 +159,24 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust driverDialect.setQueryTimeout(this._monitoringProperties, undefined, queryTimeout); } + protected getConnectionHandler(): MonitoringConnectionHandler { + if (this.connectionHandler === null) { + this.connectionHandler = this.createConnectionHandler(); + } + return this.connectionHandler; + } + + protected createConnectionHandler(): MonitoringConnectionHandler { + return new AuroraMonitoringConnectionHandler( + this._pluginService, + this._monitoringProperties, + () => this.monitoringClient, + (client) => { + this.monitoringClient = client; + } + ); + } + get pluginService(): PluginService { return this._pluginService; } @@ -175,17 +210,15 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust await this.closeConnection(hostMonitorsReaderClientToClose); } + await this.cleanUpHarvestedConnections(); this.submittedHosts.clear(); - this.hostMonitors.clear(); } async forceMonitoringRefresh(shouldVerifyWriter: boolean, timeoutMs: number): Promise { if (shouldVerifyWriter) { - this.isVerifiedWriterConnection = false; - if (this.monitoringClient) { - const client = this.monitoringClient; - this.monitoringClient = null; - // Abort needed for MySQLClientWrapper in case client already closed. + const client = this.monitoringClient; + this.monitoringClient = null; + if (client) { await this.closeConnection(client); } } @@ -194,12 +227,12 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust } async forceRefresh(client: ClientWrapper, timeoutMs: number): Promise { - if (this.isVerifiedWriterConnection) { - // Get the monitoring task to refresh the topology using a verified connection. + if (this.monitoringClient) { + // Get the monitoring task to refresh the topology using the monitoring connection. return await this.waitTillTopologyGetsUpdated(timeoutMs); } - // Otherwise, use the provided unverified connection to update the topology. + // Otherwise, use the provided connection to update the topology. return await this.fetchTopologyAndUpdateCache(client); } @@ -244,39 +277,45 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust return null; } - private async openAnyClientAndUpdateTopology(): Promise { + protected async openAnyClientAndUpdateTopology(): Promise { if (!this.monitoringClient) { let client: ClientWrapper; try { client = await this.servicesContainer.pluginService.forceConnect(this.initialHostInfo, this._monitoringProperties); } catch (connectError) { - // Unable to connect to host; + // Unable to connect to host. return null; } - if (client && this.monitoringClient === null) { - this.monitoringClient = client; - logger.debug(Messages.get("ClusterTopologyMonitor.openedMonitoringConnection", this.initialHostInfo.host)); - try { - if (await this.topologyUtils.isWriterInstance(this.monitoringClient)) { - this.isVerifiedWriterConnection = true; + logger.debug(Messages.get("ClusterTopologyMonitor.openedMonitoringConnection", this.initialHostInfo.host)); - if (this.rdsUtils.isRdsInstance(this.initialHostInfo.host)) { - this.writerHostInfo = this.initialHostInfo; - logger.info(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.writerHostInfo.host)); - } 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)); - } + let isWriter = false; + try { + isWriter = await this.topologyUtils.isWriterInstance(client); + } catch (error) { + // Do nothing — assume not a writer. + } + + if (isWriter) { + try { + if (this.rdsUtils.isRdsInstance(this.initialHostInfo.host)) { + this.writerHostInfo = this.initialHostInfo; + this.lastKnownWriterHostInfo = this.initialHostInfo; + logger.info(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.writerHostInfo.host)); + } else { + const pair: [string, string] = await this.topologyUtils.getInstanceId(client); + const instanceTemplate: HostInfo = await this.getInstanceTemplate(pair[1], client); + this.writerHostInfo = this.topologyUtils.createHost(pair[0], pair[1], true, 0, Date.now(), this.initialHostInfo, instanceTemplate); + this.lastKnownWriterHostInfo = this.writerHostInfo; + logger.debug(Messages.get("ClusterTopologyMonitor.writerMonitoringConnection", this.writerHostInfo.host)); } } catch (error) { // Do nothing. - logger.error(Messages.get("ClusterTopologyMonitor.invalidWriterQuery", error?.message)); } - } else if (client) { - // Monitoring connection already set by another task, close the new connection. + } + + // Offer the connection to the handler. If rejected, close it. + if (!this.getConnectionHandler().acceptConnection(client, isWriter, this.initialHostInfo)) { await this.closeConnection(client); } } @@ -284,7 +323,6 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust const hosts: HostInfo[] = await this.fetchTopologyAndUpdateCache(this.monitoringClient); if (hosts === null) { - this.isVerifiedWriterConnection = false; await this.updateMonitoringClient(null); } return hosts; @@ -321,6 +359,31 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust await client?.abort(); } + isMonitoringClient(client: ClientWrapper): boolean { + return client === this.monitoringClient; + } + + get isStopped(): boolean { + return this._stop; + } + + /** + * Adopts ownership of a live connection handed off by a stopping HostMonitor, storing it in the harvest map + * so the main loop can promote one as the monitoring connection during reader-consensus panic exit. If an + * entry already exists for the host, the previous connection is closed to avoid a leak. + * + * @param hostInfo the host the connection belongs to + * @param client the live connection being handed off + */ + harvestConnection(hostInfo: HostInfo, client: ClientWrapper): void { + const previous = this.hostMonitorsHarvestedConnections.get(hostInfo); + this.hostMonitorsHarvestedConnections.set(hostInfo, client); + if (previous && previous !== client) { + // Should not normally happen, but clean up any previous entry to avoid leaks. + void this.closeConnection(previous); + } + } + async updateMonitoringClient(newClient: ClientWrapper | null): Promise { const clientToClose = this.monitoringClient; this.monitoringClient = newClient; @@ -355,6 +418,7 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust await this.closeConnection(monitoringClientToClose); } + await this.cleanUpHarvestedConnections(); this.submittedHosts.clear(); return super.stop(); @@ -386,19 +450,23 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust await this.closeHostMonitors(); - if (!(hosts !== null && !this.isVerifiedWriterConnection)) { + if (hosts === null || this.monitoringClient !== null) { await this.delay(true); continue; } - for (const hostInfo of hosts) { + const monitoredHosts = this.filterHostsForHostMonitoring(hosts); + const someRegionsInaccessible: boolean = monitoredHosts.length < hosts.length; + this.hostMonitorsSomeRegionsInaccessible = someRegionsInaccessible; + const baselineWriter: HostInfo = this.lastKnownWriterHostInfo; + for (const hostInfo of monitoredHosts) { if (!this.submittedHosts.get(hostInfo.host)) { const minimalServiceContainer = ServiceUtils.instance.createMinimalServiceContainerFrom( this.servicesContainer, this._monitoringProperties ); await minimalServiceContainer.pluginManager.init(); - const hostMonitor = new HostMonitor(minimalServiceContainer, this, hostInfo, this.writerHostInfo); + const hostMonitor = new HostMonitor(minimalServiceContainer, this, hostInfo, baselineWriter, someRegionsInaccessible); const promise = hostMonitor.run(); this.submittedHosts.set(hostInfo.host, promise); } @@ -414,13 +482,27 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust logger.debug(Messages.get("ClusterTopologyMonitor.writerPickedUpFromHostMonitors", writerClientHostInfo.toString())); const oldMonitoringClient = this.monitoringClient; - this.monitoringClient = writerClient; + + this.hostMonitorsWriterClient = null; + this.hostMonitorsWriterInfo = null; + this.monitoringClient = null; + if (!this.getConnectionHandler().acceptConnection(writerClient, true, writerClientHostInfo)) { + // Should not happen — the handler always accepts when there is no current monitoring client. + // Fall back to the writer connection so we still exit panic mode. + this.monitoringClient = writerClient; + } this.writerHostInfo = writerClientHostInfo; - this.isVerifiedWriterConnection = true; + this.lastKnownWriterHostInfo = writerClientHostInfo; this.highRefreshRateEndTimeNs = getTimeInNanos() + BigInt(this.highRefreshRateNs); this.hostMonitorsStop = true; await this.closeHostMonitors(); + + // A verified writer connection was promoted, so any connections harvested from host monitors + // during this panic cycle are no longer needed. Close them (skipping the current monitoring + // client) to avoid leaking sockets. + await this.cleanUpHarvestedConnections(); + this.submittedHosts.clear(); this.stableTopologiesStartNs = BigInt(0); this.readerTopologiesById.clear(); @@ -431,21 +513,33 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust await this.closeConnection(oldMonitoringClient); } + await this.delay(true); + continue; + } else if ( + this.hostMonitorsReaderConsensusRequested && + (await this.adoptHarvestedMonitoringConnection(this.hostMonitorsLatestTopology ?? this.getStoredHosts() ?? [])) + ) { + // A reader observed a writer change while the writer is in an inaccessible region. We adopted a + // harvested reader connection as the monitoring connection to exit panic mode. await this.delay(true); continue; } else { // Update host monitors with the new instances in the topology. const hosts: HostInfo[] | null = this.hostMonitorsLatestTopology; if (hosts && !this.hostMonitorsStop) { - for (const hostInfo of hosts) { + const monitoredHosts = this.filterHostsForHostMonitoring(hosts); + const someRegionsInaccessible: boolean = monitoredHosts.length < hosts.length; + this.hostMonitorsSomeRegionsInaccessible = someRegionsInaccessible; + const baselineWriter: HostInfo = this.lastKnownWriterHostInfo; + + for (const hostInfo of monitoredHosts) { if (!this.submittedHosts.get(hostInfo.host)) { const minimalServiceContainer = ServiceUtils.instance.createMinimalServiceContainerFrom( this.servicesContainer, this._monitoringProperties ); await minimalServiceContainer.pluginManager.init(); - // Intentionally not calling await on hostMonitor.run(). - const hostMonitor = new HostMonitor(minimalServiceContainer, this, hostInfo, this.writerHostInfo); + const hostMonitor = new HostMonitor(minimalServiceContainer, this, hostInfo, baselineWriter, someRegionsInaccessible); const promise = hostMonitor.run(); this.submittedHosts.set(hostInfo.host, promise); } @@ -454,7 +548,7 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust } } - this.checkForStableReaderTopologies(); + await this.checkForStableReaderTopologies(); await this.delay(true); } else { // We are in regular mode. @@ -469,15 +563,25 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust const hosts: HostInfo[] = await this.fetchTopologyAndUpdateCache(this.monitoringClient); if (hosts === null) { // Attempt to fetch topology failed, so we switch to panic mode. + // Clear writerHostInfo but keep lastKnownWriterHostInfo so host monitors + // can use it as a baseline for writer-change detection. const clientToClose = this.monitoringClient; this.monitoringClient = null; await this.closeConnection(clientToClose); - this.isVerifiedWriterConnection = false; this.writerHostInfo = null; await this.delay(false); continue; } + // Refresh lastKnownWriterHostInfo from topology so that if the monitoring + // connection later breaks, panic-mode host monitors have an accurate baseline. + const topologyWriter = hosts.find((h) => h.role === HostRole.WRITER); + if (topologyWriter) { + this.lastKnownWriterHostInfo = topologyWriter; + } + + await this.getConnectionHandler().attemptConnectionUpgrade(this.filterHostsForHostMonitoring(hosts)); + if (this.highRefreshRateEndTimeNs > 0 && getTimeInNanos() > this.highRefreshRateEndTimeNs) { this.highRefreshRateEndTimeNs = BigInt(0); } @@ -503,14 +607,14 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust return Promise.resolve(); } - protected checkForStableReaderTopologies(): void { + protected async checkForStableReaderTopologies(): Promise { const latestHosts: HostInfo[] = this.getStoredHosts(); if (!latestHosts || latestHosts.length === 0) { this.stableTopologiesStartNs = BigInt(0); return; } - const readerIds: string[] = latestHosts.map((host) => host.hostId); + const readerIds: string[] = this.filterHostsForHostMonitoring(latestHosts).map((host) => host.hostId); for (const id of readerIds) { const completedCycle = this.completedOneCycle.get(id) ?? false; if (!completedCycle) { @@ -561,16 +665,89 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust ) ); this.updateTopologyCache(readerTopology); + + // Reader topology is stable. Even though no writer was detected by the host monitors (e.g. the writer may + // live in a region we don't monitor), the readers we did probe have established connections we can use as + // the monitoring connection. Adopt one so we can exit panic mode. This is only attempted when some regions + // are inaccessible; otherwise we let the standard writer-detection path run, which also verifies a working + // writer connection and is more reliable. + await this.adoptHarvestedMonitoringConnection(readerTopology); } } + /** + * Attempts to exit panic mode by adopting one of the connections harvested from the host monitors as the + * monitoring connection. Used when the writer resides in an inaccessible region, so no host monitor can obtain + * a verified writer connection. The connection handler picks the best harvested connection according to its + * priority; unselected connections are closed. No-op unless we are in panic mode with some regions inaccessible + * and at least one harvested connection is available. + * + * @param readerTopology the reader-observed topology used to inform the handler's selection + * @returns true if a harvested connection was adopted as the monitoring connection + */ + protected async adoptHarvestedMonitoringConnection(readerTopology: HostInfo[]): Promise { + if (this.monitoringClient !== null || !this.hostMonitorsSomeRegionsInaccessible) { + return false; + } + + this.hostMonitorsStop = true; + await this.closeHostMonitors(); + + if (this.hostMonitorsHarvestedConnections.size === 0) { + return false; + } + + const selected: HostInfo | null = this.getConnectionHandler().acceptConnections( + this.hostMonitorsHarvestedConnections, + this.writerHostInfo, + readerTopology + ); + + if (selected) { + this.lastKnownWriterHostInfo = readerTopology.find((h) => h.role === HostRole.WRITER) ?? this.lastKnownWriterHostInfo; + this.highRefreshRateEndTimeNs = getTimeInNanos() + BigInt(this.highRefreshRateNs); + logger.debug(Messages.get("ClusterTopologyMonitor.exitPanicModeViaReaderConsensus", selected.host)); + } + + // Close any harvested connections that were not adopted as the monitoring connection. + await this.cleanUpHarvestedConnections(); + + this.submittedHosts.clear(); + this.stableTopologiesStartNs = BigInt(0); + this.readerTopologiesById.clear(); + this.completedOneCycle.clear(); + this.hostMonitorsReaderConsensusRequested = false; + + return selected !== null; + } + + /** + * Closes every harvested host-monitor connection except the one currently in use as the monitoring + * connection, then clears the harvest map. + */ + protected async cleanUpHarvestedConnections(): Promise { + for (const [, client] of this.hostMonitorsHarvestedConnections) { + if (client && client !== this.monitoringClient) { + try { + await this.closeConnection(client); + } catch (e: any) { + // Ignore. + } + } + } + this.hostMonitorsHarvestedConnections.clear(); + } + protected async reset(): Promise { logger.debug(Messages.get("ClusterTopologyMonitor.reset", this.clusterId, this.initialHostInfo.host)); this.hostMonitorsStop = true; await this.closeHostMonitors(); await this.hostMonitorClientCleanUp(); + await this.cleanUpHarvestedConnections(); this.hostMonitorsStop = false; + this.hostMonitorsSomeRegionsInaccessible = false; + this.hostMonitorsReaderConsensusRequested = false; this.submittedHosts.clear(); this.stableTopologiesStartNs = BigInt(0); this.readerTopologiesById.clear(); @@ -580,8 +757,8 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust this.hostMonitorsLatestTopology = []; await this.updateMonitoringClient(null); - this.isVerifiedWriterConnection = false; this.writerHostInfo = null; + this.lastKnownWriterHostInfo = null; this.highRefreshRateEndTimeNs = BigInt(0); this.requestToUpdateTopology = false; this.clearTopologyCache(); @@ -630,8 +807,12 @@ export class ClusterTopologyMonitorImpl extends AbstractMonitor implements Clust await this.hostMonitorClientCleanUp(); } + protected filterHostsForHostMonitoring(hosts: HostInfo[]): HostInfo[] { + return hosts; + } + private isInPanicMode(): boolean { - return !this.monitoringClient || !this.isVerifiedWriterConnection; + return !this.monitoringClient; } private getStoredHosts(): HostInfo[] | null { @@ -659,15 +840,23 @@ export class HostMonitor { protected readonly monitor: ClusterTopologyMonitorImpl; protected readonly hostInfo: HostInfo; protected readonly writerHostInfo: HostInfo | null; + protected readonly someRegionsInaccessible: boolean; protected writerChanged: boolean = false; protected connectionAttempts: number = 0; protected client: ClientWrapper | null = null; - constructor(servicesContainer: FullServicesContainer, monitor: ClusterTopologyMonitorImpl, hostInfo: HostInfo, writerHostInfo: HostInfo | null) { + constructor( + servicesContainer: FullServicesContainer, + monitor: ClusterTopologyMonitorImpl, + hostInfo: HostInfo, + writerHostInfo: HostInfo | null, + someRegionsInaccessible: boolean + ) { this.servicesContainer = servicesContainer; this.monitor = monitor; this.hostInfo = hostInfo; this.writerHostInfo = writerHostInfo; + this.someRegionsInaccessible = someRegionsInaccessible; } async run() { @@ -780,7 +969,18 @@ export class HostMonitor { this.monitor.completedOneCycle.set(this.hostInfo.hostId, true); this.monitor.readerTopologiesById.delete(this.hostInfo.hostId); - await this.monitor.closeConnection(this.client); + if (this.client && !this.monitor.isMonitoringClient(this.client)) { + // When some regions are inaccessible, the writer may be unreachable and no host monitor can promote a + // verified writer connection. Hand off this live (reader) connection to the monitor so the main loop can + // adopt it as the monitoring connection to exit panic mode. Otherwise close it as usual. + if (this.someRegionsInaccessible && !this.monitor.isStopped && !this.monitor.hostMonitorsWriterClient) { + this.monitor.harvestConnection(this.hostInfo, this.client); + } else { + await this.monitor.closeConnection(this.client); + } + // Ownership transferred (or connection closed); don't touch it again. + this.client = null; + } logger.debug(Messages.get("HostMonitor.endMonitoring", this.hostInfo.hostId, (Date.now() - startTime).toString())); } } @@ -818,6 +1018,15 @@ export class HostMonitor { this.monitor.updateHostsAvailability(hosts); this.monitor.updateTopologyCache(hosts); logger.debug(logTopology(hosts, `[hostMonitor ${this.hostInfo.hostId}] `)); + + // With some regions inaccessible, no host monitor may be able to connect to the new writer to verify it, + // so a reader-observed writer change is the only fast way to exit panic mode. Signal the main loop to adopt + // a harvested reader connection as the monitoring connection. + if (this.someRegionsInaccessible) { + logger.debug(Messages.get("HostMonitor.writerChangeExitTriggered", latestWriterHostInfo.host)); + this.monitor.hostMonitorsReaderConsensusRequested = true; + this.monitor.hostMonitorsStop = true; + } } } diff --git a/common/lib/host_list_provider/monitoring/gdb_monitoring_connection_handler.ts b/common/lib/host_list_provider/monitoring/gdb_monitoring_connection_handler.ts new file mode 100644 index 00000000..a2c1b2d6 --- /dev/null +++ b/common/lib/host_list_provider/monitoring/gdb_monitoring_connection_handler.ts @@ -0,0 +1,241 @@ +/* + 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 { HostInfo } from "../../host_info"; +import { HostRole } from "../../host_role"; +import { PluginService } from "../../plugin_service"; +import { WrapperProperties } from "../../wrapper_property"; +import { RdsUtils } from "../../utils/rds_utils"; +import { ClientWrapper } from "../../client_wrapper"; +import { logger } from "../../../logutils"; +import { Messages } from "../../utils/messages"; +import { AbstractMonitoringConnectionHandler } from "./monitoring_connection_handler"; +import { equalsIgnoreCase } from "../../utils/utils"; + +export enum GdbMonitoringConnectionPriority { + STRICT_WRITER_PRIMARY = "strict-writer-primary", + STRICT_WRITER_SECONDARY = "strict-writer-secondary", + STRICT_READER_PRIMARY = "strict-reader-primary", + STRICT_READER_SECONDARY = "strict-reader-secondary", + WRITER_OR_READER_PRIMARY = "writer-or-reader-primary", + WRITER_OR_READER_SECONDARY = "writer-or-reader-secondary", + REGION = "region" +} + +interface GdbPriorityConfig { + type: GdbMonitoringConnectionPriority; + region?: string; +} + +// AWS region identifiers look like "us-east-1", "eu-west-2", "ap-southeast-1". +const REGION_SHAPE = /^[a-z]{2}-[a-z]+-\d+$/; + +function parseGdbPriority(value: string | null): GdbPriorityConfig | null { + if (!value) { + return null; + } + const lower = value.toLowerCase().trim(); + switch (lower) { + case "strict-writer-primary": + return { type: GdbMonitoringConnectionPriority.STRICT_WRITER_PRIMARY }; + case "strict-writer-secondary": + return { type: GdbMonitoringConnectionPriority.STRICT_WRITER_SECONDARY }; + case "strict-reader-primary": + return { type: GdbMonitoringConnectionPriority.STRICT_READER_PRIMARY }; + case "strict-reader-secondary": + return { type: GdbMonitoringConnectionPriority.STRICT_READER_SECONDARY }; + case "writer-or-reader-primary": + return { type: GdbMonitoringConnectionPriority.WRITER_OR_READER_PRIMARY }; + case "writer-or-reader-secondary": + return { type: GdbMonitoringConnectionPriority.WRITER_OR_READER_SECONDARY }; + default: + // Any unrecognized token is treated as a region literal. If it doesn't look like an AWS + // region identifier, it is most likely a typo (e.g. "strict-wrtier-primary") that will + // never match any host, so warn to aid diagnosis. + if (!REGION_SHAPE.test(lower)) { + logger.warn(Messages.get("GdbMonitoringConnectionHandler.unrecognizedPriority", value)); + } + return { type: GdbMonitoringConnectionPriority.REGION, region: lower }; + } +} + +export class GdbMonitoringConnectionHandler extends AbstractMonitoringConnectionHandler { + private readonly rdsUtils: RdsUtils = new RdsUtils(); + private readonly accessibleRegions: string[] | null; + private primaryRegion: string | null = null; + private currentHostInfo: HostInfo | null = null; + + constructor( + pluginService: PluginService, + monitoringProperties: Map, + accessibleRegions: string[] | null, + homeRegion: string | null, + getMonitoringClient: () => ClientWrapper | null, + setMonitoringClient: (client: ClientWrapper | null) => void + ) { + const priorities = GdbMonitoringConnectionHandler.parsePriorities(WrapperProperties.GDB_MONITORING_CONNECTION_PRIORITY.get(monitoringProperties)); + super(pluginService, monitoringProperties, priorities, getMonitoringClient, setMonitoringClient); + this.accessibleRegions = accessibleRegions; + logger.debug(Messages.get("GdbMonitoringConnectionHandler.initialized", JSON.stringify(this.priorities))); + } + + private static parsePriorities(value: string | null): GdbPriorityConfig[] { + if (!value) { + return [{ type: GdbMonitoringConnectionPriority.STRICT_WRITER_PRIMARY }]; + } + const results: GdbPriorityConfig[] = []; + for (const part of value.split(",")) { + const p = parseGdbPriority(part.trim()); + if (p) { + results.push(p); + } + } + return results.length > 0 ? results : [{ type: GdbMonitoringConnectionPriority.STRICT_WRITER_PRIMARY }]; + } + + override acceptConnection(client: ClientWrapper, isWriter: boolean, hostInfo: HostInfo): boolean { + if (isWriter) { + this.primaryRegion = this.getHostRegion(hostInfo); + } + const accepted = super.acceptConnection(client, isWriter, hostInfo); + if (accepted) { + this.currentHostInfo = hostInfo; + } + return accepted; + } + + override acceptConnections(connections: Map, writerHostInfo: HostInfo | null, topology: HostInfo[]): HostInfo | null { + if (writerHostInfo) { + this.primaryRegion = this.getHostRegion(writerHostInfo); + } + const selected = super.acceptConnections(connections, writerHostInfo, topology); + if (selected) { + this.currentHostInfo = selected; + } + return selected; + } + + override async attemptConnectionUpgrade(currentTopology: HostInfo[]): Promise { + this.updatePrimaryRegion(currentTopology); + if (this.currentPriorityIndex > 0 && this.currentHostInfo) { + const newIndex = this.effectiveIndex(this.getPriorityIndex(this.currentHostInfo, this.currentHostInfo.role === HostRole.WRITER)); + if (newIndex < this.currentPriorityIndex) { + this.currentPriorityIndex = newIndex; + } + } + return super.attemptConnectionUpgrade(currentTopology); + } + + protected getPriorityIndex(hostInfo: HostInfo, isWriter: boolean): number { + for (let i = 0; i < this.priorities.length; i++) { + if (this.isSatisfiedBy(this.priorities[i], hostInfo, isWriter)) { + return i; + } + } + return -1; + } + + private isSatisfiedBy(priority: GdbPriorityConfig, hostInfo: HostInfo, isWriter: boolean): boolean { + switch (priority.type) { + case GdbMonitoringConnectionPriority.STRICT_WRITER_PRIMARY: + return isWriter && this.isInPrimaryRegion(hostInfo); + case GdbMonitoringConnectionPriority.STRICT_WRITER_SECONDARY: + return isWriter && !this.isInPrimaryRegion(hostInfo); + case GdbMonitoringConnectionPriority.STRICT_READER_PRIMARY: + return !isWriter && this.isInPrimaryRegion(hostInfo); + case GdbMonitoringConnectionPriority.STRICT_READER_SECONDARY: + return !isWriter && !this.isInPrimaryRegion(hostInfo); + case GdbMonitoringConnectionPriority.WRITER_OR_READER_PRIMARY: + return this.isInPrimaryRegion(hostInfo); + case GdbMonitoringConnectionPriority.WRITER_OR_READER_SECONDARY: + return !this.isInPrimaryRegion(hostInfo); + case GdbMonitoringConnectionPriority.REGION: + return equalsIgnoreCase(this.getHostRegion(hostInfo), priority.region); + default: + return false; + } + } + + protected findHostsForPriority(priorityIndex: number, candidates: HostInfo[]): HostInfo[] { + const priority = this.priorities[priorityIndex]; + if (!priority) { + return []; + } + const filtered = this.filterAccessible(candidates); + this.updatePrimaryRegion(filtered); + switch (priority.type) { + case GdbMonitoringConnectionPriority.STRICT_WRITER_PRIMARY: + return filtered.filter((h) => h.role === HostRole.WRITER && this.isInPrimaryRegion(h)); + + case GdbMonitoringConnectionPriority.STRICT_WRITER_SECONDARY: + return filtered.filter((h) => h.role === HostRole.WRITER && !this.isInPrimaryRegion(h)); + + case GdbMonitoringConnectionPriority.STRICT_READER_PRIMARY: + return filtered.filter((h) => h.role === HostRole.READER && this.isInPrimaryRegion(h)); + + case GdbMonitoringConnectionPriority.STRICT_READER_SECONDARY: + return filtered.filter((h) => h.role === HostRole.READER && !this.isInPrimaryRegion(h)); + + case GdbMonitoringConnectionPriority.WRITER_OR_READER_PRIMARY: { + const writers = filtered.filter((h) => h.role === HostRole.WRITER && this.isInPrimaryRegion(h)); + return writers.length > 0 ? writers : filtered.filter((h) => h.role === HostRole.READER && this.isInPrimaryRegion(h)); + } + + case GdbMonitoringConnectionPriority.WRITER_OR_READER_SECONDARY: { + const writers = filtered.filter((h) => h.role === HostRole.WRITER && !this.isInPrimaryRegion(h)); + return writers.length > 0 ? writers : filtered.filter((h) => h.role === HostRole.READER && !this.isInPrimaryRegion(h)); + } + + case GdbMonitoringConnectionPriority.REGION: { + const targetRegion = priority.region!; + const writers = filtered.filter((h) => h.role === HostRole.WRITER && equalsIgnoreCase(this.getHostRegion(h), targetRegion)); + return writers.length > 0 ? writers : filtered.filter((h) => equalsIgnoreCase(this.getHostRegion(h), targetRegion)); + } + + default: + return []; + } + } + + private filterAccessible(candidates: HostInfo[]): HostInfo[] { + if (!this.accessibleRegions) { + return candidates; + } + return candidates.filter((h) => { + const region = this.rdsUtils.getRdsRegion(h.host); + return region !== null && this.accessibleRegions!.includes(region.toLowerCase()); + }); + } + + private updatePrimaryRegion(candidates: HostInfo[]): void { + const writer = candidates.find((h) => h.role === HostRole.WRITER); + if (writer) { + this.primaryRegion = this.getHostRegion(writer); + } + } + + private getHostRegion(host: HostInfo): string | null { + return this.rdsUtils.getRdsRegion(host.host); + } + + private isInPrimaryRegion(host: HostInfo): boolean { + if (!this.primaryRegion) { + return false; + } + const hostRegion = this.getHostRegion(host); + return equalsIgnoreCase(this.primaryRegion, hostRegion); + } +} diff --git a/common/lib/host_list_provider/monitoring/global_aurora_topology_monitor.ts b/common/lib/host_list_provider/monitoring/global_aurora_topology_monitor.ts index 9582c522..4fa53150 100644 --- a/common/lib/host_list_provider/monitoring/global_aurora_topology_monitor.ts +++ b/common/lib/host_list_provider/monitoring/global_aurora_topology_monitor.ts @@ -15,13 +15,19 @@ */ import { ClusterTopologyMonitorImpl } from "./cluster_topology_monitor"; -import { GdbTopologyUtils, GlobalTopologyUtils } from "../global_topology_utils"; +import { GdbTopologyUtils } 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"; +import { AccessibleRegions } from "../../utils/accessible_regions"; +import { MonitoringConnectionHandler } from "./monitoring_connection_handler"; +import { GdbMonitoringConnectionHandler } from "./gdb_monitoring_connection_handler"; +import { WrapperProperties } from "../../wrapper_property"; +import { RdsUtils } from "../../utils/rds_utils"; +import { logger } from "../../../logutils"; function isGdbTopologyUtils(utils: TopologyUtils): utils is TopologyUtils & GdbTopologyUtils { return "getRegion" in utils && typeof (utils as unknown as GdbTopologyUtils).getRegion === "function"; @@ -29,6 +35,8 @@ function isGdbTopologyUtils(utils: TopologyUtils): utils is TopologyUtils & GdbT export class GlobalAuroraTopologyMonitor extends ClusterTopologyMonitorImpl { protected readonly instanceTemplatesByRegion: Map; + protected readonly accessibleRegions: string[] | null; + protected readonly gdbRdsUtils: RdsUtils = new RdsUtils(); declare public readonly topologyUtils: TopologyUtils; constructor( @@ -46,6 +54,48 @@ export class GlobalAuroraTopologyMonitor extends ClusterTopologyMonitorImpl { this.instanceTemplatesByRegion = instanceTemplatesByRegion; this.topologyUtils = topologyUtils; + this.accessibleRegions = AccessibleRegions.parse(properties); + + if (this.accessibleRegions) { + logger.debug(`GlobalAuroraTopologyMonitor: accessible regions = ${this.accessibleRegions.join(",")}`); + } + } + + protected override createConnectionHandler(): MonitoringConnectionHandler { + const homeRegion = + WrapperProperties.FAILOVER_HOME_REGION.get(this.monitoringProperties) ?? this.gdbRdsUtils.getRdsRegion(this.initialHostInfo.host); + return new GdbMonitoringConnectionHandler( + this.pluginService, + this.monitoringProperties, + this.accessibleRegions, + homeRegion, + () => this.monitoringClient, + (client) => { + this.monitoringClient = client; + } + ); + } + + protected override filterHostsForHostMonitoring(hosts: HostInfo[]): HostInfo[] { + if (!this.accessibleRegions) { + return hosts; + } + return hosts.filter((host) => { + const region = this.gdbRdsUtils.getRdsRegion(host.host); + return region !== null && this.accessibleRegions!.includes(region.toLowerCase()); + }); + } + + protected override async openAnyClientAndUpdateTopology(): Promise { + if (this.accessibleRegions) { + const region = this.gdbRdsUtils.getRdsRegion(this.initialHostInfo.host); + if (region && !this.accessibleRegions.includes(region.toLowerCase())) { + const msg = Messages.get("GlobalAuroraTopologyMonitor.initialHostNotInAccessibleRegion", this.initialHostInfo.host, region); + throw new AwsWrapperError(msg); + } + } + + return super.openAnyClientAndUpdateTopology(); } protected override async getInstanceTemplate(hostId: string, targetClient: ClientWrapper): Promise { diff --git a/common/lib/host_list_provider/monitoring/monitoring_connection_handler.ts b/common/lib/host_list_provider/monitoring/monitoring_connection_handler.ts new file mode 100644 index 00000000..42a36fa4 --- /dev/null +++ b/common/lib/host_list_provider/monitoring/monitoring_connection_handler.ts @@ -0,0 +1,222 @@ +/* + 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 { HostInfo } from "../../host_info"; +import { HostRole } from "../../host_role"; +import { PluginService } from "../../plugin_service"; +import { ClientWrapper } from "../../client_wrapper"; + +export enum MonitoringConnectionPriority { + STRICT_WRITER = "strict-writer", + STRICT_READER = "strict-reader", + WRITER_OR_READER = "writer-or-reader" +} + +export function monitoringConnectionPriorityFromValue(value: string | null): MonitoringConnectionPriority { + if (!value) { + return MonitoringConnectionPriority.STRICT_WRITER; + } + const lower = value.toLowerCase(); + switch (lower) { + case "strict-writer": + return MonitoringConnectionPriority.STRICT_WRITER; + case "strict-reader": + return MonitoringConnectionPriority.STRICT_READER; + case "writer-or-reader": + return MonitoringConnectionPriority.WRITER_OR_READER; + default: + return MonitoringConnectionPriority.STRICT_WRITER; + } +} + +export function parseMonitoringConnectionPriorities(value: string | null): MonitoringConnectionPriority[] { + if (!value) { + return [MonitoringConnectionPriority.STRICT_WRITER]; + } + const results: MonitoringConnectionPriority[] = []; + for (const part of value.split(",")) { + results.push(monitoringConnectionPriorityFromValue(part.trim())); + } + return results.length > 0 ? results : [MonitoringConnectionPriority.STRICT_WRITER]; +} + +/** + * Handles monitoring connection lifecycle: accepting connections offered by the monitor, + * and upgrading to a higher-priority connection when possible. + */ +export interface MonitoringConnectionHandler { + /** + * Called when a connection is offered to the handler (e.g., from openAnyClientAndUpdateTopology). + * The handler decides whether to accept it as the monitoring connection or reject it. + * + * @param client the offered client connection + * @param isWriter true if the connection is to a writer instance + * @param hostInfo the host info of the connection + * @returns true if the connection was accepted (handler sets it as monitoring connection), + * false if rejected (caller should close it) + */ + acceptConnection(client: ClientWrapper, isWriter: boolean, hostInfo: HostInfo): boolean; + + /** + * Offers a batch of harvested connections (from host monitor threads after panic mode resolves) + * to the handler. The handler picks the best one according to its priority and sets it as the + * monitoring connection. Returns the host of the selected connection so the caller can clean up + * the rest. + * + * @param connections map of host -> client harvested from node threads + * @param writerHostInfo the writer host (if known) + * @param topology the current topology + * @returns the host info of the selected connection, or null if none selected + */ + acceptConnections(connections: Map, writerHostInfo: HostInfo | null, topology: HostInfo[]): HostInfo | null; + + /** + * Non-blocking attempt to upgrade the monitoring connection to a higher-priority node. + * If the current connection already satisfies the highest priority, this is a no-op. + * + * @param currentTopology the current filtered cluster topology + */ + attemptConnectionUpgrade(currentTopology: HostInfo[]): Promise; + + /** + * Cleans up resources held by the handler. + */ + close(): Promise; +} + +/** + * Base class for monitoring connection handlers that manage a priority-ordered connection + * lifecycle. Subclasses provide priority-specific logic via abstract hooks. + * + * @typeParam P the priority type (e.g. MonitoringConnectionPriority, GdbPriorityConfig) + */ +export abstract class AbstractMonitoringConnectionHandler

implements MonitoringConnectionHandler { + protected readonly pluginService: PluginService; + protected readonly monitoringProperties: Map; + protected readonly priorities: P[]; + protected readonly getMonitoringClient: () => ClientWrapper | null; + protected readonly setMonitoringClient: (client: ClientWrapper | null) => void; + protected currentPriorityIndex: number = -1; + + protected constructor( + pluginService: PluginService, + monitoringProperties: Map, + priorities: P[], + getMonitoringClient: () => ClientWrapper | null, + setMonitoringClient: (client: ClientWrapper | null) => void + ) { + this.pluginService = pluginService; + this.monitoringProperties = monitoringProperties; + this.priorities = priorities; + this.getMonitoringClient = getMonitoringClient; + this.setMonitoringClient = setMonitoringClient; + } + + protected abstract getPriorityIndex(hostInfo: HostInfo, isWriter: boolean): number; + protected abstract findHostsForPriority(priorityIndex: number, candidates: HostInfo[]): HostInfo[]; + + protected effectiveIndex(priorityIndex: number): number { + return priorityIndex >= 0 ? priorityIndex : this.priorities.length; + } + + acceptConnection(client: ClientWrapper, isWriter: boolean, hostInfo: HostInfo): boolean { + const priorityIndex = this.getPriorityIndex(hostInfo, isWriter); + const effectiveIndex = this.effectiveIndex(priorityIndex); + + if (this.getMonitoringClient() === null || this.currentPriorityIndex < 0) { + this.setMonitoringClient(client); + this.currentPriorityIndex = effectiveIndex; + return true; + } + + if (effectiveIndex < this.currentPriorityIndex) { + this.setMonitoringClient(client); + this.currentPriorityIndex = effectiveIndex; + return true; + } + + return false; + } + + acceptConnections(connections: Map, writerHostInfo: HostInfo | null, topology: HostInfo[]): HostInfo | null { + if (!connections || connections.size === 0) { + return null; + } + + let bestHost: HostInfo | null = null; + let bestIndex = this.priorities.length; + + for (const [hostInfo, client] of connections) { + if (!client) { + continue; + } + const isWriter = writerHostInfo !== null && writerHostInfo.host === hostInfo.host; + const effectiveIndex = this.effectiveIndex(this.getPriorityIndex(hostInfo, isWriter)); + if (bestHost === null || effectiveIndex < bestIndex) { + bestIndex = effectiveIndex; + bestHost = hostInfo; + } + } + + if (!bestHost) { + return null; + } + + const bestClient = connections.get(bestHost); + this.setMonitoringClient(bestClient); + this.currentPriorityIndex = bestIndex; + return bestHost; + } + + async attemptConnectionUpgrade(currentTopology: HostInfo[]): Promise { + if (this.currentPriorityIndex <= 0) { + return; + } + + const candidates = this.findUpgradeCandidates(currentTopology); + if (candidates.length === 0) { + return; + } + + for (const candidate of candidates) { + try { + const newClient = await this.pluginService.forceConnect(candidate, this.monitoringProperties); + const oldClient = this.getMonitoringClient(); + this.setMonitoringClient(newClient); + const isWriter = candidate.role === HostRole.WRITER; + this.currentPriorityIndex = this.effectiveIndex(this.getPriorityIndex(candidate, isWriter)); + await oldClient?.abort(); + return; + } catch { + // Try next candidate. + } + } + } + + private findUpgradeCandidates(hosts: HostInfo[]): HostInfo[] { + const candidates: HostInfo[] = []; + const limit = Math.min(this.currentPriorityIndex, this.priorities.length); + for (let i = 0; i < limit; i++) { + const matching = this.findHostsForPriority(i, hosts); + candidates.push(...matching); + } + return candidates; + } + + async close(): Promise { + this.currentPriorityIndex = -1; + } +} diff --git a/common/lib/partial_plugin_service.ts b/common/lib/partial_plugin_service.ts index be99ac06..5bb41216 100644 --- a/common/lib/partial_plugin_service.ts +++ b/common/lib/partial_plugin_service.ts @@ -40,7 +40,7 @@ import { FullServicesContainer } from "./utils/full_services_container"; import { HostListProviderService } from "./host_list_provider_service"; import { StorageService } from "./utils/storage/storage_service"; import { CoreServicesContainer } from "./utils/core_services_container"; -import type { TrackedConnectionListHost } from "./plugins/connection_tracker/tracked_connection_list"; +import type { TrackedConnection } from "./plugins/connection_tracker/tracked_connection_list"; /** * A PluginService containing some methods that are not intended to be called. This class is intended to be used @@ -63,7 +63,7 @@ export class PartialPluginService implements PluginService, HostListProviderServ protected readonly driverDialect: DriverDialect; protected allowedAndBlockedHosts: AllowedAndBlockedHosts | null = null; private _isPooledClient: boolean = false; - private _trackedConnectionHost: TrackedConnectionListHost | null = null; + private _trackedConnectionHost: TrackedConnection | null = null; private connectionUrlParser: ConnectionUrlParser; constructor( @@ -529,11 +529,11 @@ export class PartialPluginService implements PluginService, HostListProviderServ this._isPooledClient = isPooledClient; } - getTrackedConnectionHost(): TrackedConnectionListHost | null { + getTrackedConnectionHost(): TrackedConnection | null { return this._trackedConnectionHost; } - setTrackedConnectionHost(host: TrackedConnectionListHost | null): void { + setTrackedConnectionHost(host: TrackedConnection | null): void { this._trackedConnectionHost = host; } } diff --git a/common/lib/plugin_service.ts b/common/lib/plugin_service.ts index f2f0af52..80ba3020 100644 --- a/common/lib/plugin_service.ts +++ b/common/lib/plugin_service.ts @@ -47,7 +47,7 @@ import { AllowedAndBlockedHosts } from "./allowed_and_blocked_hosts"; import { ConnectionPlugin } from "./connection_plugin"; import { FullServicesContainer } from "./utils/full_services_container"; import { StorageService } from "./utils/storage/storage_service"; -import type { TrackedConnectionListHost } from "./plugins/connection_tracker/tracked_connection_list"; +import type { TrackedConnection } from "./plugins/connection_tracker/tracked_connection_list"; export interface PluginService extends ErrorHandler { isInTransaction(): boolean; @@ -158,9 +158,9 @@ export interface PluginService extends ErrorHandler { setIsPooledClient(isPooledClient: boolean): void; - getTrackedConnectionHost(): TrackedConnectionListHost | null; + getTrackedConnectionHost(): TrackedConnection | null; - setTrackedConnectionHost(host: TrackedConnectionListHost | null): void; + setTrackedConnectionHost(host: TrackedConnection | null): void; } export class PluginServiceImpl implements PluginService, HostListProviderService { @@ -184,7 +184,7 @@ export class PluginServiceImpl implements PluginService, HostListProviderService private allowedAndBlockedHosts: AllowedAndBlockedHosts | null = null; protected _isPooledClient: boolean = false; - protected _trackedConnectionHost: TrackedConnectionListHost | null = null; + protected _trackedConnectionHost: TrackedConnection | null = null; constructor( container: FullServicesContainer, @@ -808,11 +808,11 @@ export class PluginServiceImpl implements PluginService, HostListProviderService this._isPooledClient = isPooledClient; } - getTrackedConnectionHost(): TrackedConnectionListHost | null { + getTrackedConnectionHost(): TrackedConnection | null { return this._trackedConnectionHost; } - setTrackedConnectionHost(host: TrackedConnectionListHost | null): void { + setTrackedConnectionHost(host: TrackedConnection | null): void { this._trackedConnectionHost = host; } } diff --git a/common/lib/plugins/aurora_initial_connection_strategy_plugin.ts b/common/lib/plugins/aurora_initial_connection_strategy_plugin.ts index 82688be6..55bba0c5 100644 --- a/common/lib/plugins/aurora_initial_connection_strategy_plugin.ts +++ b/common/lib/plugins/aurora_initial_connection_strategy_plugin.ts @@ -28,12 +28,14 @@ import { sleep } from "../utils/utils"; import { HostAvailability } from "../host_availability/host_availability"; import { logger } from "../../logutils"; import { ClientWrapper } from "../client_wrapper"; +import { AccessibleRegions } from "../utils/accessible_regions"; export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlugin { private static readonly subscribedMethods = new Set(["initHostProvider", "connect"]); private pluginService: PluginService; private hostListProviderService?: HostListProviderService; private rdsUtils = new RdsUtils(); + private accessibleRegions: string[] | null = null; constructor(pluginService: PluginService) { super(); @@ -60,6 +62,8 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu isInitialConnection: boolean, connectFunc: () => Promise ): Promise { + this.accessibleRegions = AccessibleRegions.parse(props); + const type = this.rdsUtils.identifyRdsType(hostInfo.host); if (!type.isRdsCluster) { @@ -228,14 +232,17 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu } private getWriter(): HostInfo | null { - return this.pluginService.getAllHosts().find((x) => x.role === HostRole.WRITER) ?? null; + return this.getAccessibleHosts().find((x) => x.role === HostRole.WRITER) ?? null; } private getReader(props: Map): HostInfo | undefined { const strategy = WrapperProperties.READER_HOST_SELECTOR_STRATEGY.get(props); if (this.pluginService.acceptsStrategy(HostRole.READER, strategy)) { try { - return this.pluginService.getHostInfoByStrategy(HostRole.READER, strategy); + // Restrict strategy-based selection to accessible regions so the initial connection never + // targets a host we can't reach. When no regions are configured, the full host list is used. + const accessibleReaders = this.accessibleRegions ? this.getAccessibleHosts().filter((x) => x.role === HostRole.READER) : undefined; + return this.pluginService.getHostInfoByStrategy(HostRole.READER, strategy, accessibleReaders); } catch (error: any) { // Host isn't found logger.error(error.message); @@ -245,6 +252,23 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu } private hasNoReaders(): boolean { - return this.pluginService.getAllHosts().find((x) => x.role === HostRole.READER) !== undefined; + return this.getAccessibleHosts().find((x) => x.role === HostRole.READER) === undefined; + } + + /** + * Returns the current host list filtered to the configured accessible regions. When no accessible + * regions are configured, the full host list is returned unchanged. Region filtering is applied + * before any strategy or role-based selection so the initial connection never targets an + * unreachable region. + */ + private getAccessibleHosts(): HostInfo[] { + const hosts = this.pluginService.getAllHosts(); + if (!this.accessibleRegions) { + return hosts; + } + return hosts.filter((host) => { + const region = this.rdsUtils.getRdsRegion(host.host); + return region !== null && this.accessibleRegions!.includes(region.toLowerCase()); + }); } } diff --git a/common/lib/plugins/connection_tracker/opened_connection_tracker.ts b/common/lib/plugins/connection_tracker/opened_connection_tracker.ts index 764827aa..f5c594a5 100644 --- a/common/lib/plugins/connection_tracker/opened_connection_tracker.ts +++ b/common/lib/plugins/connection_tracker/opened_connection_tracker.ts @@ -21,7 +21,7 @@ import { logger } from "../../../logutils"; import { MapUtils } from "../../utils/map_utils"; import { Messages } from "../../utils/messages"; import { PluginService } from "../../plugin_service"; -import { TrackedConnectionList, TrackedConnectionListHost } from "./tracked_connection_list"; +import { TrackedConnectionList, TrackedConnection } from "./tracked_connection_list"; export class OpenedConnectionTracker { static readonly openedConnections: Map = new Map(); @@ -32,7 +32,7 @@ export class OpenedConnectionTracker { this.pluginService = pluginService; } - populateOpenedConnectionQueue(hostInfo: HostInfo, client: ClientWrapper): TrackedConnectionListHost | null { + populateOpenedConnectionQueue(hostInfo: HostInfo, client: ClientWrapper): TrackedConnection | null { if (!hostInfo || !client) { return null; } @@ -45,7 +45,7 @@ export class OpenedConnectionTracker { } // It might be a custom domain name. Let's track by hostId and custom domain name. - let lastHost: TrackedConnectionListHost | null = null; + let lastHost: TrackedConnection | null = null; if (hostInfo.hostId) { lastHost = this.trackConnection(hostInfo.hostId, client); } @@ -78,7 +78,7 @@ export class OpenedConnectionTracker { } } - removeConnectionTracking(host: TrackedConnectionListHost | null): void { + removeConnectionTracking(host: TrackedConnection | null): void { host?.remove(); } @@ -97,7 +97,7 @@ export class OpenedConnectionTracker { } } - private trackConnection(instanceEndpoint: string, client: ClientWrapper): TrackedConnectionListHost { + private trackConnection(instanceEndpoint: string, client: ClientWrapper): TrackedConnection { const connectionList = MapUtils.computeIfAbsent(OpenedConnectionTracker.openedConnections, instanceEndpoint, (_) => new TrackedConnectionList()); return connectionList!.add(client); } diff --git a/common/lib/plugins/gdb_failover/global_db_failover_plugin.ts b/common/lib/plugins/gdb_failover/global_db_failover_plugin.ts index dd366c1c..ed7a5905 100644 --- a/common/lib/plugins/gdb_failover/global_db_failover_plugin.ts +++ b/common/lib/plugins/gdb_failover/global_db_failover_plugin.ts @@ -30,6 +30,7 @@ import { ReaderFailoverResult } from "../failover/reader_failover_result"; import { containsHostAndPort, convertNanosToMs, equalsIgnoreCase, getTimeInNanos, getWriter, logTopology, sleep } from "../../utils/utils"; import { Failover2Plugin } from "../failover2/failover2_plugin"; import { FullServicesContainer } from "../../utils/full_services_container"; +import { AccessibleRegions } from "../../utils/accessible_regions"; export class GlobalDbFailoverPlugin extends Failover2Plugin { private static readonly TELEMETRY_FAILOVER = "failover"; @@ -37,6 +38,7 @@ export class GlobalDbFailoverPlugin extends Failover2Plugin { protected activeHomeFailoverMode: GlobalDbFailoverMode = GlobalDbFailoverMode.UNKNOWN; protected inactiveHomeFailoverMode: GlobalDbFailoverMode = GlobalDbFailoverMode.UNKNOWN; protected homeRegion: string | null = null; + protected accessibleRegions: string[] | null = null; constructor(servicesContainer: FullServicesContainer, properties: Map, rdsHelper: RdsUtils) { super(servicesContainer, properties, rdsHelper); @@ -95,6 +97,17 @@ export class GlobalDbFailoverPlugin extends Failover2Plugin { } } + this.accessibleRegions = AccessibleRegions.parse(this.properties); + if (this.accessibleRegions) { + logger.debug(Messages.get("Failover.parameterValue", "gdbAccessibleRegions", this.accessibleRegions.join(","))); + + // The home region must be reachable. If it is excluded from the accessible regions, failover + // candidate filtering would always drop it, so fail loudly at configuration time. + if (this.homeRegion && !this.accessibleRegions.includes(this.homeRegion.toLowerCase())) { + throw new AwsWrapperError(Messages.get("Gdb.homeRegionNotAccessible", this.homeRegion, this.accessibleRegions.join(","))); + } + } + logger.debug(Messages.get("Failover.parameterValue", "activeHomeFailoverMode", this.activeHomeFailoverMode)); logger.debug(Messages.get("Failover.parameterValue", "inactiveHomeFailoverMode", this.inactiveHomeFailoverMode)); } @@ -119,17 +132,26 @@ export class GlobalDbFailoverPlugin extends Failover2Plugin { throw new FailoverFailedError(Messages.get("Failover.unableToRefreshHostList")); } - const updatedHosts = this.pluginService.getAllHosts(); + const allHosts = this.pluginService.getAllHosts(); + const updatedHosts = await this.filterByAccessibleRegions(allHosts); const writerCandidate = getWriter(updatedHosts); if (!writerCandidate) { this.failoverWriterTriggeredCounter.inc(); this.failoverWriterFailedCounter.inc(); - const message = logTopology(updatedHosts, Messages.get("Failover.unableToDetermineWriter")); + const message = logTopology(allHosts, Messages.get("Failover.unableToDetermineWriter")); logger.error(message); throw new FailoverFailedError(message); } + if (this.accessibleRegions && !this.isHostInAccessibleRegion(writerCandidate)) { + this.failoverWriterTriggeredCounter.inc(); + this.failoverWriterFailedCounter.inc(); + const writerRegion = this.rdsHelper.getRdsRegion(writerCandidate.host) ?? "unknown"; + logger.error(Messages.get("GlobalDbFailoverPlugin.writerInInaccessibleRegion", writerCandidate.host, writerRegion)); + throw new FailoverFailedError(Messages.get("GlobalDbFailoverPlugin.writerInInaccessibleRegion", writerCandidate.host, writerRegion)); + } + // Check writer region to determine failover mode const writerRegion = this.rdsHelper.getRdsRegion(writerCandidate.host); const isHomeRegion = equalsIgnoreCase(this.homeRegion, writerRegion); @@ -144,29 +166,28 @@ export class GlobalDbFailoverPlugin extends Failover2Plugin { break; case GlobalDbFailoverMode.STRICT_HOME_READER: await this.failoverToAllowedHost( - () => this.pluginService.getHosts().filter((x) => x.role === HostRole.READER && this.isHostInHomeRegion(x)), + () => this.getAccessibleHosts().filter((x) => x.role === HostRole.READER && this.isHostInHomeRegion(x)), HostRole.READER, failoverEndTimeNs ); break; case GlobalDbFailoverMode.STRICT_OUT_OF_HOME_READER: await this.failoverToAllowedHost( - () => this.pluginService.getHosts().filter((x) => x.role === HostRole.READER && !this.isHostInHomeRegion(x)), + () => this.getAccessibleHosts().filter((x) => x.role === HostRole.READER && !this.isHostInHomeRegion(x)), HostRole.READER, failoverEndTimeNs ); break; case GlobalDbFailoverMode.STRICT_ANY_READER: await this.failoverToAllowedHost( - () => this.pluginService.getHosts().filter((x) => x.role === HostRole.READER), + () => this.getAccessibleHosts().filter((x) => x.role === HostRole.READER), HostRole.READER, failoverEndTimeNs ); break; case GlobalDbFailoverMode.HOME_READER_OR_WRITER: await this.failoverToAllowedHost( - () => - this.pluginService.getHosts().filter((x) => x.role === HostRole.WRITER || (x.role === HostRole.READER && this.isHostInHomeRegion(x))), + () => this.getAccessibleHosts().filter((x) => x.role === HostRole.WRITER || (x.role === HostRole.READER && this.isHostInHomeRegion(x))), null, failoverEndTimeNs ); @@ -174,15 +195,13 @@ export class GlobalDbFailoverPlugin extends Failover2Plugin { case GlobalDbFailoverMode.OUT_OF_HOME_READER_OR_WRITER: await this.failoverToAllowedHost( () => - this.pluginService - .getHosts() - .filter((x) => x.role === HostRole.WRITER || (x.role === HostRole.READER && !this.isHostInHomeRegion(x))), + this.getAccessibleHosts().filter((x) => x.role === HostRole.WRITER || (x.role === HostRole.READER && !this.isHostInHomeRegion(x))), null, failoverEndTimeNs ); break; case GlobalDbFailoverMode.ANY_READER_OR_WRITER: - await this.failoverToAllowedHost(() => [...this.pluginService.getHosts()], null, failoverEndTimeNs); + await this.failoverToAllowedHost(() => this.getAccessibleHosts(), null, failoverEndTimeNs); break; case GlobalDbFailoverMode.UNKNOWN: default: @@ -202,8 +221,33 @@ export class GlobalDbFailoverPlugin extends Failover2Plugin { } private isHostInHomeRegion(host: HostInfo): boolean { + const hostRegion: string | null = this.rdsHelper.getRdsRegion(host.host); + return hostRegion !== null && equalsIgnoreCase(hostRegion, this.homeRegion); + } + + private isHostInAccessibleRegion(host: HostInfo): boolean { + if (!this.accessibleRegions) { + return true; + } const hostRegion = this.rdsHelper.getRdsRegion(host.host); - return equalsIgnoreCase(hostRegion, this.homeRegion); + return hostRegion !== null && this.accessibleRegions.includes(hostRegion.toLowerCase()); + } + + private getAccessibleHosts(): HostInfo[] { + const hosts = this.pluginService.getHosts(); + if (!this.accessibleRegions) { + return hosts; + } + return hosts.filter((x) => this.isHostInAccessibleRegion(x)); + } + + private async filterByAccessibleRegions(hosts: HostInfo[]): Promise { + if (!this.accessibleRegions) { + return hosts; + } + const dialect = this.pluginService.getDialect(); + // filterAvailableHosts is an optional SPI method; non-region-aware dialects don't implement it. + return dialect.filterAvailableHosts ? dialect.filterAvailableHosts(hosts, this.accessibleRegions) : hosts; } protected async failoverToWriter(writerCandidate: HostInfo): Promise { diff --git a/common/lib/plugins/read_write_splitting/gdb_read_write_splitting_plugin.ts b/common/lib/plugins/read_write_splitting/gdb_read_write_splitting_plugin.ts index b9f4d94f..bdd7a060 100644 --- a/common/lib/plugins/read_write_splitting/gdb_read_write_splitting_plugin.ts +++ b/common/lib/plugins/read_write_splitting/gdb_read_write_splitting_plugin.ts @@ -23,12 +23,14 @@ import { Messages } from "../../utils/messages"; import { logger } from "../../../logutils"; import { ClientWrapper } from "../../client_wrapper"; import { equalsIgnoreCase } from "../../utils/utils"; +import { AccessibleRegions } from "../../utils/accessible_regions"; export class GdbReadWriteSplittingPlugin extends ReadWriteSplittingPlugin { protected readonly rdsUtils: RdsUtils = new RdsUtils(); protected restrictWriterToHomeRegion: boolean; protected restrictReaderToHomeRegion: boolean; + protected accessibleRegions: string[] | null = null; protected isInitialized: boolean = false; protected homeRegion: string; @@ -52,6 +54,18 @@ export class GdbReadWriteSplittingPlugin extends ReadWriteSplittingPlugin { throw new ReadWriteSplittingError(Messages.get("GdbReadWriteSplittingPlugin.missingHomeRegion", initHostInfo.host)); } + this.accessibleRegions = AccessibleRegions.parse(properties); + if (this.accessibleRegions) { + logger.debug(Messages.get("GdbReadWriteSplittingPlugin.parameterValue", "gdbAccessibleRegions", this.accessibleRegions.join(","))); + + // The home region must be reachable. If it is excluded from the accessible regions, every + // reader/writer selection would filter it out, so fail loudly at connect time rather than + // surfacing confusing "no available hosts" errors later. + if (!this.accessibleRegions.includes(this.homeRegion.toLowerCase())) { + throw new ReadWriteSplittingError(Messages.get("Gdb.homeRegionNotAccessible", this.homeRegion, this.accessibleRegions.join(","))); + } + } + logger.debug(Messages.get("GdbReadWriteSplittingPlugin.parameterValue", "gdbRwHomeRegion", this.homeRegion)); this.isInitialized = true; @@ -68,6 +82,11 @@ export class GdbReadWriteSplittingPlugin extends ReadWriteSplittingPlugin { } override setWriterClient(writerTargetClient: ClientWrapper | undefined, writerHostInfo: HostInfo) { + if (writerHostInfo != null && !this.isHostInAccessibleRegion(writerHostInfo)) { + const writerRegion = this.rdsUtils.getRdsRegion(writerHostInfo.host) ?? "unknown"; + throw new ReadWriteSplittingError(Messages.get("GdbReadWriteSplittingPlugin.writerInInaccessibleRegion", writerHostInfo.host, writerRegion)); + } + if ( this.restrictWriterToHomeRegion && writerHostInfo != null && @@ -81,16 +100,35 @@ export class GdbReadWriteSplittingPlugin extends ReadWriteSplittingPlugin { } protected getReaderHostCandidates(): HostInfo[] { + let candidates = this.pluginService.getHosts(); + + if (this.accessibleRegions) { + candidates = candidates.filter((x) => this.isHostInAccessibleRegion(x)); + } + if (this.restrictReaderToHomeRegion) { - const hostsInRegion: HostInfo[] = this.pluginService - .getHosts() - .filter((x) => equalsIgnoreCase(this.rdsUtils.getRdsRegion(x.host), this.homeRegion)); + const hostsInRegion = candidates.filter((x) => equalsIgnoreCase(this.rdsUtils.getRdsRegion(x.host), this.homeRegion)); if (hostsInRegion.length === 0) { throw new ReadWriteSplittingError(Messages.get("GdbReadWriteSplittingPlugin.noAvailableReadersInHomeRegion", this.homeRegion)); } return hostsInRegion; } - return super.getReaderHostCandidates(); + + if (this.accessibleRegions && candidates.length === 0) { + throw new ReadWriteSplittingError( + Messages.get("GdbReadWriteSplittingPlugin.noAvailableReadersInAccessibleRegions", this.accessibleRegions.join(",")) + ); + } + + return candidates.length > 0 ? candidates : super.getReaderHostCandidates(); + } + + private isHostInAccessibleRegion(host: HostInfo): boolean { + if (!this.accessibleRegions) { + return true; + } + const hostRegion = this.rdsUtils.getRdsRegion(host.host); + return hostRegion !== null && this.accessibleRegions.includes(hostRegion.toLowerCase()); } } diff --git a/common/lib/utils/accessible_regions.ts b/common/lib/utils/accessible_regions.ts new file mode 100644 index 00000000..b7f29c22 --- /dev/null +++ b/common/lib/utils/accessible_regions.ts @@ -0,0 +1,33 @@ +/* + 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 { WrapperProperties } from "../wrapper_property"; + +export class AccessibleRegions { + static parse(props: Map): string[] | null { + const value = WrapperProperties.GDB_ACCESSIBLE_REGIONS.get(props); + if (!value || value.trim().length === 0) { + return null; + } + + const regions = value + .split(",") + .map((r: string) => r.trim().toLowerCase()) + .filter((r: string) => r.length > 0); + + return regions.length > 0 ? regions : null; + } +} diff --git a/common/lib/utils/messages.ts b/common/lib/utils/messages.ts index e7aebe3d..08d85603 100644 --- a/common/lib/utils/messages.ts +++ b/common/lib/utils/messages.ts @@ -315,12 +315,16 @@ const MESSAGES: Record = { "ClusterTopologyMonitor.errorDuringMonitoring": "Error thrown during cluster topology monitoring: '%s'.", "ClusterTopologyMonitor.endMonitoring": "Stop cluster topology monitoring.", "ClusterTopologyMonitor.matchingReaderTopologies": "Reader topologies have been consistent for '%s' ms. Updating topology cache.", + "ClusterTopologyMonitor.exitPanicModeViaReaderConsensus": + "Exiting panic mode by adopting harvested reader connection to '%s' as the monitoring connection. The writer is likely in an inaccessible region, so a verified writer connection cannot be established.", "ClusterTopologyMonitor.reset": "[clusterId: '%s'] Resetting cluster topology monitor for '%s'.", "ClusterTopologyMonitor.resetEventReceived": "MonitorResetEvent received.", "HostMonitor.startMonitoring": "Host monitor '%s' started.", "HostMonitor.detectedWriter": "Detected writer: '%s'.", "HostMonitor.endMonitoring": "Host monitor '%s' completed in '%s' ms.", "HostMonitor.writerHostChanged": "Writer host has changed from '%s' to '%s'.", + "HostMonitor.writerChangeExitTriggered": + "A reader observed that the writer changed to '%s'. Since some regions are inaccessible, signalling the topology monitor to exit panic mode via reader consensus.", "HostMonitor.writerIsStale": "Connected writer instance '%s' is stale.", "HostMonitor.loginErrorDuringMonitoring": "Login error detected during monitoring.", "SlidingExpirationCacheWithCleanupTask.cleaningUp": "Cleanup interval of '%s' minutes has passed, cleaning up sliding expiration cache '%s'.", @@ -408,7 +412,12 @@ const MESSAGES: Record = { "Utils.globalClusterInstanceHostPatternsRequired": "The 'globalClusterInstanceHostPatterns' property is required for Global Aurora Databases.", "Utils.invalidPatternFormat": "Invalid pattern format '%s'. Expected format: 'region:host-pattern' (e.g., 'us-east-1:?.cluster-xyz.us-east-1.rds.amazonaws.com').", + "AuroraMonitoringConnectionHandler.initialized": "AuroraMonitoringConnectionHandler initialized with priorities: '%s'.", + "GdbMonitoringConnectionHandler.initialized": "GdbMonitoringConnectionHandler initialized with priorities: '%s'.", + "GdbMonitoringConnectionHandler.unrecognizedPriority": + "Unrecognized 'gdbMonitoringConnectionPriority' value '%s'. It does not match a known priority variant and does not look like an AWS region, so it will be treated as a region literal that never matches. This is likely a typo.", "GlobalAuroraTopologyMonitor.cannotFindRegionTemplate": "Cannot find cluster instance template for region '%s'.", + "GlobalAuroraTopologyMonitor.initialHostNotInAccessibleRegion": "Initial host '%s' in region '%s' is not within the list of accessible regions.", "GlobalAuroraTopologyMonitor.invalidTopologyUtils": "TopologyUtils must implement GdbTopologyUtils for GlobalAuroraTopologyMonitor.", "GlobalDbFailoverPlugin.missingHomeRegion": "The 'failoverHomeRegion' property is required when connecting to a Global Aurora Database without a region in the URL.", @@ -420,11 +429,17 @@ const MESSAGES: Record = { "GlobalDbFailoverPlugin.unableToFindCandidateWithMatchingRole": "Unable to find a candidate host with the expected role (%s) based on the given host selection strategy: %s", "GlobalDbFailoverPlugin.unableToConnect": "Unable to establish a connection during Global DB failover.", + "GlobalDbFailoverPlugin.writerInInaccessibleRegion": + "Writer host '%s' from region '%s' is not within the list of accessible regions. Failover cannot proceed.", "GdbReadWriteSplittingPlugin.missingHomeRegion": "Unable to parse home region from endpoint '%s'. Please ensure you have set the 'gdbRwHomeRegion' connection parameter.", "GdbReadWriteSplittingPlugin.cantConnectWriterOutOfHomeRegion": "Writer connection to '%s' is not allowed since it is out of home region '%s'.", - "GdbReadWriteSplittingPlugin.noAvailableReadersInHomeRegion": "No available reader nodes in home region '%s'.", + "GdbReadWriteSplittingPlugin.writerInInaccessibleRegion": "Writer host '%s' from region '%s' is not within the list of accessible regions.", + "GdbReadWriteSplittingPlugin.noAvailableReadersInHomeRegion": "No available reader hosts in home region '%s'.", + "GdbReadWriteSplittingPlugin.noAvailableReadersInAccessibleRegions": "No available reader hosts in accessible regions '%s'.", "GdbReadWriteSplittingPlugin.parameterValue": "%s=%s", + "Gdb.homeRegionNotAccessible": + "The home region '%s' must be included in the accessible regions '%s'. Please add the home region to 'gdbAccessibleRegions' or adjust the configured home region.", "BatchingEventPublisher.errorDeliveringImmediateEvent": "Error delivering immediate event: %s", "WrapperProperty.invalidValue": "Invalid value '%s' for property '%s'. Allowed values: %s" }; diff --git a/common/lib/utils/region_utils.ts b/common/lib/utils/region_utils.ts index 4c5a008a..333e3ee4 100644 --- a/common/lib/utils/region_utils.ts +++ b/common/lib/utils/region_utils.ts @@ -88,8 +88,8 @@ export class RegionUtils { } const region = regionString.toLowerCase().trim(); - if (!RegionUtils.REGIONS.includes(regionString)) { - throw new AwsWrapperError(Messages.get("AwsSdk.unsupportedRegion", regionString)); + if (!RegionUtils.REGIONS.includes(region)) { + throw new AwsWrapperError(Messages.get("AwsSdk.unsupportedRegion", region)); } return region; diff --git a/common/lib/wrapper_property.ts b/common/lib/wrapper_property.ts index efb65a77..e3c14e9c 100644 --- a/common/lib/wrapper_property.ts +++ b/common/lib/wrapper_property.ts @@ -575,6 +575,33 @@ export class WrapperProperties { true ); + static readonly GDB_ACCESSIBLE_REGIONS = new WrapperProperty( + "gdbAccessibleRegions", + "Comma-separated list of AWS regions that are accessible from this application. " + + "When specified, the wrapper restricts Global Aurora Database operations to the listed regions only. " + + "Regions not included in this list will be filtered out from topology information, " + + "failover candidates, and read/write splitting targets.", + null + ); + + static readonly MONITORING_CONNECTION_PRIORITY = new WrapperProperty( + "monitoringConnectionPriority", + "Defines the priority for monitoring connections. " + + "Determines which type of node the topology monitor should connect to for monitoring purposes.", + "strict-writer", + ["strict-writer", "strict-reader", "writer-or-reader"] + ); + + static readonly GDB_MONITORING_CONNECTION_PRIORITY = new WrapperProperty( + "gdbMonitoringConnectionPriority", + "Defines the priority for monitoring connections in a Global Aurora Database context. " + + "Supports region-aware variants that direct the topology monitor to connect to preferred node types " + + "or specific regions. Possible values include: strict-writer-primary, strict-writer-secondary, " + + "strict-reader-primary, strict-reader-secondary, writer-or-reader-primary, writer-or-reader-secondary, " + + "or a specific AWS region name.", + null + ); + private static readonly PREFIXES = [ WrapperProperties.MONITORING_PROPERTY_PREFIX, WrapperProperties.TOPOLOGY_MONITORING_PROPERTY_PREFIX, diff --git a/docs/README.md b/docs/README.md index cbb466a7..f11ef3fb 100644 --- a/docs/README.md +++ b/docs/README.md @@ -20,6 +20,8 @@ - [Host Monitoring Plugin](./using-the-nodejs-wrapper/using-plugins/UsingTheHostMonitoringPlugin.md) - [Read-Write Splitting Plugin](./using-the-nodejs-wrapper/using-plugins/UsingTheReadWriteSplittingPlugin.md) - [GDB Read-Write Splitting Plugin](./using-the-nodejs-wrapper/using-plugins/UsingTheGdbReadWriteSplittingPlugin.md) + - [Global Aurora Accessible Regions](./using-the-nodejs-wrapper/using-plugins/UsingGlobalAuroraAccessibleRegions.md) + - [Monitoring Connection Priority](./using-the-nodejs-wrapper/using-plugins/UsingMonitoringConnectionPriority.md) - [Fastest Response Strategy Plugin](./using-the-nodejs-wrapper/using-plugins/UsingTheFastestResponseStrategyPlugin.md) - [Okta Authentication Plugin](./using-the-nodejs-wrapper/using-plugins/UsingTheOktaAuthPlugin.md) - [Federated Authentication Plugin](./using-the-nodejs-wrapper/using-plugins/UsingTheFederatedAuthPlugin.md) diff --git a/docs/using-the-nodejs-wrapper/using-plugins/UsingGlobalAuroraAccessibleRegions.md b/docs/using-the-nodejs-wrapper/using-plugins/UsingGlobalAuroraAccessibleRegions.md new file mode 100644 index 00000000..730fe758 --- /dev/null +++ b/docs/using-the-nodejs-wrapper/using-plugins/UsingGlobalAuroraAccessibleRegions.md @@ -0,0 +1,101 @@ +# Global Aurora Accessible Regions + +The `gdbAccessibleRegions` parameter allows applications to restrict Global Aurora Database operations to a subset of accessible AWS regions. This is useful when network policies, compliance requirements, or latency constraints prevent an application from reaching certain geographic regions in a Global Aurora Database. + +## Feature Availability + +This feature is available since version 3.0.0. + +## Overview + +When specified, the `gdbAccessibleRegions` parameter filters out hosts from inaccessible regions across all Global Aurora Database operations: + +- **Topology Monitoring** - Skips monitoring host workers for excluded regions, reducing unnecessary network calls. +- **Failover** - Filters failover candidates to only accessible regions. Fails fast with an error if the writer is in an inaccessible region rather than attempting connections that will time out. +- **Read/Write Splitting** - Rejects writer connections to inaccessible regions and filters reader hosts to only those in accessible regions. +- **Initial Connection Strategy** - Delegates region filtering to the dialect layer, ensuring initial connections only target accessible hosts. + +## Configuration + +| Parameter | Value | Required | Description | Default Value | +| ---------------------- | :------: | :------: | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------- | +| `gdbAccessibleRegions` | `String` | No | Comma-separated list of AWS regions that are accessible from this application. When specified, the wrapper restricts Global Aurora Database operations to the listed regions only. Regions not included in this list will be filtered out from topology information and connection targets. | `null` | + +## Usage + +Set the `gdbAccessibleRegions` parameter to a comma-separated list of AWS region names that your application can reach. + +```typescript +const params = { + plugins: "initialConnection,gdbFailover,efm2", + wrapperDialect: "global-aurora-pg", + failoverHomeRegion: "us-west-2", + globalClusterInstanceHostPatterns: "?.XYZ1.us-east-1.rds.amazonaws.com,?.XYZ2.us-east-2.rds.amazonaws.com,?.XYZ3.us-west-2.rds.amazonaws.com", + gdbAccessibleRegions: "us-west-2,us-east-1" + // Add other connection properties below... +}; + +// If using MySQL: +const client = new AwsMySQLClient(params); +await client.connect(); + +// If using Postgres: +const client = new AwsPGClient(params); +await client.connect(); +``` + +In this example, the application can only reach `us-west-2` and `us-east-1`. Any hosts in `us-east-2` will be filtered out from topology monitoring, failover candidates, and read/write splitting targets. + +## Behavior + +### When the writer is in an inaccessible region + +If the current writer host resides in a region not listed in `gdbAccessibleRegions`, the wrapper will throw an error rather than silently retrying connections that cannot succeed. This fail-fast behavior prevents long timeouts and makes it clear to the application that the writer is currently unreachable. + +### Interaction with `failoverHomeRegion` + +The `gdbAccessibleRegions` parameter works alongside the `failoverHomeRegion` parameter. While `failoverHomeRegion` defines the preferred region for failover logic, `gdbAccessibleRegions` defines which regions can be reached at all. If a `failoverHomeRegion` is specified that is not in the `gdbAccessibleRegions` list, the home region will be unreachable. + +> [!WARNING] +> Ensure that `failoverHomeRegion` is included in the `gdbAccessibleRegions` list. Otherwise, home region failover logic will not function correctly. + +### Interaction with GDB Read/Write Splitting + +When using the `gdbReadWriteSplitting` plugin, accessible regions filtering is applied before reader/writer host selection. The `gdbRwHomeRegion` should also be included in the accessible regions list. + +## Configuration Examples + +### Example 1: Application restricted to two regions + +**Scenario:** An application deployed in `us-west-2` connects to a Global Database spanning `us-east-1`, `us-east-2`, and `us-west-2`. Network policy blocks access to `us-east-2`. + +```typescript +const params = { + plugins: "initialConnection,gdbFailover,efm2", + wrapperDialect: "global-aurora-pg", + failoverHomeRegion: "us-west-2", + globalClusterInstanceHostPatterns: "?.XYZ1.us-east-1.rds.amazonaws.com,?.XYZ2.us-east-2.rds.amazonaws.com,?.XYZ3.us-west-2.rds.amazonaws.com", + gdbAccessibleRegions: "us-west-2,us-east-1", + activeHomeFailoverMode: "strict-writer", + inactiveHomeFailoverMode: "strict-writer" +}; +``` + +### Example 2: Application restricted to home region only + +**Scenario:** An application must only connect to instances in its own region for data sovereignty compliance. + +```typescript +const params = { + plugins: "initialConnection,gdbFailover,efm2", + wrapperDialect: "global-aurora-mysql", + failoverHomeRegion: "eu-west-1", + globalClusterInstanceHostPatterns: "?.XYZ1.us-east-1.rds.amazonaws.com,?.XYZ2.eu-west-1.rds.amazonaws.com,?.XYZ3.ap-southeast-1.rds.amazonaws.com", + gdbAccessibleRegions: "eu-west-1", + activeHomeFailoverMode: "strict-home-reader", + inactiveHomeFailoverMode: "strict-home-reader" +}; +``` + +> [!NOTE] +> When restricting to a single region, be aware that if the writer fails over to another region, the application will only have access to reader hosts. Configure your failover mode accordingly (e.g., `strict-home-reader`). diff --git a/docs/using-the-nodejs-wrapper/using-plugins/UsingMonitoringConnectionPriority.md b/docs/using-the-nodejs-wrapper/using-plugins/UsingMonitoringConnectionPriority.md new file mode 100644 index 00000000..c4c08d73 --- /dev/null +++ b/docs/using-the-nodejs-wrapper/using-plugins/UsingMonitoringConnectionPriority.md @@ -0,0 +1,118 @@ +# Monitoring Connection Priority + +The monitoring connection priority parameters allow you to control which type of node the topology monitor connects to for monitoring purposes. This is useful for optimizing monitoring connections in both standard Aurora clusters and Global Aurora Databases. + +## Feature Availability + +This feature is available since version 3.0.0. + +## Overview + +By default, the topology monitor connects to a writer node to observe cluster topology changes. However, in some scenarios it may be preferable to direct monitoring connections to a reader node or to a node in a specific region. + +Two parameters are available: + +- **`monitoringConnectionPriority`** - For standard Aurora clusters. Controls the node type used for monitoring connections. +- **`gdbMonitoringConnectionPriority`** - For Global Aurora Databases. Extends the standard parameter with region-aware options. + +## Configuration Parameters + +| Parameter | Value | Required | Description | Default Value | +| --------------------------------- | :------: | :------: | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------- | +| `monitoringConnectionPriority` | `String` | No | Defines the priority for monitoring connections. Determines which type of host the topology monitor should connect to.

Possible values: `strict-writer`, `strict-reader`, `writer-or-reader`. | `strict-writer` | +| `gdbMonitoringConnectionPriority` | `String` | No | Defines the priority for monitoring connections in a Global Aurora Database context. Supports region-aware variants and specific region names.

See [GDB Monitoring Connection Priority Values](#gdb-monitoring-connection-priority-values). | `null` | + +## Monitoring Connection Priority Values + +### Standard Values (`monitoringConnectionPriority`) + +| Value | Description | +| ------------------ | ------------------------------------------------------------------------------------------------------------- | +| `strict-writer` | The topology monitor connects exclusively to a writer host. If a writer is unavailable, monitoring will fail. | +| `strict-reader` | The topology monitor connects exclusively to a reader host. If no reader is available, monitoring will fail. | +| `writer-or-reader` | The topology monitor connects to a writer host if available; otherwise falls back to a reader host. | + +### GDB Values (`gdbMonitoringConnectionPriority`) + +| Value | Description | +| ---------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- | +| `strict-writer-primary` | Connect to the writer host in the primary region of the Global Database. | +| `strict-writer-secondary` | Connect to a writer host in a secondary (non-primary) region of the Global Database. | +| `strict-reader-primary` | Connect to a reader host in the primary region of the Global Database. | +| `strict-reader-secondary` | Connect to a reader host in a secondary (non-primary) region of the Global Database. | +| `writer-or-reader-primary` | Connect to a writer in the primary region if available; otherwise fall back to a reader in the primary region. | +| `writer-or-reader-secondary` | Connect to a writer in a secondary region if available; otherwise fall back to a reader in a secondary region. | +| `` | Connect to any available host in the specified AWS region (e.g., `us-west-2`). The monitor will attempt writer first, then reader in that region. | + +## Usage + +### Standard Aurora Cluster + +```typescript +const params = { + plugins: "failover2,efm2", + monitoringConnectionPriority: "writer-or-reader" + // Add other connection properties below... +}; + +// If using MySQL: +const client = new AwsMySQLClient(params); +await client.connect(); + +// If using Postgres: +const client = new AwsPGClient(params); +await client.connect(); +``` + +### Global Aurora Database + +```typescript +const params = { + plugins: "initialConnection,gdbFailover,efm2", + wrapperDialect: "global-aurora-pg", + failoverHomeRegion: "us-west-2", + globalClusterInstanceHostPatterns: "?.XYZ1.us-east-1.rds.amazonaws.com,?.XYZ2.us-west-2.rds.amazonaws.com", + gdbMonitoringConnectionPriority: "strict-writer-primary" + // Add other connection properties below... +}; + +// If using MySQL: +const client = new AwsMySQLClient(params); +await client.connect(); + +// If using Postgres: +const client = new AwsPGClient(params); +await client.connect(); +``` + +### Using a specific region for monitoring + +```typescript +const params = { + plugins: "initialConnection,gdbFailover,efm2", + wrapperDialect: "global-aurora-pg", + failoverHomeRegion: "us-west-2", + globalClusterInstanceHostPatterns: "?.XYZ1.us-east-1.rds.amazonaws.com,?.XYZ2.us-west-2.rds.amazonaws.com", + gdbMonitoringConnectionPriority: "us-west-2" + // Add other connection properties below... +}; +``` + +## Interaction with Accessible Regions + +When `gdbAccessibleRegions` is configured, the monitoring connection priority respects the accessible regions filter. If the preferred monitoring target is in an inaccessible region, the monitor will fall back to an available host in an accessible region. + +> [!WARNING] +> If `gdbMonitoringConnectionPriority` specifies a region that is not in the `gdbAccessibleRegions` list, the monitoring connection may fail. Ensure consistency between these parameters. + +## Async Upgrade Semantics + +The topology monitor may start with a connection to an available host and asynchronously upgrade to a higher-priority host when one becomes available. For example, if `strict-writer-primary` is configured but the primary writer is temporarily unavailable, the monitor may temporarily connect to another host and upgrade once the primary writer is reachable. + +## Tuning Guidance + +- Use `strict-writer` (default) for most applications. Writer connections provide the most accurate and timely topology information. +- Use `strict-reader` when you want to minimize load on the writer host and can tolerate slightly delayed topology updates. +- Use `writer-or-reader` for maximum monitoring availability at the cost of potentially connecting to a reader that may have slightly stale topology information. +- For Global Databases, prefer `strict-writer-primary` to get the most up-to-date topology from the primary region's writer. +- Use a specific region name when you want to keep monitoring traffic local to reduce cross-region latency. diff --git a/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbFailoverPlugin.md b/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbFailoverPlugin.md index 91e1b1f1..3f81e838 100644 --- a/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbFailoverPlugin.md +++ b/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbFailoverPlugin.md @@ -63,6 +63,8 @@ In addition to the parameters that you can configure for the underlying driver, | `telemetryFailoverAdditionalTopTrace` | `Boolean` | No | Allows the driver to produce an additional telemetry span associated with failover. Such span helps to facilitate telemetry analysis in AWS CloudWatch. | `false` | | `skipFailoverOnInterruptedThread` | `Boolean` | No | Enable to skip failover if the current thread is interrupted. This may leave the Connection in an invalid state so the Connection should be disposed. | `false` | | `skipInactiveWriterClusterEndpointCheck` | `Boolean` | No | Enable to skip a connection role verification when connection is opened with inactive cluster writer endpoint. This parameter is applicable to Aurora Global Databases. Region-bound cluster writer endpoints are available in Global Databases. However, depending on the GDB primary region, they may be inactive and may act differently. The parameter is aimed at supporting such inactive cluster endpoints and helps configure the desired behavior for them. | `false` | +| `gdbAccessibleRegions` | `String` | No | Comma-separated list of AWS regions that are accessible from this application. When specified, the wrapper restricts Global Aurora Database operations to the listed regions only. Regions not included in this list will be filtered out from topology information, failover candidates, and read/write splitting targets. See [Global Aurora Accessible Regions](./UsingGlobalAuroraAccessibleRegions.md). | | +| `gdbMonitoringConnectionPriority` | `String` | No | Defines the priority for monitoring connections in a Global Aurora Database context. Supports region-aware variants that direct the topology monitor to connect to preferred node types or specific regions. Possible values: `strict-writer-primary`, `strict-writer-secondary`, `strict-reader-primary`, `strict-reader-secondary`, `writer-or-reader-primary`, `writer-or-reader-secondary`, or a specific AWS region name. See [Monitoring Connection Priority](./UsingMonitoringConnectionPriority.md). | | Please refer to the original [Failover Plugin](./UsingTheFailoverPlugin.md) and [Failover2 Plugin](./UsingTheFailover2Plugin.md) for more details about error codes, configurations, connection pooling and sample codes. diff --git a/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbReadWriteSplittingPlugin.md b/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbReadWriteSplittingPlugin.md index 22932ddd..0ab88109 100644 --- a/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbReadWriteSplittingPlugin.md +++ b/docs/using-the-nodejs-wrapper/using-plugins/UsingTheGdbReadWriteSplittingPlugin.md @@ -53,5 +53,6 @@ The GDB Read/Write Splitting plugin can be used against Aurora clusters and RDS | `gdbRwHomeRegion` | `str` | If connecting using an IP address, a custom domain URL, Global Database endpoint or other endpoint with no region: Yes

Otherwise: No | Defines a home region.

Examples: `us-west-2`, `us-east-1`.

If this parameter is omitted, the value is parsed from the connection configuration. For regional cluster endpoints and instance endpoints, it's set to the region of the provided endpoint. If the provided endpoint has no region (for example, a Global Database endpoint or IP address), the configuration parameter is mandatory. | For regional cluster endpoints and instance endpoints, it's set to the region of the provided endpoint.

Otherwise: `null` | | `gdbRwRestrictWriterToHomeRegion` | `bool` | No | If set to `true`, prevents following and connecting to a writer node outside the defined home region. An exception will be raised when such a connection to a writer outside the home region is requested. | `true` | | `gdbRwRestrictReaderToHomeRegion` | `bool` | No | If set to `true`, prevents connecting to a reader node outside the defined home region. If no reader nodes in the home region are available, an exception will be raised. | `true` | +| `gdbAccessibleRegions` | `str` | No | Comma-separated list of AWS regions that are accessible from this application. When specified, reader and writer host selection is restricted to these regions only. See [Global Aurora Accessible Regions](./UsingGlobalAuroraAccessibleRegions.md). | | Please refer to the original [Read/Write Splitting plugin](./UsingTheReadWriteSplittingPlugin.md) for more details about error codes, configurations, connection pooling and sample codes. diff --git a/mysql/lib/dialect/global_aurora_mysql_database_dialect.ts b/mysql/lib/dialect/global_aurora_mysql_database_dialect.ts index faf89a3c..668fea64 100644 --- a/mysql/lib/dialect/global_aurora_mysql_database_dialect.ts +++ b/mysql/lib/dialect/global_aurora_mysql_database_dialect.ts @@ -20,10 +20,14 @@ import { ClientWrapper } from "../../../common/lib/client_wrapper"; import { TopologyQueryResult } from "../../../common/lib/host_list_provider/topology_utils"; import { FullServicesContainer } from "../../../common/lib/utils/full_services_container"; import { HostListProvider } from "../../../common/lib/host_list_provider/host_list_provider"; +import { HostInfo } from "../../../common/lib/host_info"; import { GlobalAuroraHostListProvider } from "../../../common/lib/host_list_provider/global_aurora_host_list_provider"; import { GlobalTopologyUtils } from "../../../common/lib/host_list_provider/global_topology_utils"; +import { RdsUtils } from "../../../common/lib/utils/rds_utils"; export class GlobalAuroraMySQLDatabaseDialect extends AuroraMySQLDatabaseDialect implements GlobalAuroraTopologyDialect { + private readonly rdsUtils: RdsUtils = new RdsUtils(); + private static readonly GLOBAL_STATUS_TABLE_EXISTS_QUERY = "SELECT 1 AS tmp FROM information_schema.tables WHERE" + " upper(table_schema) = 'INFORMATION_SCHEMA' AND upper(table_name) = 'AURORA_GLOBAL_DB_STATUS'"; @@ -81,6 +85,17 @@ export class GlobalAuroraMySQLDatabaseDialect extends AuroraMySQLDatabaseDialect ); } + async filterAvailableHosts(hosts: HostInfo[], accessibleRegions: string[]): Promise { + if (!accessibleRegions || accessibleRegions.length === 0) { + return hosts; + } + const lowerRegions = accessibleRegions.map((r) => r.toLowerCase()); + return hosts.filter((host) => { + const region = this.rdsUtils.getRdsRegion(host.host); + return region !== null && lowerRegions.includes(region.toLowerCase()); + }); + } + async queryForTopology(targetClient: ClientWrapper): Promise { const res = await targetClient.query(GlobalAuroraMySQLDatabaseDialect.GLOBAL_TOPOLOGY_QUERY); const results: TopologyQueryResult[] = []; diff --git a/mysql/lib/dialect/mysql_database_dialect.ts b/mysql/lib/dialect/mysql_database_dialect.ts index f0fe20ad..84a0dad5 100644 --- a/mysql/lib/dialect/mysql_database_dialect.ts +++ b/mysql/lib/dialect/mysql_database_dialect.ts @@ -28,6 +28,7 @@ import { ErrorHandler } from "../../../common/lib/error_handler"; import { MySQLErrorHandler } from "../mysql_error_handler"; import { Messages } from "../../../common/lib/utils/messages"; import { HostRole } from "../../../common/lib/host_role"; +import { HostInfo } from "../../../common/lib/host_info"; import { FullServicesContainer } from "../../../common/lib/utils/full_services_container"; export class MySQLDatabaseDialect implements DatabaseDialect { @@ -208,6 +209,10 @@ export class MySQLDatabaseDialect implements DatabaseDialect { return undefined; } + async filterAvailableHosts(hosts: HostInfo[], accessibleRegions: string[]): Promise { + return hosts; + } + async getHostRole(targetClient: ClientWrapper): Promise { throw new UnsupportedMethodError(`Method getHostRole not supported for dialect: ${this.dialectName}`); } diff --git a/pg/lib/dialect/global_aurora_pg_database_dialect.ts b/pg/lib/dialect/global_aurora_pg_database_dialect.ts index 9bc88611..bd31005a 100644 --- a/pg/lib/dialect/global_aurora_pg_database_dialect.ts +++ b/pg/lib/dialect/global_aurora_pg_database_dialect.ts @@ -20,10 +20,14 @@ import { ClientWrapper } from "../../../common/lib/client_wrapper"; import { TopologyQueryResult } from "../../../common/lib/host_list_provider/topology_utils"; import { FullServicesContainer } from "../../../common/lib/utils/full_services_container"; import { HostListProvider } from "../../../common/lib/host_list_provider/host_list_provider"; +import { HostInfo } from "../../../common/lib/host_info"; import { GlobalAuroraHostListProvider } from "../../../common/lib/host_list_provider/global_aurora_host_list_provider"; import { GlobalTopologyUtils } from "../../../common/lib/host_list_provider/global_topology_utils"; +import { RdsUtils } from "../../../common/lib/utils/rds_utils"; export class GlobalAuroraPgDatabaseDialect extends AuroraPgDatabaseDialect implements GlobalAuroraTopologyDialect { + private readonly rdsUtils: RdsUtils = new RdsUtils(); + private static readonly GLOBAL_STATUS_FUNC_EXISTS_QUERY = "select 'pg_catalog.aurora_global_db_status'::regproc"; private static readonly GLOBAL_INSTANCE_STATUS_FUNC_EXISTS_QUERY = "select 'pg_catalog.aurora_global_db_instance_status'::regproc"; @@ -90,6 +94,17 @@ export class GlobalAuroraPgDatabaseDialect extends AuroraPgDatabaseDialect imple ); } + async filterAvailableHosts(hosts: HostInfo[], accessibleRegions: string[]): Promise { + if (!accessibleRegions || accessibleRegions.length === 0) { + return hosts; + } + const lowerRegions = accessibleRegions.map((r) => r.toLowerCase()); + return hosts.filter((host) => { + const region = this.rdsUtils.getRdsRegion(host.host); + return region !== null && lowerRegions.includes(region.toLowerCase()); + }); + } + async queryForTopology(targetClient: ClientWrapper): Promise { const res = await targetClient.queryWithTimeout(GlobalAuroraPgDatabaseDialect.GLOBAL_TOPOLOGY_QUERY); const hosts: TopologyQueryResult[] = []; diff --git a/pg/lib/dialect/pg_database_dialect.ts b/pg/lib/dialect/pg_database_dialect.ts index 0afa54f2..9e70f466 100644 --- a/pg/lib/dialect/pg_database_dialect.ts +++ b/pg/lib/dialect/pg_database_dialect.ts @@ -25,6 +25,7 @@ import { FailoverRestriction } from "../../../common/lib/plugins/failover/failov import { ErrorHandler } from "../../../common/lib/error_handler"; import { PgErrorHandler } from "../pg_error_handler"; import { Messages } from "../../../common/lib/utils/messages"; +import { HostInfo } from "../../../common/lib/host_info"; import { FullServicesContainer } from "../../../common/lib/utils/full_services_container"; export class PgDatabaseDialect implements DatabaseDialect { @@ -189,6 +190,10 @@ export class PgDatabaseDialect implements DatabaseDialect { return undefined; } + async filterAvailableHosts(hosts: HostInfo[], accessibleRegions: string[]): Promise { + return hosts; + } + async getHostRole(targetClient: ClientWrapper): Promise { throw new UnsupportedMethodError(`Method getHostRole not supported for dialect: ${this.dialectName}`); } diff --git a/tests/integration/host/src/test/java/integration/host/DriverHelper.java b/tests/integration/host/src/test/java/integration/host/DriverHelper.java index 3deacaec..5199432b 100644 --- a/tests/integration/host/src/test/java/integration/host/DriverHelper.java +++ b/tests/integration/host/src/test/java/integration/host/DriverHelper.java @@ -59,10 +59,6 @@ public static String getDriverClassname(TestDriver testDriver) { } } - public static void setConnectTimeout(Properties props, long timeout, TimeUnit timeUnit) { - setConnectTimeout(TestEnvironment.getCurrent().getCurrentDriver(), props, timeout, timeUnit); - } - public static void setConnectTimeout( TestDriver testDriver, Properties props, long timeout, TimeUnit timeUnit) { switch (testDriver) { diff --git a/tests/unit/accessible_regions.test.ts b/tests/unit/accessible_regions.test.ts new file mode 100644 index 00000000..185fbe4b --- /dev/null +++ b/tests/unit/accessible_regions.test.ts @@ -0,0 +1,49 @@ +/* + 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 { AccessibleRegions } from "../../common/lib/utils/accessible_regions"; +import { WrapperProperties } from "../../common/lib/wrapper_property"; + +describe("AccessibleRegions", () => { + let props: Map; + + beforeEach(() => { + props = new Map(); + }); + + it.each([ + [undefined, "not set"], + ["", "empty string"], + [" ", "whitespace only"], + [",,", "only commas"] + ])("returns null when property is %s (%s)", (value, _desc) => { + if (value !== undefined) { + props.set(WrapperProperties.GDB_ACCESSIBLE_REGIONS.name, value); + } + expect(AccessibleRegions.parse(props)).toBeNull(); + }); + + it.each([ + ["us-east-1", ["us-east-1"], "single region"], + ["us-east-1,us-west-2,eu-west-1", ["us-east-1", "us-west-2", "eu-west-1"], "multiple regions"], + ["US-EAST-1,Us-West-2", ["us-east-1", "us-west-2"], "normalizes to lowercase"], + [" us-east-1 , us-west-2 ", ["us-east-1", "us-west-2"], "trims whitespace"], + ["us-east-1,,us-west-2,", ["us-east-1", "us-west-2"], "filters empty entries from trailing comma"] + ])("parses '%s' → %j (%s)", (input, expected) => { + props.set(WrapperProperties.GDB_ACCESSIBLE_REGIONS.name, input); + expect(AccessibleRegions.parse(props)).toEqual(expected); + }); +}); diff --git a/tests/unit/aurora_initial_connection_strategy_plugin.test.ts b/tests/unit/aurora_initial_connection_strategy_plugin.test.ts index 920057ec..4bf0886c 100644 --- a/tests/unit/aurora_initial_connection_strategy_plugin.test.ts +++ b/tests/unit/aurora_initial_connection_strategy_plugin.test.ts @@ -125,7 +125,7 @@ describe("Aurora initial connection strategy plugin", () => { when(mockPluginService.connect(anything(), anything(), anything())).thenResolve(readerClient); when(mockPluginService.acceptsStrategy(anything(), anything())).thenReturn(true); when(mockPluginService.getHostRole(readerClient)).thenReturn(Promise.resolve(HostRole.READER)); - when(mockPluginService.getHostInfoByStrategy(anything(), anything())).thenReturn(instance(mockReaderHostInfo)); + when(mockPluginService.getHostInfoByStrategy(anything(), anything(), anything())).thenReturn(instance(mockReaderHostInfo)); expect(await plugin.connect(hostInfo, props, true, mockFunc)).toBe(readerClient); verify(mockPluginService.forceRefreshHostList()).never(); @@ -142,13 +142,13 @@ describe("Aurora initial connection strategy plugin", () => { it("test reader - return writer", async () => { when(mockRdsUtils.identifyRdsType(anything())).thenReturn(RdsUrlType.RDS_READER_CLUSTER); - when(mockPluginService.getAllHosts()) - .thenReturn([hostInfoBuilder.withRole(HostRole.READER).build()]) - .thenReturn([hostInfoBuilder.withRole(HostRole.WRITER).build()]); + // The cluster has no reader instances, so hasNoReaders() is true and the plugin falls back to + // returning the writer connection (simulating Aurora reader-cluster-endpoint behavior). + when(mockPluginService.getAllHosts()).thenReturn([hostInfoBuilder.withRole(HostRole.WRITER).build()]); when(mockPluginService.connect(anything(), anything(), anything())).thenResolve(writerClient); when(mockPluginService.acceptsStrategy(anything(), anything())).thenReturn(true); when(mockPluginService.getHostRole(writerClient)).thenReturn(Promise.resolve(HostRole.WRITER)); - when(mockPluginService.getHostInfoByStrategy(anything(), anything())).thenReturn(instance(mockReaderHostInfo)); + when(mockPluginService.getHostInfoByStrategy(anything(), anything(), anything())).thenReturn(instance(mockReaderHostInfo)); expect(await plugin.connect(hostInfo, props, true, mockFunc)).toBe(writerClient); }); diff --git a/tests/unit/aurora_monitoring_connection_handler.test.ts b/tests/unit/aurora_monitoring_connection_handler.test.ts new file mode 100644 index 00000000..286c5ec0 --- /dev/null +++ b/tests/unit/aurora_monitoring_connection_handler.test.ts @@ -0,0 +1,211 @@ +/* + 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 { HostInfo } from "../../common/lib/host_info"; +import { HostRole } from "../../common/lib/host_role"; +import { HostInfoBuilder } from "../../common/lib/host_info_builder"; +import { SimpleHostAvailabilityStrategy } from "../../common/lib/host_availability/simple_host_availability_strategy"; +import { PluginService } from "../../common/lib/plugin_service"; +import { WrapperProperties } from "../../common/lib/wrapper_property"; +import { AuroraMonitoringConnectionHandler } from "../../common/lib/host_list_provider/monitoring/aurora_monitoring_connection_handler"; +import { ClientWrapper } from "../../common/lib/client_wrapper"; +import { instance, mock, when, anything } from "ts-mockito"; + +const builder = new HostInfoBuilder({ hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }); + +const writerHost = builder.withHost("writer.cluster-abc.us-east-1.rds.amazonaws.com").withRole(HostRole.WRITER).build(); +const readerHost1 = builder.withHost("reader1.cluster-abc.us-east-1.rds.amazonaws.com").withRole(HostRole.READER).build(); +const readerHost2 = builder.withHost("reader2.cluster-abc.us-east-1.rds.amazonaws.com").withRole(HostRole.READER).build(); + +const allCandidates = [writerHost, readerHost1, readerHost2]; + +describe("AuroraMonitoringConnectionHandler", () => { + let mockPluginService: PluginService; + let props: Map; + let monitoringClient: ClientWrapper | null; + + beforeEach(() => { + mockPluginService = mock(); + props = new Map(); + monitoringClient = null; + }); + + function createHandler(priority: string | null): AuroraMonitoringConnectionHandler { + if (priority) { + props.set(WrapperProperties.MONITORING_CONNECTION_PRIORITY.name, priority); + } + return new AuroraMonitoringConnectionHandler( + instance(mockPluginService), + props, + () => monitoringClient, + (client) => { + monitoringClient = client; + } + ); + } + + describe("acceptConnection", () => { + it("accepts when monitoringClient is null", () => { + const handler = createHandler(null); + const mockClient = {} as ClientWrapper; + const result = handler.acceptConnection(mockClient, true, writerHost); + expect(result).toBe(true); + expect(monitoringClient).toBe(mockClient); + }); + + it("rejects when offered connection is not higher priority", () => { + const handler = createHandler(null); + const firstClient = {} as ClientWrapper; + handler.acceptConnection(firstClient, true, writerHost); + + const newClient = {} as ClientWrapper; + const result = handler.acceptConnection(newClient, false, readerHost1); + expect(result).toBe(false); + expect(monitoringClient).toBe(firstClient); + }); + + it("replaces when offered connection is higher priority", () => { + const handler = createHandler(null); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerHost1); + + const writerClient = {} as ClientWrapper; + const result = handler.acceptConnection(writerClient, true, writerHost); + expect(result).toBe(true); + expect(monitoringClient).toBe(writerClient); + }); + }); + + describe("attemptConnectionUpgrade", () => { + it("does not upgrade when already at best priority (writer with strict-writer)", async () => { + const handler = createHandler("strict-writer"); + const existingClient = {} as ClientWrapper; + handler.acceptConnection(existingClient, true, writerHost); + + when(mockPluginService.forceConnect(anything(), anything())).thenResolve({} as any); + await handler.attemptConnectionUpgrade(allCandidates); + + // Should still be the same client — no upgrade needed. + expect(monitoringClient).toBe(existingClient); + }); + + it("upgrades from reader to writer with strict-writer priority", async () => { + const handler = createHandler("strict-writer"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerHost1); + + const writerClient = { abort: async () => {} } as unknown as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(writerClient); + + await handler.attemptConnectionUpgrade(allCandidates); + + expect(monitoringClient).toBe(writerClient); + }); + + it("does not upgrade when reader with strict-reader priority", async () => { + const handler = createHandler("strict-reader"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerHost1); + + await handler.attemptConnectionUpgrade(allCandidates); + + expect(monitoringClient).toBe(readerClient); + }); + + it("does not upgrade when monitoringClient is null", async () => { + const handler = createHandler("strict-writer"); + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBeNull(); + }); + + it("upgrades from writer to reader with strict-reader priority after acceptConnection records the index", async () => { + // Regression coverage for the panic-mode writer-detected path: the monitor now offers the verified + // writer connection through acceptConnection (rather than assigning monitoringClient directly), so the + // handler records the writer's priority index. With a strict-reader priority, a subsequent upgrade must + // then be able to move to a reader. If the index were not recorded, attemptConnectionUpgrade would + // short-circuit and the configured priority would be silently ignored. + const handler = createHandler("strict-reader"); + const writerClient = { abort: async () => {} } as unknown as ClientWrapper; + handler.acceptConnection(writerClient, true, writerHost); + + const readerClient = {} as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(readerClient); + + await handler.attemptConnectionUpgrade(allCandidates); + + expect(monitoringClient).toBe(readerClient); + }); + + it("does not upgrade when no suitable candidate exists", async () => { + const handler = createHandler("strict-writer"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerHost1); + + await handler.attemptConnectionUpgrade([readerHost1, readerHost2]); + + expect(monitoringClient).toBe(readerClient); + }); + + it("keeps current connection when forceConnect fails", async () => { + const handler = createHandler("strict-writer"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerHost1); + + when(mockPluginService.forceConnect(anything(), anything())).thenReject(new Error("conn refused")); + await handler.attemptConnectionUpgrade(allCandidates); + + expect(monitoringClient).toBe(readerClient); + }); + }); + + describe("acceptConnections", () => { + it("selects preferred host from connections map", () => { + const handler = createHandler("strict-writer"); + const writerClient = {} as ClientWrapper; + const readerClient = {} as ClientWrapper; + const connections = new Map([ + [readerHost1, readerClient], + [writerHost, writerClient] + ]); + + const selected = handler.acceptConnections(connections, writerHost, allCandidates); + expect(selected).toBe(writerHost); + expect(monitoringClient).toBe(writerClient); + }); + + it("falls back to any connection when preferred not in map", () => { + const handler = createHandler("strict-writer"); + const readerClient = {} as ClientWrapper; + const connections = new Map([[readerHost1, readerClient]]); + + const selected = handler.acceptConnections(connections, null, [readerHost1, readerHost2]); + expect(selected).toBe(readerHost1); + expect(monitoringClient).toBe(readerClient); + }); + }); + + describe("writer-or-reader priority", () => { + it("does not upgrade when holding a reader with writer-or-reader priority", async () => { + const handler = createHandler("writer-or-reader"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerHost1); + + // writer-or-reader is satisfied by any connection, so no upgrade needed. + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(readerClient); + }); + }); +}); diff --git a/tests/unit/cluster_topology_monitor.test.ts b/tests/unit/cluster_topology_monitor.test.ts new file mode 100644 index 00000000..2b8dc9c5 --- /dev/null +++ b/tests/unit/cluster_topology_monitor.test.ts @@ -0,0 +1,196 @@ +/* + 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 { anything, instance, mock, when, verify } from "ts-mockito"; +import { HostInfo } from "../../common/lib/host_info"; +import { HostInfoBuilder } from "../../common/lib/host_info_builder"; +import { HostRole } from "../../common/lib/host_role"; +import { SimpleHostAvailabilityStrategy } from "../../common/lib/host_availability/simple_host_availability_strategy"; +import { PluginService, PluginServiceImpl } from "../../common/lib/plugin_service"; +import { ClusterTopologyMonitorImpl } from "../../common/lib/host_list_provider/monitoring/cluster_topology_monitor"; +import { MonitoringConnectionHandler } from "../../common/lib/host_list_provider/monitoring/monitoring_connection_handler"; +import { FullServicesContainer } from "../../common/lib/utils/full_services_container"; +import { TopologyUtils } from "../../common/lib/host_list_provider/topology_utils"; +import { HostListProviderService } from "../../common/lib/host_list_provider_service"; +import { StorageService } from "../../common/lib/utils/storage/storage_service"; +import { DriverDialect } from "../../common/lib/driver_dialect/driver_dialect"; +import { EventPublisher } from "../../common/lib/utils/events/event"; +import { ClientWrapper } from "../../common/lib/client_wrapper"; + +const builder = new HostInfoBuilder({ hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }); +const writerHost = builder.withHost("writer.cluster-abc.us-east-1.rds.amazonaws.com").withRole(HostRole.WRITER).build(); +const readerHost1 = builder.withHost("reader1.cluster-abc.us-east-1.rds.amazonaws.com").withRole(HostRole.READER).build(); +const readerHost2 = builder.withHost("reader2.cluster-abc.us-east-1.rds.amazonaws.com").withRole(HostRole.READER).build(); +const allHosts = [writerHost, readerHost1, readerHost2]; + +class TestableClusterTopologyMonitor extends ClusterTopologyMonitorImpl { + getConnectionHandlerForTest(): MonitoringConnectionHandler { + return this.getConnectionHandler(); + } + + getMonitoringClientForTest(): ClientWrapper | null { + return this.monitoringClient; + } + + setMonitoringClientForTest(client: ClientWrapper | null): void { + this.monitoringClient = client; + } + + getWriterHostInfoForTest(): HostInfo | null { + return this.writerHostInfo; + } + + setConnectionHandlerForTest(handler: MonitoringConnectionHandler): void { + this.connectionHandler = handler; + } + + async callOpenAnyClientAndUpdateTopology(): Promise { + return this.openAnyClientAndUpdateTopology(); + } +} + +describe("ClusterTopologyMonitorImpl - connection handler integration", () => { + let mockPluginService: PluginService; + let mockTopologyUtils: TopologyUtils; + let mockStorageService: StorageService; + let mockHostListProviderService: HostListProviderService; + let mockDriverDialect: DriverDialect; + let mockEventPublisher: EventPublisher; + let servicesContainer: FullServicesContainer; + let props: Map; + let monitor: TestableClusterTopologyMonitor; + + beforeEach(() => { + mockPluginService = mock(PluginServiceImpl); + mockTopologyUtils = mock(); + mockStorageService = mock(); + mockHostListProviderService = mock(); + mockDriverDialect = mock(); + mockEventPublisher = mock(); + + when(mockPluginService.getDriverDialect()).thenReturn(instance(mockDriverDialect)); + when(mockDriverDialect.setConnectTimeout(anything(), anything())).thenReturn(); + when(mockDriverDialect.setQueryTimeout(anything(), anything(), anything())).thenReturn(); + + servicesContainer = { + pluginService: instance(mockPluginService), + storageService: instance(mockStorageService), + hostListProviderService: instance(mockHostListProviderService), + eventPublisher: instance(mockEventPublisher), + importantEventService: { registerEvent: () => {} } + } as unknown as FullServicesContainer; + + props = new Map(); + + monitor = new TestableClusterTopologyMonitor( + servicesContainer, + instance(mockTopologyUtils), + "cluster-id", + writerHost, + props, + writerHost, + 30_000_000_000, + 5_000_000_000 + ); + }); + + describe("getConnectionHandler (lazy)", () => { + it("creates handler on first access", () => { + const handler = monitor.getConnectionHandlerForTest(); + expect(handler).toBeDefined(); + expect(handler.acceptConnection).toBeDefined(); + expect(handler.acceptConnections).toBeDefined(); + expect(handler.attemptConnectionUpgrade).toBeDefined(); + expect(handler.close).toBeDefined(); + }); + + it("returns same handler on subsequent access", () => { + const handler1 = monitor.getConnectionHandlerForTest(); + const handler2 = monitor.getConnectionHandlerForTest(); + expect(handler1).toBe(handler2); + }); + }); + + describe("openAnyClientAndUpdateTopology", () => { + it("connects and offers to handler via acceptConnection", async () => { + const mockClient = {} as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(mockClient); + when(mockTopologyUtils.isWriterInstance(anything())).thenResolve(true); + when(mockTopologyUtils.queryForTopology(anything(), anything(), anything(), anything())).thenResolve(allHosts); + + await monitor.callOpenAnyClientAndUpdateTopology(); + + // Handler accepted it (default AuroraMonitoringConnectionHandler accepts when monitoringClient is null) + expect(monitor.getMonitoringClientForTest()).toBe(mockClient); + }); + + it("connects as reader — handler still accepts", async () => { + const mockClient = {} as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(mockClient); + when(mockTopologyUtils.isWriterInstance(anything())).thenResolve(false); + when(mockTopologyUtils.queryForTopology(anything(), anything(), anything(), anything())).thenResolve(allHosts); + + await monitor.callOpenAnyClientAndUpdateTopology(); + + expect(monitor.getMonitoringClientForTest()).toBe(mockClient); + }); + + it("returns null when forceConnect fails", async () => { + when(mockPluginService.forceConnect(anything(), anything())).thenReject(new Error("conn refused")); + + const result = await monitor.callOpenAnyClientAndUpdateTopology(); + + expect(result).toBeNull(); + expect(monitor.getMonitoringClientForTest()).toBeNull(); + }); + + it("closes connection when handler rejects", async () => { + let abortCallCount = 0; + const mockClient = { + abort: async () => { + abortCallCount++; + } + } as unknown as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(mockClient); + when(mockTopologyUtils.isWriterInstance(anything())).thenResolve(false); + when(mockTopologyUtils.queryForTopology(anything(), anything(), anything(), anything())).thenResolve(allHosts); + + // Set a handler that always rejects. + const mockHandler = mock(); + when(mockHandler.acceptConnection(anything(), anything(), anything())).thenReturn(false); + monitor.setConnectionHandlerForTest(instance(mockHandler)); + + await monitor.callOpenAnyClientAndUpdateTopology(); + + expect(abortCallCount).toBe(1); + }); + }); + + describe("attemptConnectionUpgrade", () => { + it("delegates to handler.attemptConnectionUpgrade", async () => { + const mockClient = {} as ClientWrapper; + monitor.setMonitoringClientForTest(mockClient); + + const mockHandler = mock(); + when(mockHandler.attemptConnectionUpgrade(anything())).thenResolve(); + monitor.setConnectionHandlerForTest(instance(mockHandler)); + + await monitor.getConnectionHandlerForTest().attemptConnectionUpgrade(allHosts); + + verify(mockHandler.attemptConnectionUpgrade(anything())).once(); + }); + }); +}); diff --git a/tests/unit/gdb_monitoring_connection_handler.test.ts b/tests/unit/gdb_monitoring_connection_handler.test.ts new file mode 100644 index 00000000..9ee2aac4 --- /dev/null +++ b/tests/unit/gdb_monitoring_connection_handler.test.ts @@ -0,0 +1,219 @@ +/* + 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 { HostInfo } from "../../common/lib/host_info"; +import { HostRole } from "../../common/lib/host_role"; +import { HostInfoBuilder } from "../../common/lib/host_info_builder"; +import { SimpleHostAvailabilityStrategy } from "../../common/lib/host_availability/simple_host_availability_strategy"; +import { PluginService } from "../../common/lib/plugin_service"; +import { WrapperProperties } from "../../common/lib/wrapper_property"; +import { GdbMonitoringConnectionHandler } from "../../common/lib/host_list_provider/monitoring/gdb_monitoring_connection_handler"; +import { ClientWrapper } from "../../common/lib/client_wrapper"; +import { instance, mock, when, anything } from "ts-mockito"; + +const builder = new HostInfoBuilder({ hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }); + +function hostInRegion(host: string, role: HostRole): HostInfo { + return builder.withHost(host).withRole(role).build(); +} + +const writerPrimary = hostInRegion("writer-instance.cluster-abc.us-east-1.rds.amazonaws.com", HostRole.WRITER); +const readerPrimary = hostInRegion("reader-instance.cluster-abc.us-east-1.rds.amazonaws.com", HostRole.READER); +const writerSecondary = hostInRegion("writer-instance.cluster-xyz.us-west-2.rds.amazonaws.com", HostRole.WRITER); +const readerSecondary = hostInRegion("reader-instance.cluster-xyz.us-west-2.rds.amazonaws.com", HostRole.READER); +const readerEuWest = hostInRegion("reader-instance.cluster-xyz.eu-west-1.rds.amazonaws.com", HostRole.READER); + +const allCandidates = [writerPrimary, readerPrimary, writerSecondary, readerSecondary, readerEuWest]; + +describe("GdbMonitoringConnectionHandler", () => { + let mockPluginService: PluginService; + let props: Map; + let monitoringClient: ClientWrapper | null; + + beforeEach(() => { + mockPluginService = mock(); + props = new Map(); + monitoringClient = null; + }); + + function createHandler(priority: string | null, accessibleRegions: string[] | null, homeRegion: string | null): GdbMonitoringConnectionHandler { + if (priority) { + props.set(WrapperProperties.GDB_MONITORING_CONNECTION_PRIORITY.name, priority); + } + return new GdbMonitoringConnectionHandler( + instance(mockPluginService), + props, + accessibleRegions, + homeRegion, + () => monitoringClient, + (client) => { + monitoringClient = client; + } + ); + } + + describe("acceptConnection", () => { + it("accepts when monitoringClient is null", () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + const mockClient = {} as ClientWrapper; + const result = handler.acceptConnection(mockClient, true, writerPrimary); + expect(result).toBe(true); + expect(monitoringClient).toBe(mockClient); + }); + + it("rejects when offered connection is not higher priority", () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + const firstClient = {} as ClientWrapper; + handler.acceptConnection(firstClient, true, writerPrimary); + + const readerClient = {} as ClientWrapper; + const result = handler.acceptConnection(readerClient, false, readerPrimary); + expect(result).toBe(false); + expect(monitoringClient).toBe(firstClient); + }); + }); + + describe("attemptConnectionUpgrade", () => { + it("does not upgrade when already at best priority", async () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + const writerClient = {} as ClientWrapper; + handler.acceptConnection(writerClient, true, writerPrimary); + + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(writerClient); + }); + + it("upgrades from reader to writer in primary region", async () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerPrimary); + + const writerClient = { abort: async () => {} } as unknown as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(writerClient); + + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(writerClient); + }); + + it("does not upgrade when monitoringClient is null", async () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBeNull(); + }); + + it("keeps current connection when forceConnect fails", async () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerPrimary); + + when(mockPluginService.forceConnect(anything(), anything())).thenReject(new Error("conn refused")); + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(readerClient); + }); + + it("does not upgrade when already at best priority", async () => { + const handler = createHandler("strict-reader-primary", null, "us-east-1"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerPrimary); + + // First upgrade sets primaryRegion and recognizes we're at priority 0. + await handler.attemptConnectionUpgrade(allCandidates); + // Second call should short-circuit since currentPriorityIndex is now 0. + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(readerClient); + }); + }); + + describe("acceptConnections", () => { + it("selects preferred host from connections map", () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + const writerClient = {} as ClientWrapper; + const readerClient = {} as ClientWrapper; + const connections = new Map([ + [readerPrimary, readerClient], + [writerPrimary, writerClient] + ]); + + const selected = handler.acceptConnections(connections, writerPrimary, allCandidates); + expect(selected).toBe(writerPrimary); + expect(monitoringClient).toBe(writerClient); + }); + + it("falls back to any connection when preferred not in map", () => { + const handler = createHandler("strict-writer-primary", null, "us-east-1"); + const readerClient = {} as ClientWrapper; + const connections = new Map([[readerSecondary, readerClient]]); + + const selected = handler.acceptConnections(connections, null, [readerSecondary]); + expect(selected).toBe(readerSecondary); + expect(monitoringClient).toBe(readerClient); + }); + }); + + describe("accessible regions filtering", () => { + it("filters out hosts not in accessible regions during upgrade", async () => { + const handler = createHandler("strict-writer-primary", ["us-west-2"], "us-east-1"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerSecondary); + + const writerClient = { abort: async () => {} } as unknown as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(writerClient); + + await handler.attemptConnectionUpgrade(allCandidates); + // Writer in us-west-2 (accessible) should be selected. + expect(monitoringClient).toBe(writerClient); + }); + + it("does not upgrade when accessible regions filter removes all better candidates", async () => { + const handler = createHandler("strict-writer-primary", ["ap-southeast-1"], "us-east-1"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerPrimary); + + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(readerClient); + }); + }); + + describe("region priority", () => { + it("selects host in specified region", async () => { + const handler = createHandler("eu-west-1", null, "us-east-1"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerPrimary); + + const euClient = { abort: async () => {} } as unknown as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(euClient); + + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(euClient); + }); + }); + + describe("multi-priority list", () => { + it("falls through priorities: strict-writer-primary,strict-reader-secondary", async () => { + const handler = createHandler("strict-writer-primary,strict-reader-secondary", null, "us-east-1"); + const readerClient = {} as ClientWrapper; + handler.acceptConnection(readerClient, false, readerSecondary); + + // readerSecondary satisfies index 1 (strict-reader-secondary). + // Should upgrade to writerPrimary (index 0). + const writerClient = { abort: async () => {} } as unknown as ClientWrapper; + when(mockPluginService.forceConnect(anything(), anything())).thenResolve(writerClient); + + await handler.attemptConnectionUpgrade(allCandidates); + expect(monitoringClient).toBe(writerClient); + }); + }); +});