Skip to content

Commit 0f2e4d4

Browse files
karenc-bqkarenc-bq
committed
feat: gdb rw splitting plugin
1 parent 14c1bb3 commit 0f2e4d4

9 files changed

Lines changed: 184 additions & 12 deletions

‎common/lib/connection_plugin_chain_builder.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ import { HostMonitoring2PluginFactory } from "./plugins/efm2/host_monitoring2_pl
4545
import { BlueGreenPluginFactory } from "./plugins/bluegreen/blue_green_plugin_factory";
4646
import { GlobalDbFailoverPluginFactory } from "./plugins/gdb_failover/global_db_failover_plugin_factory";
4747
import { FullServicesContainer } from "./utils/full_services_container";
48+
import { GdbReadWriteSplittingPluginFactory } from "./plugins/read_write_splitting/gdb_read_write_splitting_plugin_factory";
4849

4950
/*
5051
Type alias used for plugin factory sorting. It holds a reference to a plugin
@@ -65,6 +66,7 @@ export class ConnectionPluginChainBuilder {
6566
["staleDns", { factory: StaleDnsPluginFactory, weight: 500 }],
6667
["bg", { factory: BlueGreenPluginFactory, weight: 550 }],
6768
["readWriteSplitting", { factory: ReadWriteSplittingPluginFactory, weight: 600 }],
69+
["gdbReadWriteSplitting", { factory: GdbReadWriteSplittingPluginFactory, weight: 610 }],
6870
["failover", { factory: FailoverPluginFactory, weight: 700 }],
6971
["failover2", { factory: Failover2PluginFactory, weight: 710 }],
7072
["gdbFailover", { factory: GlobalDbFailoverPluginFactory, weight: 720 }],
@@ -87,6 +89,7 @@ export class ConnectionPluginChainBuilder {
8789
[StaleDnsPluginFactory, 500],
8890
[BlueGreenPluginFactory, 550],
8991
[ReadWriteSplittingPluginFactory, 600],
92+
[GdbReadWriteSplittingPluginFactory, 610],
9093
[FailoverPluginFactory, 700],
9194
[Failover2PluginFactory, 710],
9295
[GlobalDbFailoverPluginFactory, 720],

‎common/lib/plugins/read_write_splitting/abstract_read_write_splitting_plugin.ts‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,11 +30,13 @@ import { FailoverError } from "../../utils/errors";
3030
import { WrapperProperties } from "../../wrapper_property";
3131
import { convertMsToNanos, getTimeInNanos, logAndThrowError } from "../../utils/utils";
3232
import { CacheItem } from "../../utils/cache_map";
33+
import { FullServicesContainer } from "../../utils/full_services_container";
3334

3435
export abstract class AbstractReadWriteSplittingPlugin extends AbstractConnectionPlugin implements CanReleaseResources {
3536
private static readonly subscribedMethods: Set<string> = new Set(["initHostProvider", "connect", "notifyConnectionChanged", "query"]);
3637

3738
protected _hostListProviderService: HostListProviderService | undefined;
39+
protected servicesContainer: FullServicesContainer;
3840
protected pluginService: PluginService;
3941
protected readonly _properties: Map<string, any>;
4042
protected readerHostInfo?: HostInfo = undefined;
@@ -48,9 +50,10 @@ export abstract class AbstractReadWriteSplittingPlugin extends AbstractConnectio
4850

4951
private _inReadWriteSplit = false;
5052

51-
protected constructor(pluginService: PluginService, properties: Map<string, any>) {
53+
protected constructor(serviceContainer: FullServicesContainer, properties: Map<string, any>) {
5254
super();
53-
this.pluginService = pluginService;
55+
this.servicesContainer = serviceContainer;
56+
this.pluginService = this.servicesContainer.getPluginService();
5457
this._properties = properties;
5558
this.readerSelectorStrategy = WrapperProperties.READER_HOST_SELECTOR_STRATEGY.get(properties);
5659
}

‎common/lib/plugins/read_write_splitting/gdb_read_write_splitting_plugin_factory.ts‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
/*
2+
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License").
5+
You may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
import { ConnectionPluginFactory } from "../../plugin_factory";
18+
import { PluginService } from "../../plugin_service";
19+
import { ConnectionPlugin } from "../../connection_plugin";
20+
import { AwsWrapperError } from "../../utils/errors";
21+
import { Messages } from "../../utils/messages";
22+
import { FullServicesContainer } from "../../utils/full_services_container";
23+
24+
export class GdbReadWriteSplittingPluginFactory extends ConnectionPluginFactory {
25+
private static gdbReadWriteSplittingPlugin: any;
26+
27+
async getInstance(servicesContainer: FullServicesContainer, properties: Map<string, any>): Promise<ConnectionPlugin> {
28+
try {
29+
if (!GdbReadWriteSplittingPluginFactory.gdbReadWriteSplittingPlugin) {
30+
GdbReadWriteSplittingPluginFactory.gdbReadWriteSplittingPlugin = await import("./gdb_read_writer_splitting_plugin");
31+
}
32+
return new GdbReadWriteSplittingPluginFactory.gdbReadWriteSplittingPlugin.GdbReadWriteSplittingPlugin(servicesContainer, properties);
33+
} catch (error: any) {
34+
throw new AwsWrapperError(Messages.get("ConnectionPluginChainBuilder.errorImportingPlugin", error.message, "gdbReadWriteSplittingPlugin"));
35+
}
36+
}
37+
}

‎common/lib/plugins/read_write_splitting/gdb_read_writer_splitting_plugin.ts‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
/*
2+
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License").
5+
You may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
import { ReadWriteSplittingPlugin } from "./read_write_splitting_plugin";
18+
import { PluginService } from "../../plugin_service";
19+
import { WrapperProperties } from "../../wrapper_property";
20+
import { HostInfo } from "../../host_info";
21+
import { RdsUtils } from "../../utils/rds_utils";
22+
import { ReadWriteSplittingError } from "../../utils/errors";
23+
import { Messages } from "../../utils/messages";
24+
import { logger } from "../../../logutils";
25+
import { ClientWrapper } from "../../client_wrapper";
26+
import { equalsIgnoreCase } from "../../utils/utils";
27+
import { FullServicesContainer } from "../../utils/full_services_container";
28+
29+
export class GdbReadWriteSplittingPlugin extends ReadWriteSplittingPlugin {
30+
protected readonly rdsUtils: RdsUtils = new RdsUtils();
31+
32+
protected readonly restrictWriterToHomeRegion: boolean;
33+
protected readonly restrictReaderToHomeRegion: boolean;
34+
35+
protected isInitialized: boolean = false;
36+
protected homeRegion: string;
37+
38+
constructor(serviceContainer: FullServicesContainer, properties: Map<string, any>) {
39+
super(serviceContainer, properties);
40+
this.restrictWriterToHomeRegion = WrapperProperties.GDB_RW_RESTRICT_WRITER_TO_HOME_REGION.get(properties);
41+
this.restrictReaderToHomeRegion = WrapperProperties.GDB_RW_RESTRICT_READER_TO_HOME_REGION.get(properties);
42+
}
43+
44+
protected initSettings(initHostInfo: HostInfo, properties: Map<string, any>): void {
45+
if (this.isInitialized) {
46+
return;
47+
}
48+
49+
this.isInitialized = true;
50+
51+
this.homeRegion = WrapperProperties.GDB_RW_HOME_REGION.get(properties);
52+
if (!this.homeRegion) {
53+
const rdsUrlType = this.rdsUtils.identifyRdsType(initHostInfo.host);
54+
if (rdsUrlType.hasRegion) {
55+
this.homeRegion = this.rdsUtils.getRdsRegion(initHostInfo.host);
56+
}
57+
}
58+
59+
if (!this.homeRegion) {
60+
throw new ReadWriteSplittingError(Messages.get("GdbReadWriteSplittingPlugin.missingHomeRegion", initHostInfo.host));
61+
}
62+
63+
logger.debug(Messages.get("GdbReadWriteSplittingPlugin.parameterValue", "gdbRwHomeRegion", this.homeRegion));
64+
}
65+
66+
override async connect(
67+
hostInfo: HostInfo,
68+
props: Map<string, any>,
69+
isInitialConnection: boolean,
70+
connectFunc: () => Promise<ClientWrapper>
71+
): Promise<ClientWrapper> {
72+
this.initSettings(hostInfo, props);
73+
return super.connect(hostInfo, props, isInitialConnection, connectFunc);
74+
}
75+
76+
override setWriterClient(writerTargetClient: ClientWrapper | undefined, writerHostInfo: HostInfo) {
77+
if (
78+
this.restrictWriterToHomeRegion &&
79+
this.writerHostInfo != null &&
80+
!equalsIgnoreCase(this.rdsUtils.getRdsRegion(this.writerHostInfo.host), this.homeRegion)
81+
) {
82+
throw new ReadWriteSplittingError(
83+
Messages.get("GdbReadWriteSplittingPlugin.cantConnectWriterOutOfHomeRegion", writerHostInfo.host, this.homeRegion)
84+
);
85+
}
86+
super.setWriterClient(writerTargetClient, writerHostInfo);
87+
}
88+
89+
protected getReaderHostCandidates(): HostInfo[] {
90+
if (this.restrictReaderToHomeRegion) {
91+
const hostsInRegion: HostInfo[] = this.pluginService
92+
.getHosts()
93+
.filter((x) => equalsIgnoreCase(this.rdsUtils.getRdsRegion(x.host), this.homeRegion));
94+
95+
if (hostsInRegion.length === 0) {
96+
throw new ReadWriteSplittingError(Messages.get("GdbReadWriteSplittingPlugin.noAvailableReadersInHomeRegion", this.homeRegion));
97+
}
98+
return hostsInRegion;
99+
}
100+
return super.getReaderHostCandidates();
101+
}
102+
}

‎common/lib/plugins/read_write_splitting/read_write_splitting_plugin.ts‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,7 @@
1414
limitations under the License.
1515
*/
1616

17-
import { AbstractConnectionPlugin } from "../../abstract_connection_plugin";
18-
import { HostInfo, FailoverError, HostRole } from "../../index";
17+
import { HostInfo, HostRole } from "../../index";
1918
import { PluginService } from "../../plugin_service";
2019
import { HostListProviderService } from "../../host_list_provider_service";
2120
import { Messages } from "../../utils/messages";
@@ -25,26 +24,27 @@ import { AbstractReadWriteSplittingPlugin } from "./abstract_read_write_splittin
2524
import { WrapperProperties } from "../../wrapper_property";
2625
import { logger } from "../../../logutils";
2726
import { CacheItem } from "../../utils/cache_map";
27+
import { FullServicesContainer } from "../../utils/full_services_container";
2828

2929
export class ReadWriteSplittingPlugin extends AbstractReadWriteSplittingPlugin {
3030
protected hosts: HostInfo[] = [];
3131

32-
constructor(pluginService: PluginService, properties: Map<string, any>);
32+
constructor(serviceContainer: FullServicesContainer, properties: Map<string, any>);
3333
constructor(
34-
pluginService: PluginService,
34+
serviceContainer: FullServicesContainer,
3535
properties: Map<string, any>,
3636
hostListProviderService: HostListProviderService,
3737
writerClient: ClientWrapper,
3838
readerClient: ClientWrapper
3939
);
4040
constructor(
41-
pluginService: PluginService,
41+
serviceContainer: FullServicesContainer,
4242
properties: Map<string, any>,
4343
hostListProviderService?: HostListProviderService,
4444
writerClient?: ClientWrapper,
4545
readerClient?: ClientWrapper
4646
) {
47-
super(pluginService, properties);
47+
super(serviceContainer, properties);
4848
this._hostListProviderService = hostListProviderService;
4949
this.writerTargetClient = writerClient;
5050
this.readerCacheItem = new CacheItem(readerClient, BigInt(0));

‎common/lib/plugins/read_write_splitting/read_write_splitting_plugin_factory.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
/*
22
Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3-
3+
44
Licensed under the Apache License, Version 2.0 (the "License").
55
You may not use this file except in compliance with the License.
66
You may obtain a copy of the License at
7-
7+
88
http://www.apache.org/licenses/LICENSE-2.0
9-
9+
1010
Unless required by applicable law or agreed to in writing, software
1111
distributed under the License is distributed on an "AS IS" BASIS,
1212
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

‎common/lib/utils/errors.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,8 @@ export class FailoverFailedError extends FailoverError {}
4848

4949
export class TransactionResolutionUnknownError extends FailoverError {}
5050

51+
export class ReadWriteSplittingError extends AwsWrapperError {}
52+
5153
export class LoginError extends AwsWrapperError {}
5254

5355
export class AwsTimeoutError extends AwsWrapperError {}

‎common/lib/utils/messages.ts‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -417,7 +417,14 @@ const MESSAGES: Record<string, string> = {
417417
"GlobalDbFailoverPlugin.currentFailoverMode": "Current Global DB failover mode: %s",
418418
"GlobalDbFailoverPlugin.failoverElapsed": "Global DB failover elapsed time: %s ms",
419419
"GlobalDbFailoverPlugin.candidateNull": "Candidate host is null for role: %s",
420-
"GlobalDbFailoverPlugin.unableToConnect": "Unable to establish a connection during Global DB failover."
420+
"GlobalDbFailoverPlugin.unableToConnect": "Unable to establish a connection during Global DB failover.",
421+
"GlobalDbFailoverPlugin.unableToConnect": "Unable to establish a connection during failover.",
422+
"GdbReadWriteSplittingPlugin.missingHomeRegion":
423+
"Unable to parse home region from endpoint '%s'. Please ensure you have set the 'gdbRwHomeRegion' connection parameter.",
424+
"GdbReadWriteSplittingPlugin.cantConnectWriterOutOfHomeRegion":
425+
"Writer connection to '%s' is not allowed since it is out of home region '%s'.",
426+
"GdbReadWriteSplittingPlugin.noAvailableReadersInHomeRegion": "No available reader nodes in home region '%s'.",
427+
"GdbReadWriteSplittingPlugin.parameterValue": "%s=%s"
421428
};
422429

423430
export class Messages {

‎common/lib/wrapper_property.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -564,6 +564,24 @@ export class WrapperProperties {
564564
["writer", "none"]
565565
);
566566

567+
static readonly GDB_RW_HOME_REGION = new WrapperProperty<string>(
568+
"gdbRwHomeRegion",
569+
"Specifies the home region for read/write splitting.",
570+
null
571+
);
572+
573+
static readonly GDB_RW_RESTRICT_WRITER_TO_HOME_REGION = new WrapperProperty<boolean>(
574+
"gdbRwRestrictWriterToHomeRegion",
575+
"Prevents connections to a writer node outside of the defined home region.",
576+
true
577+
);
578+
579+
static readonly GDB_RW_RESTRICT_READER_TO_HOME_REGION = new WrapperProperty<boolean>(
580+
"gdbRwRestrictReaderToHomeRegion",
581+
"Prevents connections to a reader node outside of the defined home region.",
582+
true
583+
);
584+
567585
private static readonly PREFIXES = [
568586
WrapperProperties.MONITORING_PROPERTY_PREFIX,
569587
ClusterTopologyMonitorImpl.MONITORING_PROPERTY_PREFIX,

0 commit comments

Comments
 (0)