diff --git a/.github/workflows/aurora_performance.yml b/.github/workflows/aurora_performance.yml index 72ebb931..5d0bd082 100644 --- a/.github/workflows/aurora_performance.yml +++ b/.github/workflows/aurora_performance.yml @@ -18,25 +18,25 @@ jobs: steps: - name: Clone repository - uses: actions/checkout@v4 + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 with: fetch-depth: 50 - name: "Set up JDK 8" - uses: actions/setup-java@v3 + uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 # v5 with: distribution: "corretto" java-version: 8 - name: Set up Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 with: - node-version: "20.x" + node-version: "22.x" - name: Install dependencies run: npm install --no-save - name: Configure AWS Credentials id: creds - uses: aws-actions/configure-aws-credentials@v4 + uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b # v6 with: role-to-assume: arn:aws:iam::${{ secrets.AWS_ACCOUNT_ID }}:role/${{ secrets.AWS_DEPLOY_ROLE }} role-session-name: nodejs_aurora_perf_tests @@ -59,7 +59,7 @@ jobs: - name: "Get Github Action IP" if: always() id: ip - uses: haythem/public-ip@v1.3 + uses: haythem/public-ip@bdddd92c198b0955f0b494a8ebeac529754262ff # v1.3 - name: "Remove Github Action IP" if: always() @@ -73,7 +73,7 @@ jobs: - name: Archive Performance results if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: ${{ matrix.db }}-performance-results path: ./tests/integration/container/reports diff --git a/.github/workflows/dependabot_auto_merge.yml b/.github/workflows/dependabot_auto_merge.yml index c7574bab..536f793b 100644 --- a/.github/workflows/dependabot_auto_merge.yml +++ b/.github/workflows/dependabot_auto_merge.yml @@ -17,7 +17,7 @@ jobs: steps: - name: Dependabot metadata id: metadata - uses: dependabot/fetch-metadata@v1 + uses: dependabot/fetch-metadata@25dd0e34f4fe68f24cc83900b1fe3fe149efef98 # v2 with: github-token: "${{ secrets.GITHUB_TOKEN }}" - name: Approve PR diff --git a/.github/workflows/integration_tests.yml b/.github/workflows/integration_tests.yml index 4b0e28e9..2218d9ea 100644 --- a/.github/workflows/integration_tests.yml +++ b/.github/workflows/integration_tests.yml @@ -5,6 +5,7 @@ on: push: branches: - main + - chore/remove-aliases paths-ignore: - '**/*.md' - '**/*.jpg' @@ -63,24 +64,24 @@ jobs: steps: - name: 'Clone repository' - uses: actions/checkout@v4 + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 with: ref: ${{ github.event.pull_request.head.sha || github.sha }} - name: "Set up JDK 8" - uses: actions/setup-java@v3 + uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 # v5 with: distribution: "corretto" java-version: 8 - name: Set up Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 with: - node-version: "20.x" + node-version: "22.x" - name: Install dependencies run: npm install --no-save - name: Configure AWS Credentials id: creds - uses: aws-actions/configure-aws-credentials@v4 + uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b # v6 with: role-to-assume: arn:aws:iam::${{ secrets.AWS_ACCOUNT_ID }}:role/${{ secrets.AWS_DEPLOY_ROLE }} role-session-name: nodejs_int_default_tests @@ -102,7 +103,7 @@ jobs: - name: "Get Github Action IP" if: always() id: ip - uses: haythem/public-ip@v1.3 + uses: haythem/public-ip@bdddd92c198b0955f0b494a8ebeac529754262ff # v1.3 - name: "Remove Github Action IP" if: always() @@ -116,7 +117,7 @@ jobs: - name: Archive results if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: integration-report-default-${{ matrix.dbEngine }} path: ./tests/integration/container/reports @@ -126,6 +127,9 @@ jobs: name: Run Integration Tests (Latest) runs-on: ubuntu-latest needs: run-integration-tests-default + if: | + always() && + needs.run-integration-tests-default.result == 'success' strategy: fail-fast: false matrix: @@ -133,22 +137,22 @@ jobs: steps: - name: Clone repository - uses: actions/checkout@v4 + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 - name: "Set up JDK 8" - uses: actions/setup-java@v3 + uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 # v5 with: distribution: "corretto" java-version: 8 - name: Set up Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 with: - node-version: "20.x" + node-version: "22.x" - name: Install dependencies run: npm install --no-save - name: Configure AWS Credentials id: creds - uses: aws-actions/configure-aws-credentials@v4 + uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b # v6 with: role-to-assume: arn:aws:iam::${{ secrets.AWS_ACCOUNT_ID }}:role/${{ secrets.AWS_DEPLOY_ROLE }} role-session-name: nodejs_int_latest_tests @@ -170,7 +174,7 @@ jobs: - name: "Get Github Action IP" if: always() id: ip - uses: haythem/public-ip@v1.3 + uses: haythem/public-ip@bdddd92c198b0955f0b494a8ebeac529754262ff # v1.3 - name: "Remove Github Action IP" if: always() @@ -184,7 +188,7 @@ jobs: - name: Archive results if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: integration-report-latest-${{ matrix.dbEngine }} path: ./tests/integration/container/reports diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index ed517c71..7ba7b23a 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -16,11 +16,11 @@ jobs: run-checks-and-unit-tests: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 - name: Set up Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 with: - node-version: "20.x" + node-version: "22.x" - name: Install dependencies run: npm install --no-save - name: Run eslint - linting diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index aedac085..dcd54a9a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -11,11 +11,11 @@ jobs: packages: write steps: - name: "Clone Repository" - uses: actions/checkout@v4 + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 - name: "Set up Node.js" - uses: actions/setup-node@v4 + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 with: - node-version: "20.x" + node-version: "22.x" registry-url: "https://registry.npmjs.org" - name: "Install dependencies" run: npm install --no-save diff --git a/.github/workflows/release_draft.yml b/.github/workflows/release_draft.yml index fb2c2207..85973b38 100644 --- a/.github/workflows/release_draft.yml +++ b/.github/workflows/release_draft.yml @@ -19,11 +19,11 @@ jobs: runs-on: ubuntu-latest steps: - name: "Clone Repository" - uses: actions/checkout@v4 + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 - name: "Set up Node.js" - uses: actions/setup-node@v4 + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 with: - node-version: "20.x" + node-version: "22.x" - name: "Install dependencies" run: npm install --no-save - name: "Run eslint - linting" @@ -46,7 +46,7 @@ jobs: touch RELEASE_DETAILS.md echo "$RELEASE_DETAILS" > RELEASE_DETAILS.md - name: "Upload to Draft Release" - uses: ncipollo/release-action@v1 + uses: ncipollo/release-action@339a81892b84b4eeb0f6e744e4574d79d0d9b8dd # v1 with: draft: true name: "AWS Advanced NodeJS Wrapper - v${{ env.RELEASE_VERSION }}" diff --git a/.github/workflows/remove-old-artifacts.yml b/.github/workflows/remove-old-artifacts.yml index 5899e867..71d497c8 100644 --- a/.github/workflows/remove-old-artifacts.yml +++ b/.github/workflows/remove-old-artifacts.yml @@ -15,7 +15,7 @@ jobs: steps: - name: Remove Old Artifacts - uses: c-hive/gha-remove-artifacts@v1 + uses: c-hive/gha-remove-artifacts@44fc7acaf1b3d0987da0e8d4707a989d80e9554b # v1 with: age: "1 week" skip-tags: true diff --git a/.github/workflows/run-autoscaling-tests.yml b/.github/workflows/run-autoscaling-tests.yml index 32d243db..0c65d3ae 100644 --- a/.github/workflows/run-autoscaling-tests.yml +++ b/.github/workflows/run-autoscaling-tests.yml @@ -21,21 +21,21 @@ jobs: dbEngine: [ "mysql", "postgres" ] steps: - name: Clone repository - uses: actions/checkout@v4 + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 with: fetch-depth: 50 - name: "Set up JDK 8" - uses: actions/setup-java@v3 + uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 # v5 with: distribution: "corretto" java-version: 8 - name: Set up Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0 with: - node-version: "20.x" + node-version: "22.x" - name: Configure AWS credentials id: creds - uses: aws-actions/configure-aws-credentials@v4 + uses: aws-actions/configure-aws-credentials@e7f100cf4c008499ea8adda475de1042d6975c7b # v6 with: role-to-assume: arn:aws:iam::${{ secrets.AWS_ACCOUNT_ID }}:role/${{ secrets.AWS_DEPLOY_ROLE }} role-session-name: nodejs_autoscaling_tests @@ -56,7 +56,7 @@ jobs: - name: "Get Github Action IP" if: always() id: ip - uses: haythem/public-ip@v1.3 + uses: haythem/public-ip@bdddd92c198b0955f0b494a8ebeac529754262ff # v1.3 - name: "Remove Github Action IP" if: always() run: | @@ -68,7 +68,7 @@ jobs: 2>&1 > /dev/null; - name: Archive results if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: name: autoscaling-report path: ./tests/integration/container/reports diff --git a/common/lib/host_info.ts b/common/lib/host_info.ts index 3ac57751..4d332a38 100644 --- a/common/lib/host_info.ts +++ b/common/lib/host_info.ts @@ -25,15 +25,13 @@ export class HostInfo { public static readonly NO_PORT: number = -1; public static readonly DEFAULT_WEIGHT: number = 100; - readonly host: string; + readonly host: string; // full domain name readonly port: number; role: HostRole; readonly weight: number; // Greater or equal 0. Lesser the weight, the healthier host. readonly lastUpdateTime: number; availability: HostAvailability; - aliases: Set = new Set(); - allAliases: Set = new Set(); - hostId: string; + hostId: string; // id; could be a host name, host domain name, or a unique string hostAvailabilityStrategy: HostAvailabilityStrategy; constructor( @@ -57,7 +55,6 @@ export class HostInfo { this.weight = weight; this.lastUpdateTime = lastUpdateTime; this.hostAvailabilityStrategy = hostAvailabilityStrategy; - this.allAliases.add(this.asAlias).add(this.host); this.hostId = hostId; } @@ -65,38 +62,6 @@ export class HostInfo { return this.port != HostInfo.NO_PORT; } - addAlias(...alias: string[]) { - if (!alias || alias.length < 1) { - return; - } - - alias.forEach((x) => { - this.aliases.add(x); - this.allAliases.add(x); - }); - } - - removeAlias(aliases: string[]) { - if (!aliases || aliases.length < 1) { - return; - } - - aliases.forEach((x) => { - this.aliases.delete(x); - this.allAliases.delete(x); - }); - } - - resetAliases() { - this.aliases.clear(); - this.allAliases.clear(); - this.allAliases.add(this.asAlias); - } - - get asAlias() { - return this.isPortSpecified() ? `${this.host}:${this.port}` : this.host; - } - get url() { let url = this.hostAndPort; if (!url.endsWith("/")) { diff --git a/common/lib/host_list_provider/topology_utils.ts b/common/lib/host_list_provider/topology_utils.ts index 500f3046..3cddf486 100644 --- a/common/lib/host_list_provider/topology_utils.ts +++ b/common/lib/host_list_provider/topology_utils.ts @@ -130,7 +130,6 @@ export abstract class TopologyUtils { .withLastUpdateTime(lastUpdateTime) .withHostId(finalInstanceId) .build(); - host.addAlias(finalEndpoint); return host; } diff --git a/common/lib/partial_plugin_service.ts b/common/lib/partial_plugin_service.ts index b3b8f729..4203b230 100644 --- a/common/lib/partial_plugin_service.ts +++ b/common/lib/partial_plugin_service.ts @@ -40,6 +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"; /** * A PluginService containing some methods that are not intended to be called. This class is intended to be used @@ -62,6 +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 connectionUrlParser: ConnectionUrlParser; constructor( @@ -221,15 +223,11 @@ export class PartialPluginService implements PluginService, HostListProviderServ return hosts; } - setAvailability(hostAliases: Set, availability: HostAvailability): void { - if (hostAliases.size === 0) { - return; - } - + setAvailability(hostInfo: HostInfo, availability: HostAvailability): void { const hostsToChange = [ ...new Set( this.getAllHosts().filter( - (host: HostInfo) => hostAliases.has(host.asAlias) || [...host.aliases].some((hostAlias: string) => hostAliases.has(hostAlias)) + (host: HostInfo) => (hostInfo.hostId != null && hostInfo.hostId === host.hostId) || (hostInfo.host != null && hostInfo.host === host.host) ) ) ]; @@ -430,36 +428,12 @@ export class PartialPluginService implements PluginService, HostListProviderServ return provider.identifyConnection(targetClient); } - async fillAliases(targetClient: ClientWrapper, hostInfo: HostInfo): Promise { - if (!hostInfo) { - return; - } - - if (hostInfo.aliases.size > 0) { - logger.debug(Messages.get("PluginService.nonEmptyAliases", [...hostInfo.aliases].join(", "))); - return; - } - - hostInfo.addAlias(hostInfo.asAlias); - - try { - const res = await this.dialect.getHostAliasAndParseResults(targetClient); - if (res) { - hostInfo.addAlias(res); - } - } catch (error) { - logger.debug(Messages.get("PluginService.failedToRetrieveHostPort")); - } + getRoutedHostInfo(): HostInfo | null { + throw new AwsWrapperError(Messages.get("PartialPluginService.unexpectedMethodCall", "getRoutedHostInfo")); + } - try { - const host = await this.identifyConnection(targetClient); - if (host && host.allAliases) { - hostInfo.addAlias(...host.allAliases); - } - } catch (error) { - // Ignore errors from identifyConnection - logger.debug(Messages.get("PluginService.failedToRetrieveHostPort")); - } + setRoutedHostInfo(routedHostInfo: HostInfo | null) { + throw new AwsWrapperError(Messages.get("PartialPluginService.unexpectedMethodCall", "setRoutedHostInfo")); } getHostInfoBuilder(): HostInfoBuilder { @@ -547,4 +521,12 @@ export class PartialPluginService implements PluginService, HostListProviderServ setIsPooledClient(isPooledClient: boolean): void { this._isPooledClient = isPooledClient; } + + getTrackedConnectionHost(): TrackedConnectionListHost | null { + return this._trackedConnectionHost; + } + + setTrackedConnectionHost(host: TrackedConnectionListHost | null): void { + this._trackedConnectionHost = host; + } } diff --git a/common/lib/plugin_service.ts b/common/lib/plugin_service.ts index 8e8a5a1a..1b80affb 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 { CoreServicesContainer } from "./utils/core_services_container"; +import type { TrackedConnectionListHost } from "./plugins/connection_tracker/tracked_connection_list"; export interface PluginService extends ErrorHandler { isInTransaction(): boolean; @@ -96,14 +96,20 @@ export interface PluginService extends ErrorHandler { getHosts(): HostInfo[]; - setAvailability(hostAliases: Set, availability: HostAvailability): void; + setAvailability(hostInfo: HostInfo, availability: HostAvailability): void; updateConfigWithProperties(props: Map): void; - fillAliases(targetClient: ClientWrapper, hostInfo: HostInfo): Promise; + getRoutedHostInfo(): HostInfo | null; + + setRoutedHostInfo(routedHostInfo: HostInfo | null); identifyConnection(targetClient: ClientWrapper): Promise; + identifyConnection(targetClient: ClientWrapper, connectionHostInfo: HostInfo | null): Promise; + + identifyConnection(targetClient: ClientWrapper, connectionHostInfo?: HostInfo | null): Promise; + connect(hostInfo: HostInfo, props: Map): Promise; connect(hostInfo: HostInfo, props: Map, pluginToSkip: ConnectionPlugin | null): Promise; @@ -151,6 +157,10 @@ export interface PluginService extends ErrorHandler { isPooledClient(): boolean; setIsPooledClient(isPooledClient: boolean): void; + + getTrackedConnectionHost(): TrackedConnectionListHost | null; + + setTrackedConnectionHost(host: TrackedConnectionListHost | null): void; } export class PluginServiceImpl implements PluginService, HostListProviderService { @@ -162,6 +172,7 @@ export class PluginServiceImpl implements PluginService, HostListProviderService private _isInTransaction: boolean = false; private servicesContainer: FullServicesContainer; protected hosts: HostInfo[] = []; + protected routedHostInfo: HostInfo | null; private dbDialectProvider: DatabaseDialectProvider; private readonly initialHost: string; private dialect: DatabaseDialect; @@ -173,6 +184,7 @@ export class PluginServiceImpl implements PluginService, HostListProviderService private allowedAndBlockedHosts: AllowedAndBlockedHosts | null = null; protected _isPooledClient: boolean = false; + protected _trackedConnectionHost: TrackedConnectionListHost | null = null; constructor( container: FullServicesContainer, @@ -459,15 +471,11 @@ export class PluginServiceImpl implements PluginService, HostListProviderService return hosts; } - setAvailability(hostAliases: Set, availability: HostAvailability) { - if (hostAliases.size === 0) { - return; - } - + setAvailability(hostInfo: HostInfo, availability: HostAvailability) { const hostsToChange = [ ...new Set( this.getAllHosts().filter( - (host: HostInfo) => hostAliases.has(host.asAlias) || [...host.aliases].some((hostAlias: string) => hostAliases.has(hostAlias)) + (host: HostInfo) => (hostInfo.hostId != null && hostInfo.hostId === host.hostId) || (hostInfo.host != null && hostInfo.host === host.host) ) ) ]; @@ -496,30 +504,12 @@ export class PluginServiceImpl implements PluginService, HostListProviderService this._currentClient.config = Object.fromEntries(props.entries()); } - async fillAliases(targetClient: ClientWrapper, hostInfo: HostInfo) { - if (hostInfo == null) { - return; - } - - if (hostInfo.aliases.size > 0) { - logger.debug(Messages.get("PluginService.nonEmptyAliases", [...hostInfo.aliases].join(", "))); - return; - } - - hostInfo.addAlias(hostInfo.asAlias); - - // Add the host name and port, this host name is usually the internal IP address. - try { - const res: string = await this.dialect.getHostAliasAndParseResults(targetClient); - hostInfo.addAlias(res); - } catch (error) { - logger.debug(Messages.get("PluginService.failedToRetrieveHostPort")); - } + getRoutedHostInfo(): HostInfo | null { + return this.routedHostInfo; + } - const host: HostInfo | void | null = await this.identifyConnection(targetClient); - if (host) { - hostInfo.addAlias(...host.allAliases); - } + setRoutedHostInfo(routedHostInfo: HostInfo | null) { + this.routedHostInfo = routedHostInfo; } identifyConnection(targetClient: ClientWrapper): Promise { @@ -806,4 +796,12 @@ export class PluginServiceImpl implements PluginService, HostListProviderService setIsPooledClient(isPooledClient: boolean): void { this._isPooledClient = isPooledClient; } + + getTrackedConnectionHost(): TrackedConnectionListHost | null { + return this._trackedConnectionHost; + } + + setTrackedConnectionHost(host: TrackedConnectionListHost | 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 7d7891a6..cdabc7d1 100644 --- a/common/lib/plugins/aurora_initial_connection_strategy_plugin.ts +++ b/common/lib/plugins/aurora_initial_connection_strategy_plugin.ts @@ -121,9 +121,7 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu continue; } - if (isInitialConnection) { - this.hostListProviderService.setInitialConnectionHostInfo(writerCandidate); - } + this.pluginService.setRoutedHostInfo(writerCandidate); } return writerCandidateClient; } @@ -139,16 +137,14 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu } // Writer connection is valid and verified. - if (isInitialConnection) { - this.hostListProviderService.setInitialConnectionHostInfo(writerCandidate); - } + this.pluginService.setRoutedHostInfo(writerCandidate); return writerCandidateClient; } catch (error: any) { await this.pluginService.abortTargetClient(writerCandidateClient); if (this.pluginService.isLoginError(error) || !writerCandidate) { throw error; } else if (writerCandidate) { - this.pluginService.setAvailability(writerCandidate.allAliases, HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(writerCandidate, HostAvailability.NOT_AVAILABLE); } } } @@ -185,9 +181,7 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu if (this.hasNoReaders()) { // It seems that cluster has no readers. Simulate Aurora reader cluster endpoint logic // and return the current (writer) client. - if (isInitialConnection) { - this.hostListProviderService.setInitialConnectionHostInfo(readerCandidate); - } + this.pluginService.setRoutedHostInfo(readerCandidate); return readerCandidateClient; } await this.pluginService.abortTargetClient(readerCandidateClient); @@ -196,9 +190,7 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu } // Reader connection is valid and verified. - if (isInitialConnection) { - this.hostListProviderService.setInitialConnectionHostInfo(readerCandidate); - } + this.pluginService.setRoutedHostInfo(readerCandidate); } else { logger.debug("Reader candidate not found"); } @@ -233,7 +225,7 @@ export class AuroraInitialConnectionStrategyPlugin extends AbstractConnectionPlu if (this.pluginService.isLoginError(error) || !readerCandidate) { throw error; } else if (readerCandidate) { - this.pluginService.setAvailability(readerCandidate.allAliases, HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(readerCandidate, HostAvailability.NOT_AVAILABLE); } } } diff --git a/common/lib/plugins/bluegreen/routing/substitute_connect_routing.ts b/common/lib/plugins/bluegreen/routing/substitute_connect_routing.ts index 94f35677..812873d7 100644 --- a/common/lib/plugins/bluegreen/routing/substitute_connect_routing.ts +++ b/common/lib/plugins/bluegreen/routing/substitute_connect_routing.ts @@ -73,7 +73,6 @@ export class SubstituteConnectRouting extends BaseConnectRouting { .withHostId(iamHost.hostId) .withAvailability(HostAvailability.AVAILABLE) .build(); - reroutedHostInfo.addAlias(iamHost.host); const reroutedProperties: Map = new Map(properties); reroutedProperties.set(WrapperProperties.HOST.name, iamHost.host); diff --git a/common/lib/plugins/connection_tracker/aurora_connection_tracker_plugin.ts b/common/lib/plugins/connection_tracker/aurora_connection_tracker_plugin.ts index 651bbea8..d256d8bc 100644 --- a/common/lib/plugins/connection_tracker/aurora_connection_tracker_plugin.ts +++ b/common/lib/plugins/connection_tracker/aurora_connection_tracker_plugin.ts @@ -28,7 +28,15 @@ import { HostRole } from "../../host_role"; import { OpenedConnectionTracker } from "./opened_connection_tracker"; export class AuroraConnectionTrackerPlugin extends AbstractConnectionPlugin implements CanReleaseResources { - private static readonly subscribedMethods = new Set(["notifyHostListChanged", "connect", "query", "rollback"]); + private static readonly subscribedMethods = new Set([ + ...SubscribedMethodHelper.NETWORK_BOUND_METHODS, + "end", + "abort", + "notifyHostListChanged" + ]); + private static readonly CLOSING_METHODS = new Set(["end", "abort"]); + private static readonly TOPOLOGY_CHANGES_EXPECTED_TIME_NS = BigInt(3 * 60 * 1_000_000_000); + private static hostListRefreshEndTimeNs: bigint = 0n; private readonly pluginService: PluginService; private readonly rdsUtils: RdsUtils; @@ -56,46 +64,92 @@ export class AuroraConnectionTrackerPlugin extends AbstractConnectionPlugin impl connectFunc: () => Promise ): Promise { const targetClient = await connectFunc(); - - if (targetClient) { - const type: RdsUrlType = this.rdsUtils.identifyRdsType(hostInfo.host); - if (type.isRdsCluster) { - hostInfo.resetAliases(); - await this.pluginService.fillAliases(targetClient, hostInfo); + let connectionHostInfo: HostInfo = this.pluginService.getRoutedHostInfo() ?? hostInfo; + + if (targetClient && !this.pluginService.isPooledClient()) { + const type: RdsUrlType = this.rdsUtils.identifyRdsType(connectionHostInfo.host); + if (type.isRdsCluster || type === RdsUrlType.OTHER || type === RdsUrlType.IP_ADDRESS) { + const identifiedHostInfo: HostInfo | null = await this.pluginService.identifyConnection(targetClient, connectionHostInfo); + if (identifiedHostInfo) { + connectionHostInfo = identifiedHostInfo; + await this.pluginService.setRoutedHostInfo(connectionHostInfo); + } } - await this.tracker.populateOpenedConnectionQueue(hostInfo, targetClient); + const host = this.tracker.populateOpenedConnectionQueue(connectionHostInfo, targetClient); + this.pluginService.setTrackedConnectionHost(host); } return targetClient; } override async execute(methodName: string, methodFunc: () => Promise, methodArgs: any[]): Promise { + const currentHostInfo = this.pluginService.getCurrentHostInfo(); this.rememberWriter(); + const isClosing = AuroraConnectionTrackerPlugin.CLOSING_METHODS.has(methodName); + try { + if (!isClosing) { + let needRefreshHostList = false; + const localRefreshEndTimeNs = AuroraConnectionTrackerPlugin.hostListRefreshEndTimeNs; + if (localRefreshEndTimeNs > 0n) { + if (localRefreshEndTimeNs > process.hrtime.bigint()) { + needRefreshHostList = true; + } else { + AuroraConnectionTrackerPlugin.hostListRefreshEndTimeNs = 0n; + } + } + if (this.needUpdateCurrentWriter || needRefreshHostList) { + await this.checkWriterChanged(needRefreshHostList); + } + } + const result = await methodFunc(); - if (this.needUpdateCurrentWriter) { - await this.checkWriterChanged(); + + if (isClosing) { + const host = this.pluginService.getTrackedConnectionHost(); + if (host) { + this.tracker.removeConnectionTracking(host); + this.pluginService.setTrackedConnectionHost(null); + } else if (currentHostInfo) { + this.tracker.removeConnectionTrackingByHost(currentHostInfo, this.pluginService.getCurrentClient()?.targetClient); + } } return result; } catch (error) { if (error instanceof FailoverError) { - await this.checkWriterChanged(); + AuroraConnectionTrackerPlugin.hostListRefreshEndTimeNs = + process.hrtime.bigint() + AuroraConnectionTrackerPlugin.TOPOLOGY_CHANGES_EXPECTED_TIME_NS; + // This call may effectively close/abort the current connection. + await this.checkWriterChanged(true); } throw error; } } - private async checkWriterChanged(): Promise { + private async checkWriterChanged(needRefreshHostList: boolean): Promise { + if (needRefreshHostList) { + try { + await this.pluginService.refreshHostList(); + } catch (error) { + // Ignore: continue with whatever topology is currently available. + } + } + const hostInfoAfterFailover = this.getWriter(this.pluginService.getAllHosts()); + if (hostInfoAfterFailover === null) { + return; + } + if (this.currentWriter === null) { this.currentWriter = hostInfoAfterFailover; this.needUpdateCurrentWriter = false; - } else if (!this.currentWriter.equals(hostInfoAfterFailover!)) { - // writer changed + } else if (this.currentWriter.hostAndPort !== hostInfoAfterFailover.hostAndPort) { + // The writer changed. await this.tracker.invalidateAllConnections(this.currentWriter); this.tracker.logOpenedConnections(); this.currentWriter = hostInfoAfterFailover; this.needUpdateCurrentWriter = false; + AuroraConnectionTrackerPlugin.hostListRefreshEndTimeNs = 0n; } } diff --git a/common/lib/plugins/connection_tracker/opened_connection_tracker.ts b/common/lib/plugins/connection_tracker/opened_connection_tracker.ts index 6c8b6a58..6efa0839 100644 --- a/common/lib/plugins/connection_tracker/opened_connection_tracker.ts +++ b/common/lib/plugins/connection_tracker/opened_connection_tracker.ts @@ -21,9 +21,10 @@ 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"; export class OpenedConnectionTracker { - static readonly openedConnections: Map>> = new Map>>(); + static readonly openedConnections: Map = new Map(); readonly pluginService: PluginService; private static readonly rdsUtils = new RdsUtils(); @@ -31,101 +32,113 @@ export class OpenedConnectionTracker { this.pluginService = pluginService; } - async populateOpenedConnectionQueue(hostInfo: HostInfo, client: ClientWrapper): Promise { - const aliases = hostInfo.aliases; + populateOpenedConnectionQueue(hostInfo: HostInfo, client: ClientWrapper): TrackedConnectionListHost | null { + if (!hostInfo || !client) { + return null; + } - // Check if the connection was established using an instance endpoint + // Check if the connection was established using an instance endpoint. if (OpenedConnectionTracker.rdsUtils.isRdsInstance(hostInfo.host)) { - this.trackConnection(hostInfo.hostAndPort, client); - return; + const host = this.trackConnection(hostInfo.hostAndPort, client); + this.logOpenedConnections(); + return host; } - const instanceEndpoint = [...aliases] - .filter((x) => OpenedConnectionTracker.rdsUtils.isRdsInstance(OpenedConnectionTracker.rdsUtils.removePort(x))) - .reduce((max, s) => (s > max ? s : max), ""); + // It might be a custom domain name. Let's track by hostId and custom domain name. + let lastHost: TrackedConnectionListHost | null = null; + if (hostInfo.hostId) { + lastHost = this.trackConnection(hostInfo.hostId, client); + } + if (hostInfo.hostAndPort) { + lastHost = this.trackConnection(hostInfo.hostAndPort, client); + } + this.logOpenedConnections(); + return lastHost; + } - if (!instanceEndpoint) { - logger.debug(Messages.get("OpenedConnectionTracker.unableToPopulateOpenedConnectionQueue", hostInfo.host)); + async invalidateAllConnections(hostInfo: HostInfo): Promise { + if (!hostInfo) { return; } + await this.invalidateAllConnectionsMultipleHosts(hostInfo.hostAndPort, hostInfo.host, hostInfo.hostId); + } - this.trackConnection(instanceEndpoint, client); + async invalidateAllConnectionsMultipleHosts(...keys: string[]): Promise { + for (const key of keys) { + if (!key) { + continue; + } + try { + const connectionList = OpenedConnectionTracker.openedConnections.get(key); + this.logConnectionList(key, connectionList); + await this.invalidateConnections(connectionList); + } catch (error) { + // Ignore and continue with the remaining keys. + } + } } - async invalidateAllConnections(hostInfo: HostInfo): Promise { - await this.invalidateAllConnectionsMultipleHosts(hostInfo.asAlias); - await this.invalidateAllConnectionsMultipleHosts(...Array.from(hostInfo.aliases)); + removeConnectionTracking(host: TrackedConnectionListHost | null): void { + host?.remove(); } - async invalidateAllConnectionsMultipleHosts(...hosts: string[]): Promise { - try { - const instanceEndpoint = hosts - .filter((x) => OpenedConnectionTracker.rdsUtils.isRdsInstance(OpenedConnectionTracker.rdsUtils.removePort(x))) - .at(0); - if (!instanceEndpoint) { - return; - } - const connectionQueue = OpenedConnectionTracker.openedConnections.get(instanceEndpoint); - this.logConnectionQueue(instanceEndpoint, connectionQueue!); - await this.invalidateConnections(connectionQueue!); - } catch (error) { - // ignore + removeConnectionTrackingByHost(hostInfo: HostInfo, client: ClientWrapper | undefined | null): void { + const hostAndPort = OpenedConnectionTracker.rdsUtils.isRdsInstance(hostInfo.host) ? hostInfo.hostAndPort : null; + if (!hostAndPort) { + return; + } + + const connectionList = OpenedConnectionTracker.openedConnections.get(hostAndPort); + if (connectionList) { + connectionList.removeIf((ref) => { + const conn = ref.deref(); + return !conn || conn === client; + }); } } - private trackConnection(instanceEndpoint: string, client: ClientWrapper): void { - const connectionQueue = MapUtils.computeIfAbsent( - OpenedConnectionTracker.openedConnections, - instanceEndpoint, - (k) => new Array>() - ); - connectionQueue!.push(new WeakRef(client)); - this.logOpenedConnections(); + private trackConnection(instanceEndpoint: string, client: ClientWrapper): TrackedConnectionListHost { + const connectionList = MapUtils.computeIfAbsent(OpenedConnectionTracker.openedConnections, instanceEndpoint, (_) => new TrackedConnectionList()); + return connectionList!.add(client); } - private async invalidateConnections(connectionQueue: Array>): Promise { - let clientRef: WeakRef | undefined; - while ((clientRef = connectionQueue?.shift()) != null) { - const client = clientRef?.deref() ?? null; - if (!client) { - continue; - } + private async invalidateConnections(connectionList: TrackedConnectionList | undefined): Promise { + if (!connectionList || connectionList.isEmpty()) { + return; + } + + const connections = connectionList.drainAll(); + for (const client of connections) { await this.pluginService.abortTargetClient(client); } } logOpenedConnections(): void { - let str = ""; - const hostList = []; - - for (const queue of OpenedConnectionTracker.openedConnections.values()) { - if (queue.length !== 0) { - for (const connRef of queue) { - const conn = connRef?.deref() ?? null; - if (conn) { - hostList.push(`${conn.id} - ${conn.hostInfo.toString()}`); - } - } - str = hostList.join("\n\t"); + const hostList: string[] = []; + for (const connectionList of OpenedConnectionTracker.openedConnections.values()) { + for (const conn of connectionList.getConnections()) { + hostList.push(`${conn.id} - ${conn.hostInfo.toString()}`); } } - logger.debug(`Opened Connections Tracked: \n\t${str}`); + logger.debug(`Opened Connections Tracked: \n\t${hostList.join("\n\t")}`); } - private logConnectionQueue(host: string, queue: Array>): void { - if (!queue || queue.length === 0) { + private logConnectionList(host: string, connectionList: TrackedConnectionList | undefined): void { + if (!connectionList || connectionList.isEmpty()) { return; } - logger.debug(Messages.get("OpenedConnectionTracker.invalidatingConnections", `${host}\n[${queue.map((x) => x.deref()!.hostInfo).join()}\n]`)); + const connections = connectionList.getConnections().map((conn) => conn.hostInfo); + logger.debug(Messages.get("OpenedConnectionTracker.invalidatingConnections", `${host}\n[${connections.join()}\n]`)); } pruneNullConnections(): void { - for (const [key, queue] of OpenedConnectionTracker.openedConnections) { - OpenedConnectionTracker.openedConnections.set( - key, - queue.filter((connWeakRef: WeakRef) => connWeakRef?.deref() ?? null) - ); + for (const connectionList of OpenedConnectionTracker.openedConnections.values()) { + connectionList.removeIf((ref) => !ref.deref()); } } + + static clearCache(): void { + OpenedConnectionTracker.openedConnections.clear(); + } } diff --git a/common/lib/plugins/default_plugin.ts b/common/lib/plugins/default_plugin.ts index 2c3b45d8..ecd5ae0a 100644 --- a/common/lib/plugins/default_plugin.ts +++ b/common/lib/plugins/default_plugin.ts @@ -84,7 +84,7 @@ export class DefaultPlugin extends AbstractConnectionPlugin { ); const result: ConnectionInfo = await telemetryContext.start(async () => await connProvider.connect(hostInfo, this.pluginService, props)); - this.pluginService.setAvailability(hostInfo.allAliases, HostAvailability.AVAILABLE); + this.pluginService.setAvailability(hostInfo, HostAvailability.AVAILABLE); this.pluginService.setIsPooledClient(result.isPooled); await this.pluginService.updateDialect(result.client); return result.client; diff --git a/common/lib/plugins/efm/host_monitoring_connection_plugin.ts b/common/lib/plugins/efm/host_monitoring_connection_plugin.ts index 753e2f69..59ad986f 100644 --- a/common/lib/plugins/efm/host_monitoring_connection_plugin.ts +++ b/common/lib/plugins/efm/host_monitoring_connection_plugin.ts @@ -59,10 +59,13 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp ): Promise { const targetClient = await connectFunc(); if (targetClient != null) { - const type: RdsUrlType = this.rdsUtils.identifyRdsType(hostInfo.host); + const connectionHostInfo: HostInfo = this.pluginService.getRoutedHostInfo() ?? hostInfo; + const type: RdsUrlType = this.rdsUtils.identifyRdsType(connectionHostInfo.host); if (type.isRdsCluster) { - hostInfo.resetAliases(); - await this.pluginService.fillAliases(targetClient, hostInfo); + const identifiedHostInfo: HostInfo | null = await this.pluginService.identifyConnection(targetClient, connectionHostInfo); + if (identifiedHostInfo) { + this.pluginService.setRoutedHostInfo(identifiedHostInfo); + } } } return targetClient; @@ -110,10 +113,6 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp if (monitorContext.isHostUnhealthy) { const monitoringHostInfo = await this.getMonitoringHostInfo(); - if (monitoringHostInfo) { - this.pluginService.setAvailability(monitoringHostInfo.allAliases, HostAvailability.NOT_AVAILABLE); - } - const targetClient = this.pluginService.getCurrentClient().targetClient; let isClientValid = false; if (targetClient) { @@ -125,7 +124,9 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp await this.pluginService.abortTargetClient(targetClient); } // eslint-disable-next-line no-unsafe-finally - throw new UnavailableHostError(Messages.get("HostMonitoringConnectionPlugin.unavailableHost", monitoringHostInfo.host)); + throw new UnavailableHostError( + Messages.get("HostMonitoringConnectionPlugin.unavailableHost", monitoringHostInfo?.host ?? "Unknown host") + ); } } } @@ -149,7 +150,7 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp if (this.monitoringHostInfo) { return this.monitoringHostInfo; } - this.monitoringHostInfo = this.pluginService.getCurrentHostInfo(); + this.monitoringHostInfo = this.pluginService.getRoutedHostInfo() ?? this.pluginService.getCurrentHostInfo(); if (this.monitoringHostInfo == null) { this.throwUnableToIdentifyConnection(null); } @@ -158,12 +159,16 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp try { if (rdsUrlType.isRdsCluster) { logger.debug(Messages.get("HostMonitoringConnectionPlugin.identifyClusterConnection")); - this.monitoringHostInfo = await this.pluginService.identifyConnection(this.pluginService.getCurrentClient().targetClient!); + this.monitoringHostInfo = await this.pluginService.identifyConnection( + this.pluginService.getCurrentClient().targetClient!, + this.pluginService.getCurrentHostInfo() + ); if (this.monitoringHostInfo == null) { const host: HostInfo | null = this.pluginService.getCurrentHostInfo(); this.throwUnableToIdentifyConnection(host); } - await this.pluginService.fillAliases(this.pluginService.getCurrentClient().targetClient!, this.monitoringHostInfo); + // Update identified HostInfo for the current connection + await this.pluginService.setCurrentClient(this.pluginService.getCurrentClient().targetClient!, this.monitoringHostInfo); } } catch (error: any) { if (!(error instanceof AwsWrapperError)) { diff --git a/common/lib/plugins/failover/failover_plugin.ts b/common/lib/plugins/failover/failover_plugin.ts index ceefae36..73014d2d 100644 --- a/common/lib/plugins/failover/failover_plugin.ts +++ b/common/lib/plugins/failover/failover_plugin.ts @@ -184,12 +184,6 @@ export class FailoverPlugin extends AbstractConnectionPlugin { if (this.isHostStillValid(url, changes)) { return Promise.resolve(); } - - for (const alias of currentHost.allAliases) { - if (this.isHostStillValid(alias + "/", changes)) { - return Promise.resolve(); - } - } } logger.info(Messages.get("Failover.invalidHost", currentHost?.host ?? "empty host")); return Promise.resolve(); @@ -279,7 +273,7 @@ export class FailoverPlugin extends AbstractConnectionPlugin { await this.invalidateCurrentClient(); const currentHostInfo = this.pluginService.getCurrentHostInfo(); if (currentHostInfo !== null) { - this.pluginService.setAvailability(currentHostInfo.allAliases ?? new Set(), HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(currentHostInfo, HostAvailability.NOT_AVAILABLE); } this._lastError = e; @@ -291,7 +285,7 @@ export class FailoverPlugin extends AbstractConnectionPlugin { } async failover(failedHost: HostInfo) { - this.pluginService.setAvailability(failedHost.allAliases, HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(failedHost, HostAvailability.NOT_AVAILABLE); if (this.failoverMode === FailoverMode.STRICT_WRITER) { await this.failoverWriter(); @@ -308,8 +302,6 @@ export class FailoverPlugin extends AbstractConnectionPlugin { const telemetryContext = telemetryFactory.openTelemetryContext(FailoverPlugin.TELEMETRY_READER_FAILOVER, TelemetryTraceLevel.NESTED); this.failoverReaderTriggeredCounter.inc(); - const oldAliases = this.pluginService.getCurrentHostInfo()?.allAliases ?? new Set(); - let failedHost = null; if (failedHostInfo && failedHostInfo.getRawAvailability() === HostAvailability.AVAILABLE) { failedHost = failedHostInfo; @@ -334,7 +326,6 @@ export class FailoverPlugin extends AbstractConnectionPlugin { await this.pluginService.abortCurrentClient(); await this.pluginService.setCurrentClient(result.client, result.newHost); - this.pluginService.getCurrentHostInfo()?.removeAlias(Array.from(oldAliases)); await this.updateTopology(true); this.failoverWriterSuccessCounter.inc(); } catch (error: any) { diff --git a/common/lib/plugins/failover/reader_failover_handler.ts b/common/lib/plugins/failover/reader_failover_handler.ts index 21c82b1e..fddf12a6 100644 --- a/common/lib/plugins/failover/reader_failover_handler.ts +++ b/common/lib/plugins/failover/reader_failover_handler.ts @@ -141,7 +141,7 @@ export class ClusterAwareReaderFailoverHandler implements ReaderFailoverHandler async failoverInternal(hosts: HostInfo[], currentHost: HostInfo | null): Promise { if (currentHost) { - this.pluginService.setAvailability(currentHost.allAliases, HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(currentHost, HostAvailability.NOT_AVAILABLE); } const hostsByPriority = this.getHostsByPriority(hosts); return this.getConnectionFromHostGroup(hostsByPriority); @@ -336,7 +336,7 @@ class ConnectionAttemptTask { try { this.targetClient = await this.pluginService.forceConnect(this.newHost, copy); - this.pluginService.setAvailability(this.newHost.allAliases, HostAvailability.AVAILABLE); + this.pluginService.setAvailability(this.newHost, HostAvailability.AVAILABLE); logger.info(Messages.get("ClusterAwareReaderFailoverHandler.successfulReaderConnection", this.newHost.host)); if (this.taskHandler.getSelectedConnectionAttemptTask(this.failoverTaskId) === -1) { this.taskHandler.setSelectedConnectionAttemptTask(this.failoverTaskId, this.taskId); @@ -344,7 +344,7 @@ class ConnectionAttemptTask { } throw new AwsWrapperError(Messages.get("ClusterAwareReaderFailoverHandler.selectedTaskChosen", this.newHost.host)); } catch (error) { - this.pluginService.setAvailability(this.newHost.allAliases, HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(this.newHost, HostAvailability.NOT_AVAILABLE); throw error; } finally { await this.performFinalCleanup(); diff --git a/common/lib/plugins/failover/writer_failover_handler.ts b/common/lib/plugins/failover/writer_failover_handler.ts index e14a73fd..dbc2320d 100644 --- a/common/lib/plugins/failover/writer_failover_handler.ts +++ b/common/lib/plugins/failover/writer_failover_handler.ts @@ -37,12 +37,14 @@ export interface WriterFailoverHandler { function isCurrentHostWriter(topology: HostInfo[], originalWriterHost: HostInfo): boolean { const latestWriter = getWriter(topology); - const latestWriterAllAliases = latestWriter?.allAliases; - const currentAliases = originalWriterHost.allAliases; - if (currentAliases && latestWriterAllAliases) { - return [...currentAliases].filter((alias) => latestWriterAllAliases.has(alias)).length > 0; + if (!latestWriter || !originalWriterHost) { + return false; } - return false; + + return ( + (!!latestWriter.hostId && latestWriter.hostId === originalWriterHost.hostId) || + (!!latestWriter.hostAndPort && latestWriter.hostAndPort.toLowerCase() === originalWriterHost.hostAndPort.toLowerCase()) + ); } export class ClusterAwareWriterFailoverHandler implements WriterFailoverHandler { @@ -273,7 +275,7 @@ class ReconnectToWriterHandlerTask { } success = isCurrentHostWriter(latestTopology, this.originalWriterHost); - this.pluginService.setAvailability(this.originalWriterHost.allAliases, HostAvailability.AVAILABLE); + this.pluginService.setAvailability(this.originalWriterHost, HostAvailability.AVAILABLE); return new WriterFailoverResult( success, false, @@ -454,13 +456,13 @@ class WaitForNewWriterHandlerTask { let targetClient = null; try { targetClient = await this.pluginService.forceConnect(writerCandidate, props); - this.pluginService.setAvailability(writerCandidate.allAliases, HostAvailability.AVAILABLE); + this.pluginService.setAvailability(writerCandidate, HostAvailability.AVAILABLE); await this.callCloseClient(this.currentReaderTargetClient); await this.callCloseClient(this.currentClient); this.currentClient = targetClient; return true; } catch (error) { - this.pluginService.setAvailability(writerCandidate.allAliases, HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(writerCandidate, HostAvailability.NOT_AVAILABLE); await this.pluginService.abortTargetClient(targetClient); return false; } diff --git a/common/lib/plugins/failover2/failover2_plugin.ts b/common/lib/plugins/failover2/failover2_plugin.ts index f9e04fe6..85288c65 100644 --- a/common/lib/plugins/failover2/failover2_plugin.ts +++ b/common/lib/plugins/failover2/failover2_plugin.ts @@ -150,7 +150,7 @@ export class Failover2Plugin extends AbstractConnectionPlugin { throw error; } - this.pluginService.setAvailability(hostInfo.allAliases, HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(hostInfo, HostAvailability.NOT_AVAILABLE); try { // Unable to directly connect, attempt failover. @@ -209,7 +209,7 @@ export class Failover2Plugin extends AbstractConnectionPlugin { await this.invalidateCurrentClient(); const currentHostInfo: HostInfo = this.pluginService.getCurrentHostInfo(); if (currentHostInfo !== null) { - this.pluginService.setAvailability(currentHostInfo.allAliases ?? new Set(), HostAvailability.NOT_AVAILABLE); + this.pluginService.setAvailability(currentHostInfo, HostAvailability.NOT_AVAILABLE); } await this.failover(); this._lastError = error; @@ -247,7 +247,6 @@ export class Failover2Plugin extends AbstractConnectionPlugin { const telemetryContext = telemetryFactory.openTelemetryContext(Failover2Plugin.TELEMETRY_READER_FAILOVER, TelemetryTraceLevel.NESTED); this.failoverReaderTriggeredCounter.inc(); - const oldAliases = this.pluginService.getCurrentHostInfo()?.allAliases ?? new Set(); const failoverEndTimeMs = Date.now() + this.failoverTimeoutSettingMs; try { @@ -265,7 +264,6 @@ export class Failover2Plugin extends AbstractConnectionPlugin { this.failoverReaderSuccessCounter.inc(); await this.pluginService.abortCurrentClient(); await this.pluginService.setCurrentClient(result.client, result.newHost); - this.pluginService.getCurrentHostInfo()?.removeAlias(Array.from(oldAliases)); await this.pluginService.forceRefreshHostList(); } catch (error) { this.failoverReaderFailedCounter.inc(); diff --git a/common/lib/plugins/limitless/limitless_connection_context.ts b/common/lib/plugins/limitless/limitless_connection_context.ts index ccc244c9..5891c2da 100644 --- a/common/lib/plugins/limitless/limitless_connection_context.ts +++ b/common/lib/plugins/limitless/limitless_connection_context.ts @@ -17,6 +17,7 @@ import { HostInfo } from "../../host_info"; import { ClientWrapper } from "../../client_wrapper"; import { ConnectionPlugin } from "../../connection_plugin"; +import Map from "@arrows/array/src/map"; export class LimitlessConnectionContext { private readonly hostInfo: HostInfo; @@ -25,6 +26,7 @@ export class LimitlessConnectionContext { private readonly connectFunc: () => Promise; private routers: HostInfo[] | null; private plugin: ConnectionPlugin; + private connectionHostInfo: HostInfo | null; constructor( hostInfo: HostInfo, @@ -40,6 +42,7 @@ export class LimitlessConnectionContext { this.connectFunc = connectFunc; this.routers = routers; this.plugin = plugin; + this.connectionHostInfo = null; } public getHostInfo(): HostInfo { @@ -58,6 +61,14 @@ export class LimitlessConnectionContext { return this.connectFunc; } + getConnectionHostInfo(): HostInfo | null { + return this.connectionHostInfo; + } + + setConnectionHostInfo(value: HostInfo | null) { + this.connectionHostInfo = value; + } + public getRouters(): HostInfo[] | null { return this.routers; } diff --git a/common/lib/plugins/limitless/limitless_connection_plugin.ts b/common/lib/plugins/limitless/limitless_connection_plugin.ts index 775c6cbd..768674ae 100644 --- a/common/lib/plugins/limitless/limitless_connection_plugin.ts +++ b/common/lib/plugins/limitless/limitless_connection_plugin.ts @@ -65,6 +65,9 @@ export class LimitlessConnectionPlugin extends AbstractConnectionPlugin { await this.limitlessRouterService.establishConnection(context); if (context.getConnection() != null) { + if (context.getConnectionHostInfo() != null) { + this.pluginService.setRoutedHostInfo(context.getConnectionHostInfo()); + } return context.getConnection(); } throw new AwsWrapperError(Messages.get("LimitlessConnectionPlugin.failedToConnectToHost", hostInfo.host)); diff --git a/common/lib/plugins/limitless/limitless_router_service.ts b/common/lib/plugins/limitless/limitless_router_service.ts index b33569c2..46850838 100644 --- a/common/lib/plugins/limitless/limitless_router_service.ts +++ b/common/lib/plugins/limitless/limitless_router_service.ts @@ -118,6 +118,7 @@ export class LimitlessRouterServiceImpl implements LimitlessRouterService { try { context.setConnection(await this.pluginService.connect(selectedHostSpec, context.getProperties(), context.getPlugin())); + context.setConnectionHostInfo(selectedHostSpec); } catch (e) { logger.debug(Messages.get("LimitlessRouterServiceImpl.failedToConnectToHost", selectedHostSpec.host)); selectedHostSpec.setAvailability(HostAvailability.NOT_AVAILABLE); @@ -161,12 +162,12 @@ export class LimitlessRouterServiceImpl implements LimitlessRouterService { } } - let selectedHostSpec: HostInfo | undefined = undefined; + let selectedHostInfo: HostInfo | undefined = undefined; try { // Select healthiest router for best chance of connection over load-balancing with round-robin - selectedHostSpec = this.pluginService.getHostInfoByStrategy(HostRole.WRITER, HighestWeightHostSelector.STRATEGY_NAME, context.getRouters()); - logger.debug(Messages.get("LimitlessRouterServiceImpl.selectedHostForRetry", selectedHostSpec ? selectedHostSpec.host : "undefined")); - if (!selectedHostSpec) { + selectedHostInfo = this.pluginService.getHostInfoByStrategy(HostRole.WRITER, HighestWeightHostSelector.STRATEGY_NAME, context.getRouters()); + logger.debug(Messages.get("LimitlessRouterServiceImpl.selectedHostForRetry", selectedHostInfo ? selectedHostInfo.host : "undefined")); + if (!selectedHostInfo) { continue; } } catch (e) { @@ -179,13 +180,14 @@ export class LimitlessRouterServiceImpl implements LimitlessRouterService { } try { - context.setConnection(await this.pluginService.connect(selectedHostSpec, context.getProperties(), context.getPlugin())); + context.setConnection(await this.pluginService.connect(selectedHostInfo, context.getProperties(), context.getPlugin())); if (context.getConnection()) { + context.setConnectionHostInfo(selectedHostInfo); return; } } catch (error) { - selectedHostSpec.setAvailability(HostAvailability.NOT_AVAILABLE); - logger.debug(Messages.get("LimitlessRouterServiceImpl.failedToConnectToHost", selectedHostSpec.host)); + selectedHostInfo.setAvailability(HostAvailability.NOT_AVAILABLE); + logger.debug(Messages.get("LimitlessRouterServiceImpl.failedToConnectToHost", selectedHostInfo.host)); } } throw new AwsWrapperError(Messages.get("LimitlessRouterServiceImpl.maxRetriesExceeded")); diff --git a/common/lib/utils/full_services_container.ts b/common/lib/utils/full_services_container.ts index 59902f88..c1be9827 100644 --- a/common/lib/utils/full_services_container.ts +++ b/common/lib/utils/full_services_container.ts @@ -23,6 +23,7 @@ import { StorageService } from "./storage/storage_service"; import { MonitorService } from "./monitoring/monitor_service"; import { EventPublisher } from "./events/event"; import { ImportantEventService } from "./important_event_service"; +import { HostIdCacheService } from "./host_id_cache_service"; /** * Container for services used throughout the wrapper. @@ -37,6 +38,7 @@ export interface FullServicesContainer { hostListProviderService: HostListProviderService; pluginService: PluginService; importantEventService: ImportantEventService; + hostIdCacheService: HostIdCacheService; } export class FullServicesContainerImpl implements FullServicesContainer { @@ -49,6 +51,7 @@ export class FullServicesContainerImpl implements FullServicesContainer { hostListProviderService!: HostListProviderService; pluginService!: PluginService; importantEventService: ImportantEventService; + hostIdCacheService: HostIdCacheService; constructor( storageService: StorageService, diff --git a/common/lib/utils/host_id_cache_service.ts b/common/lib/utils/host_id_cache_service.ts new file mode 100644 index 00000000..9ac7e32c --- /dev/null +++ b/common/lib/utils/host_id_cache_service.ts @@ -0,0 +1,137 @@ +/* + 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 { ClientWrapper } from "../client_wrapper"; +import { HostInfo } from "../host_info"; +import { PluginService } from "../plugin_service"; +import { DatabaseDialect } from "../database_dialect/database_dialect"; +import { TopologyAwareDatabaseDialect } from "../database_dialect/topology_aware_database_dialect"; +import { RdsUtils } from "./rds_utils"; +import { RdsUrlType } from "./rds_url_type"; +import { AwsWrapperError } from "./errors"; +import { Messages } from "./messages"; + +/** + * Tuple of (instanceId, instanceName) identifying a connection's underlying instance, as + * returned by a topology-aware dialect. Either value may be null when identification fails. + */ +export type InstanceIdAndName = [instanceId: string | null, instanceName: string | null]; + +/** + * Identifies the underlying instance a connection is established to. + * + * For static endpoints (IP addresses or custom domain names) the identification result is cached, + * keyed by the connection's host name, to avoid repeatedly querying the database. + */ +export interface HostIdCacheService { + /** + * Identify the connected host, using the cache where possible. + * + * @param targetClient the connection to be identified. + * @param connectionHostInfo the {@link HostInfo} the connection was established with. + * @param pluginService the plugin service instance. + * @returns the identified {@link HostInfo}, or null if it cannot be determined. + */ + identifyConnection(targetClient: ClientWrapper, connectionHostInfo: HostInfo, pluginService: PluginService): Promise; +} + +function isTopologyAwareDialect(dialect: DatabaseDialect): dialect is DatabaseDialect & TopologyAwareDatabaseDialect { + return typeof (dialect as Partial).getInstanceId === "function"; +} + +export class HostIdCacheServiceImpl implements HostIdCacheService { + static readonly PROP_ENABLED = "AWS_NODEJS_HOST_CACHE_ENABLED"; + static readonly PROP_REGEXP = "AWS_NODEJS_HOST_CACHE_REGEXP"; + + private static readonly cache = new Map(); + private static readonly isEnabled = (process.env[HostIdCacheServiceImpl.PROP_ENABLED] ?? "true").toLowerCase() === "true"; + private static readonly hostRegexp = new RegExp(process.env[HostIdCacheServiceImpl.PROP_REGEXP] ?? ".*"); + private static readonly rdsHelper = new RdsUtils(); + + async identifyConnection(targetClient: ClientWrapper, connectionHostInfo: HostInfo, pluginService: PluginService): Promise { + if (!targetClient || !connectionHostInfo || !pluginService) { + return null; + } + + const urlType: RdsUrlType = HostIdCacheServiceImpl.rdsHelper.identifyRdsType(connectionHostInfo.host); + switch (urlType) { + case RdsUrlType.RDS_INSTANCE: + return connectionHostInfo; + case RdsUrlType.IP_ADDRESS: + case RdsUrlType.OTHER: + // It might be a custom domain name. Cache the identification keyed by host name when allowed. + if (HostIdCacheServiceImpl.isEnabled && HostIdCacheServiceImpl.hostRegexp.test(connectionHostInfo.host)) { + return this.getCachedHostInfo(targetClient, connectionHostInfo, pluginService); + } + return pluginService.identifyConnection(targetClient); + default: + // Other hosts are dynamic and may change at any time, so they can't be cached. + return pluginService.identifyConnection(targetClient); + } + } + + protected async getCachedHostInfo( + targetClient: ClientWrapper, + connectionHostInfo: HostInfo, + pluginService: PluginService + ): Promise { + const host = connectionHostInfo.host; + + let instanceIdAndName = HostIdCacheServiceImpl.cache.get(host); + if (!instanceIdAndName) { + instanceIdAndName = await this.queryInstanceIdAndName(targetClient, pluginService); + HostIdCacheServiceImpl.cache.set(host, instanceIdAndName); + } + + const [instanceId, instanceName] = instanceIdAndName; + if (!instanceId && !instanceName) { + // We've already tried to identify the connection, but got nothing. + return null; + } + + let topology = pluginService.getAllHosts(); + if (!topology || topology.length === 0) { + const provider = pluginService.getHostListProvider(); + topology = provider ? await provider.forceRefresh() : null; + if (!topology || topology.length === 0) { + return null; + } + } + + return topology.find((candidate) => instanceId === candidate.hostId || instanceName === candidate.host) ?? null; + } + + private async queryInstanceIdAndName(targetClient: ClientWrapper, pluginService: PluginService): Promise { + const dialect = pluginService.getDialect(); + if (!isTopologyAwareDialect(dialect)) { + return [null, null]; + } + + try { + const [instanceId, instanceName] = await dialect.getInstanceId(targetClient); + return [instanceId ?? null, instanceName ?? null]; + } catch (error: any) { + throw new AwsWrapperError(Messages.get("HostIdCacheService.errorIdentifyConnection"), error); + } + } + + /** + * Clears the static host identification cache. Intended for test cleanup. + */ + static clearCache(): void { + HostIdCacheServiceImpl.cache.clear(); + } +} diff --git a/common/lib/utils/messages.ts b/common/lib/utils/messages.ts index 32d9b9df..94e619b9 100644 --- a/common/lib/utils/messages.ts +++ b/common/lib/utils/messages.ts @@ -59,6 +59,7 @@ const MESSAGES: Record = { "RdsHostListProvider.noClusterId": "No clusterId found. Please ensure clusterId parameter is set to a non-empty string.", "ConnectionStringHostListProvider.parsedListEmpty": "Can't parse connection string: '%s'.", "ConnectionStringHostListProvider.errorIdentifyConnection": "An error occurred while obtaining the connection's host ID.", + "HostIdCacheService.errorIdentifyConnection": "An error occurred while identifying the connection's host ID.", "ExecuteTimePlugin.executeTime": "Executed method '%s' in %s milliseconds.", "ConnectTimePlugin.connectTime": "Connected to '%s' in %s milliseconds.", "ClusterAwareWriterFailoverHandler.failoverCalledWithInvalidTopology": "Failover was called with an invalid (null or empty) topology.", diff --git a/common/lib/utils/service_utils.ts b/common/lib/utils/service_utils.ts index b7bda542..35c98b38 100644 --- a/common/lib/utils/service_utils.ts +++ b/common/lib/utils/service_utils.ts @@ -31,6 +31,7 @@ import { TelemetryFactory } from "./telemetry/telemetry_factory"; import { EventPublisher } from "./events/event"; import { PartialPluginService } from "../partial_plugin_service"; import { ConnectionUrlParser } from "./connection_url_parser"; +import { HostIdCacheServiceImpl } from "./host_id_cache_service"; export class ServiceUtils { private static readonly _instance: ServiceUtils = new ServiceUtils(); @@ -70,6 +71,7 @@ export class ServiceUtils { servicesContainer.pluginService = pluginService; servicesContainer.pluginManager = pluginManager; servicesContainer.hostListProviderService = pluginService; + servicesContainer.hostIdCacheService = new HostIdCacheServiceImpl(); return servicesContainer; } @@ -104,6 +106,7 @@ export class ServiceUtils { servicesContainer.pluginService = pluginService; servicesContainer.pluginManager = pluginManager; servicesContainer.hostListProviderService = pluginService; + servicesContainer.hostIdCacheService = new HostIdCacheServiceImpl(); return servicesContainer; } diff --git a/mysql/lib/client.ts b/mysql/lib/client.ts index 45b9bed7..2048848c 100644 --- a/mysql/lib/client.ts +++ b/mysql/lib/client.ts @@ -28,6 +28,7 @@ import { AwsWrapperError, ConnectionProvider, FailoverSuccessError, + HostInfo, InternalPooledConnectionProvider, TransactionIsolationLevel, UndefinedClientError, @@ -86,7 +87,9 @@ class BaseAwsMySQLClient extends AwsClient implements MySQLClient { // Ignore } } - await this.pluginService.setCurrentClient(result, result.hostInfo); + const connectedHostInfo: HostInfo | null = this.pluginService.getRoutedHostInfo() ?? this.pluginService.getInitialConnectionHostInfo(); + await this.pluginService.setCurrentClient(result, connectedHostInfo); + this.pluginService.setRoutedHostInfo(null); await this.internalPostConnect(); }); } diff --git a/pg/lib/client.ts b/pg/lib/client.ts index b8c55464..bb0b3d9f 100644 --- a/pg/lib/client.ts +++ b/pg/lib/client.ts @@ -215,7 +215,10 @@ class BaseAwsPgClient extends AwsClient implements PGClient { // Ignore } } - await this.pluginService.setCurrentClient(result, result.hostInfo); + + const connectedHostInfo: HostInfo | null = this.pluginService.getRoutedHostInfo() ?? this.pluginService.getInitialConnectionHostInfo(); + await this.pluginService.setCurrentClient(result, connectedHostInfo); + this.pluginService.setRoutedHostInfo(null); await this.internalPostConnect(); }); }