test

common/lib/host_list_provider/monitoring/monitoring_host_list_provider.ts

@@ -27,6 +27,7 @@ import { Messages } from "../../utils/messages";
 import { WrapperProperties } from "../../wrapper_property";
 import { BlockingHostListProvider } from "../host_list_provider";
 import { logger } from "../../../logutils";
+import { isDialectTopologyAware } from "../../utils/utils";
 
 export class MonitoringRdsHostListProvider extends RdsHostListProvider implements BlockingHostListProvider {
   static readonly CACHE_CLEANUP_NANOS: bigint = BigInt(60_000_000_000); // 1 minute.
@@ -76,7 +77,7 @@ export class MonitoringRdsHostListProvider extends RdsHostListProvider implement
 
   async sqlQueryForTopology(targetClient: ClientWrapper): Promise<HostInfo[]> {
     const dialect: DatabaseDialect = this.hostListProviderService.getDialect();
-    if (!this.isTopologyAwareDatabaseDialect(dialect)) {
+    if (!isDialectTopologyAware(dialect)) {
       throw new TypeError(Messages.get("RdsHostListProvider.incorrectDialect"));
     }
     return await dialect.queryForTopology(targetClient, this).then((res: any) => this.processQueryResults(res));

common/lib/host_list_provider/rds_host_list_provider.ts

@@ -149,7 +149,7 @@ export class RdsHostListProvider implements DynamicHostListProvider {
 
   async getWriterId(client: ClientWrapper): Promise<string | null> {
     const dialect = this.hostListProviderService.getDialect();
-    if (!this.isTopologyAwareDatabaseDialect(dialect)) {
+    if (!isDialectTopologyAware(dialect)) {
       throw new TypeError(Messages.get("RdsHostListProvider.incorrectDialect"));
     }
 

common/lib/plugin_service.ts

@@ -273,6 +273,9 @@ export class PluginService implements ErrorHandler, HostListProviderService {
   }
 
   private async setHostList(oldHosts: HostInfo[], newHosts: HostInfo[]) {
+    logger.silly("Setting host list");
+    console.log(oldHosts);
+    console.log(newHosts);
     const oldHostMap: Map<string, HostInfo> = new Map(oldHosts.map((e) => [e.url, e]));
     const newHostMap: Map<string, HostInfo> = new Map(newHosts.map((e) => [e.url, e]));
 
@@ -295,7 +298,10 @@ export class PluginService implements ErrorHandler, HostListProviderService {
       }
     });
 
+    console.log("changes: ");
+    console.log(changes);
     if (changes.size > 0) {
+      logger.silly("Changes found, setting hosts");
       this.hosts = newHosts ? newHosts : [];
       await this.pluginServiceManagerContainer.pluginManager!.notifyHostListChanged(changes);
     }

common/lib/plugins/failover/failover_plugin.ts

@@ -415,7 +415,11 @@ export class FailoverPlugin extends AbstractConnectionPlugin {
             throw new AwsWrapperError(Messages.get("Failover.unableToDetermineWriter"));
           }
 
+          logger.silly("Failover writer complete, refreshing host list");
+          await this.pluginService.refreshHostList();
+          logger.silly(`Failover writer complete with writer host: ${writerHostInfo.host}`);
           const allowedHosts = this.pluginService.getHosts();
+          console.log(allowedHosts);
           if (!allowedHosts.some((hostInfo: HostInfo) => hostInfo.host === writerHostInfo?.host)) {
             const failoverErrorMessage = Messages.get(
               "Failover.newWriterNotAllowed",
@@ -430,7 +434,6 @@ export class FailoverPlugin extends AbstractConnectionPlugin {
           await this.pluginService.abortCurrentClient();
           await this.pluginService.setCurrentClient(result.client, writerHostInfo);
           logger.debug(Messages.get("Failover.establishedConnection", this.pluginService.getCurrentHostInfo()?.host ?? ""));
-          await this.pluginService.refreshHostList();
           this.throwFailoverSuccessError();
         } catch (error: any) {
           if (error instanceof FailoverSuccessError) {

common/lib/plugins/failover2/failover2_plugin.ts

@@ -119,8 +119,8 @@ export class Failover2Plugin extends AbstractConnectionPlugin implements CanRele
     return (
       this.enableFailoverSetting &&
       this._rdsUrlType !== RdsUrlType.RDS_PROXY &&
-      this.pluginService.getHosts() &&
-      this.pluginService.getHosts().length > 0
+      this.pluginService.getAllHosts() &&
+      this.pluginService.getAllHosts().length > 0
     );
   }
 
@@ -382,7 +382,7 @@ export class Failover2Plugin extends AbstractConnectionPlugin implements CanRele
           this.logAndThrowError(Messages.get("Failover2.unableToFetchTopology"));
         }
 
-        const hosts: HostInfo[] = this.pluginService.getHosts();
+        const hosts: HostInfo[] = this.pluginService.getAllHosts();
 
         let writerCandidateClient: ClientWrapper = null;
         const writerCandidateHostInfo: HostInfo = hosts.find((x) => x.role === HostRole.WRITER);

tests/integration/container/tests/autoscaling.test.ts

@@ -19,7 +19,7 @@ import { DriverHelper } from "./utils/driver_helper";
 import { AuroraTestUtility } from "./utils/aurora_test_utility";
 import { logger } from "../../../../common/logutils";
 import { TestEnvironmentFeatures } from "./utils/test_environment_features";
-import { features, instanceCount } from "./config";
+import { features, instanceCount, runTests } from "./config";
 import { InternalPooledConnectionProvider } from "../../../../common/lib/internal_pooled_connection_provider";
 import { AwsPoolConfig } from "../../../../common/lib/aws_pool_config";
 import { TestInstanceInfo } from "./utils/test_instance_info";
@@ -28,6 +28,7 @@ import { FailoverSuccessError } from "../../../../common/lib/utils/errors";
 import { PluginManager } from "../../../../common/lib";
 
 const itIf =
+  runTests &&
   !features.includes(TestEnvironmentFeatures.PERFORMANCE) &&
   features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) &&
   instanceCount >= 2

tests/integration/container/tests/basic_connectivity.test.ts

@@ -21,12 +21,14 @@ import { AuroraTestUtility } from "./utils/aurora_test_utility";
 import { logger } from "../../../../common/logutils";
 import { DatabaseEngine } from "./utils/database_engine";
 import { TestEnvironmentFeatures } from "./utils/test_environment_features";
-import { features } from "./config";
+import { features, runTests } from "./config";
 import { DatabaseEngineDeployment } from "./utils/database_engine_deployment";
 import { PluginManager } from "../../../../common/lib";
 
 const itIf =
-  !features.includes(TestEnvironmentFeatures.PERFORMANCE) && !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) ? it : it.skip;
+  runTests && !features.includes(TestEnvironmentFeatures.PERFORMANCE) && !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY)
+    ? it
+    : it.skip;
 
 let client: any;
 let auroraTestUtility: AuroraTestUtility;

tests/integration/container/tests/config.ts

@@ -34,3 +34,4 @@ const testInfo = JSON.parse(infoJson);
 const request = testInfo.request;
 export const features = request.features;
 export const instanceCount = request.numOfInstances;
+export const runTests = false; // TODO: remove

tests/integration/container/tests/custom_endpoint.test.ts

@@ -62,7 +62,7 @@ let currentWriter: string;
 
 let auroraTestUtility: AuroraTestUtility;
 
-async function initDefaultConfig(host: string, port: number, connectToProxy: boolean, failoverMode: string): Promise<any> {
+async function initDefaultConfig(host: string, port: number, connectToProxy: boolean, failoverMode: string, usingFailover1: boolean): Promise<any> {
   let config: any = {
     user: env.databaseInfo.username,
     host: host,
@@ -76,6 +76,11 @@ async function initDefaultConfig(host: string, port: number, connectToProxy: boo
     telemetryTracesBackend: "OTLP",
     telemetryMetricsBackend: "OTLP"
   };
+  if (usingFailover1) {
+    config["plugins"] = "customEndpoint,readWriteSplitting,failover";
+  } else {
+    config["plugins"] = "customEndpoint,readWriteSplitting,failover2";
+  }
   if (connectToProxy) {
     config["clusterInstanceHostPattern"] = "?." + env.proxyDatabaseInfo.instanceEndpointSuffix;
   }
@@ -232,6 +237,8 @@ describe("custom endpoint", () => {
   beforeEach(async () => {
     await TestEnvironment.verifyClusterStatus();
     currentWriter = await auroraTestUtility.getClusterWriterInstanceId(env.info.auroraClusterName);
+    logger.info(`Test started: ${expect.getState().currentTestName}`);
+    console.log("custom endpoint test before: " + currentWriter);
   }, 1000000);
 
   afterEach(async () => {
@@ -244,12 +251,59 @@ describe("custom endpoint", () => {
     }
 
     await PluginManager.releaseResources();
-  });
+  }, 1000000);
+
+  itIf.each([true, false])(
+    "test custom endpoint failover - strict reader",
+    async (usingFailover1: boolean) => {
+      await createEndpoint(env.auroraClusterName, env.instances.slice(0, 2), endpointId3, "READER");
+      endpointInfo3 = await waitUntilEndpointAvailable(endpointId3);
+
+      const config = await initDefaultConfig(endpointInfo3.Endpoint, env.databaseInfo.instanceEndpointPort, false, "strict-reader", usingFailover1);
+      client = initClientFunc(config);
+
+      await client.connect();
+
+      const endpointMembers = endpointInfo3.StaticMembers;
+      const instanceId = await auroraTestUtility.queryInstanceId(client);
+      console.log("current test " + usingFailover1);
+      console.log(instanceId);
+      expect(endpointMembers.includes(instanceId)).toBeTruthy();
+      expect(instanceId).not.toBe(currentWriter);
+
+      // Use failover API to break connection.
+      await auroraTestUtility.failoverClusterAndWaitUntilWriterChanged(
+        currentWriter,
+        env.info.auroraClusterName,
+        instanceId === instance1 ? instance1 : instance2
+      );
+
+      await expect(auroraTestUtility.queryInstanceId(client)).rejects.toThrow(FailoverSuccessError);
+
+      endpointInfo3 = await waitUntilEndpointAvailable(endpointId3);
+      const newEndpointMembers = endpointInfo3.StaticMembers;
+
+      const newInstanceId: string = await auroraTestUtility.queryInstanceId(client);
+      expect(newEndpointMembers.includes(newInstanceId)).toBeTruthy();
+
+      const newWriter = await auroraTestUtility.getClusterWriterInstanceId(env.info.auroraClusterName);
+      expect(newInstanceId).not.toBe(newWriter);
+
+      await deleteEndpoint(rdsClient, endpointId3);
+    },
+    1000000
+  );
 
-  itIf(
+  itIf.each([true, false])(
     "test custom endpoint read write splitting with custom endpoint changes",
-    async () => {
-      const config = await initDefaultConfig(endpointInfo1.Endpoint, env.databaseInfo.instanceEndpointPort, false, "reader-or-writer");
+    async (usingFailover1: boolean) => {
+      const config = await initDefaultConfig(
+        endpointInfo1.Endpoint,
+        env.databaseInfo.instanceEndpointPort,
+        false,
+        "reader-or-writer",
+        usingFailover1
+      );
       // This setting is not required for the test, but it allows us to also test re-creation of expired monitors since it
       // takes more than 30 seconds to modify the cluster endpoint (usually around 140s).
       config.customEndpointMonitorExpirationMs = 30000;
@@ -323,49 +377,10 @@ describe("custom endpoint", () => {
     1000000
   );
 
-  itIf(
-    "test custom endpoint failover - strict reader",
-    async () => {
-      await createEndpoint(env.auroraClusterName, env.instances.slice(0, 2), endpointId3, "READER");
-      endpointInfo3 = await waitUntilEndpointAvailable(endpointId3);
-
-      const config = await initDefaultConfig(endpointInfo3.Endpoint, env.databaseInfo.instanceEndpointPort, false, "strict-reader");
-      client = initClientFunc(config);
-
-      await client.connect();
-
-      const endpointMembers = endpointInfo3.StaticMembers;
-      const instanceId = await auroraTestUtility.queryInstanceId(client);
-      expect(endpointMembers.includes(instanceId)).toBeTruthy();
-      expect(instanceId).not.toBe(currentWriter);
-
-      // Use failover API to break connection.
-      await auroraTestUtility.failoverClusterAndWaitUntilWriterChanged(
-        currentWriter,
-        env.info.auroraClusterName,
-        instanceId === instance1 ? instance1 : instance2
-      );
-
-      await expect(auroraTestUtility.queryInstanceId(client)).rejects.toThrow(FailoverSuccessError);
-
-      endpointInfo3 = await waitUntilEndpointAvailable(endpointId3);
-      const newEndpointMembers = endpointInfo3.StaticMembers;
-
-      const newInstanceId: string = await auroraTestUtility.queryInstanceId(client);
-      expect(newEndpointMembers.includes(newInstanceId)).toBeTruthy();
-
-      const newWriter = await auroraTestUtility.getClusterWriterInstanceId(env.info.auroraClusterName);
-      expect(newInstanceId).not.toBe(newWriter);
-
-      await deleteEndpoint(rdsClient, endpointId3);
-    },
-    1000000
-  );
-
-  itIf(
+  itIf.each([true, false])(
     "test custom endpoint failover - strict writer",
-    async () => {
-      const config = await initDefaultConfig(endpointInfo2.Endpoint, env.databaseInfo.instanceEndpointPort, false, "strict-writer");
+    async (usingFailvoer1: boolean) => {
+      const config = await initDefaultConfig(endpointInfo2.Endpoint, env.databaseInfo.instanceEndpointPort, false, "strict-writer", usingFailvoer1);
       client = initClientFunc(config);
 
       await client.connect();
@@ -399,10 +414,16 @@ describe("custom endpoint", () => {
     1000000
   );
 
-  itIf(
+  itIf.each([true, false])(
     "test custom endpoint failover - reader or writer mode",
-    async () => {
-      const config = await initDefaultConfig(endpointInfo1.Endpoint, env.databaseInfo.instanceEndpointPort, false, "reader-or-writer");
+    async (usingFailover1: boolean) => {
+      const config = await initDefaultConfig(
+        endpointInfo1.Endpoint,
+        env.databaseInfo.instanceEndpointPort,
+        false,
+        "reader-or-writer",
+        usingFailover1
+      );
       client = initClientFunc(config);
 
       await client.connect();

tests/integration/container/tests/iam_authentication.test.ts

@@ -25,11 +25,12 @@ import { AwsMySQLClient } from "../../../../mysql/lib";
 import { IamAuthenticationPlugin } from "../../../../common/lib/authentication/iam_authentication_plugin";
 import { logger } from "../../../../common/logutils";
 import { TestEnvironmentFeatures } from "./utils/test_environment_features";
-import { features } from "./config";
+import { features, runTests } from "./config";
 import { PluginManager } from "../../../../common/lib";
 import { jest } from "@jest/globals";
 
 const itIf =
+  runTests &&
   !features.includes(TestEnvironmentFeatures.PERFORMANCE) &&
   !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) &&
   features.includes(TestEnvironmentFeatures.IAM)

tests/integration/container/tests/read_write_splitting.test.ts

@@ -23,14 +23,15 @@ import { QueryResult } from "pg";
 import { ProxyHelper } from "./utils/proxy_helper";
 import { logger } from "../../../../common/logutils";
 import { TestEnvironmentFeatures } from "./utils/test_environment_features";
-import { features, instanceCount } from "./config";
+import { features, instanceCount, runTests } from "./config";
 import { InternalPooledConnectionProvider } from "../../../../common/lib/internal_pooled_connection_provider";
 import { AwsPoolConfig } from "../../../../common/lib/aws_pool_config";
 import { InternalPoolMapping } from "../../../../common/lib/utils/internal_pool_mapping";
 import { HostInfo } from "../../../../common/lib/host_info";
 import { PluginManager } from "../../../../common/lib";
 
 const itIf =
+  runTests &&
   !features.includes(TestEnvironmentFeatures.PERFORMANCE) &&
   features.includes(TestEnvironmentFeatures.IAM) &&
   !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) &&