fix: autoscaling tests (#316)

.github/workflows/run-autoscaling-tests.yml

@@ -11,6 +11,10 @@ jobs:
   autoscaling-tests:
     name: Run Autoscaling Tests
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        dbEngine: [ "mysql", "postgres" ]
     steps:
       - name: Clone repository
         uses: actions/checkout@v4
@@ -46,17 +50,16 @@ jobs:
           echo "TEMP_AWS_SESSION_TOKEN=${creds[2]}" >> $GITHUB_ENV
       - name: Run integration tests
         run: |
-          ./gradlew --no-parallel --no-daemon test-autoscaling --info
+          ./gradlew --no-parallel --no-daemon test-autoscaling-${{ matrix.dbEngine }} --info
         env:
-          AURORA_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
-          AURORA_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
+          RDS_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
+          RDS_DB_REGION: ${{ secrets.AWS_DEFAULT_REGION }}
           AWS_ACCESS_KEY_ID: ${{ env.TEMP_AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ env.TEMP_AWS_SECRET_ACCESS_KEY }}
           AWS_SESSION_TOKEN: ${{ env.TEMP_AWS_SESSION_TOKEN }}
           NUM_INSTANCES: 5
           AURORA_MYSQL_DB_ENGINE_VERSION: "default"
-          AURORA_PG_ENGINE_VERSION: "default"
-          FILTER: "autoscaling"
+          AURORA_PG_DB_ENGINE_VERSION: "default"
       - name: "Get Github Action IP"
         if: always()
         id: ip

common/lib/plugins/read_write_splitting_plugin.ts

@@ -29,7 +29,6 @@ import { SqlMethodUtils } from "../utils/sql_method_utils";
 import { ClientWrapper } from "../client_wrapper";
 import { getWriter, logAndThrowError } from "../utils/utils";
 import { CanReleaseResources } from "../can_release_resources";
-import { InternalPooledConnectionProvider } from "../internal_pooled_connection_provider";
 import { PoolClientWrapper } from "../pool_client_wrapper";
 
 export class ReadWriteSplittingPlugin extends AbstractConnectionPlugin implements CanReleaseResources {
@@ -209,7 +208,7 @@ export class ReadWriteSplittingPlugin extends AbstractConnectionPlugin implement
   async switchClientIfRequired(readOnly: boolean) {
     const currentClient = this.pluginService.getCurrentClient();
     if (!(await currentClient.isValid())) {
-      logAndThrowError(Messages.get("ReadWriteSplittingPlugin.setReadOnlyOnClosedClient"));
+      logAndThrowError(Messages.get("ReadWriteSplittingPlugin.setReadOnlyOnClosedClient", currentClient.targetClient?.id ?? "undefined client"));
     }
     try {
       await this.pluginService.refreshHostList();
@@ -257,7 +256,7 @@ export class ReadWriteSplittingPlugin extends AbstractConnectionPlugin implement
     if (newClientHost && newTargetClient) {
       try {
         await this.pluginService.setCurrentClient(newTargetClient, newClientHost);
-        logger.debug(Messages.get("ReadWriteSplittingPlugin.settingCurrentClient", newClientHost.url));
+        logger.debug(Messages.get("ReadWriteSplittingPlugin.settingCurrentClient", newTargetClient.id, newClientHost.url));
       } catch (error) {
         // pass
       }

common/lib/utils/locales/en.json

@@ -71,7 +71,7 @@
   "AuroraStaleDnsHelper.writerHostSpec": "Writer host: '%s'.",
   "AuroraStaleDnsHelper.writerInetAddress": "Writer host address: '%s'",
   "AuroraStaleDnsHelper.staleDnsDetected": "Stale DNS data detected. Opening a connection to '%s'.",
-  "ReadWriteSplittingPlugin.setReadOnlyOnClosedClient": "setReadOnly cannot be called on a closed client.",
+  "ReadWriteSplittingPlugin.setReadOnlyOnClosedClient": "setReadOnly cannot be called on a closed client '%s'.",
   "ReadWriteSplittingPlugin.errorSwitchingToCachedReader": "An error occurred while trying to switch to a cached reader client: '%s'. The driver will attempt to establish a new reader client.",
   "ReadWriteSplittingPlugin.errorSwitchingToReader": "An error occurred while trying to switch to a reader client: '%s'.",
   "ReadWriteSplittingPlugin.errorSwitchingToWriter": "An error occurred while trying to switch to a writer client: '%s'.",
@@ -83,7 +83,7 @@
   "ReadWriteSplittingPlugin.fallbackToWriter": "Failed to switch to a reader; the current writer will be used as a fallback: '%s'",
   "ReadWriteSplittingPlugin.switchedFromWriterToReader": "Switched from a writer to a reader host. New reader host: '%s'",
   "ReadWriteSplittingPlugin.switchedFromReaderToWriter": "Switched from a reader to a writer host. New writer host: '%s'",
-  "ReadWriteSplittingPlugin.settingCurrentClient": "Setting the current client to '%s'",
+  "ReadWriteSplittingPlugin.settingCurrentClient": "Setting the current client to '%s' - '%s'",
   "ReadWriteSplittingPlugin.noWriterFound": "No writer was found in the current host list.",
   "ReadWriteSplittingPlugin.noReadersFound": "A reader instance was requested via setReadOnly, but there are no readers in the host list. The current writer will be used as a fallback: '%s'",
   "ReadWriteSplittingPlugin.emptyHostList": "Host list is empty.",

pg/lib/abstract_pg_error_handler.ts

@@ -85,7 +85,7 @@ export abstract class AbstractPgErrorHandler implements ErrorHandler {
     if (!clientWrapper || !clientWrapper.client) {
       return;
     }
-    clientWrapper.client.removeListener("error", this.noOpListener);
+    clientWrapper.client.removeListener("error", this.trackingListener);
     clientWrapper.client.on("error", this.noOpListener);
   }
 }

tests/integration/container/tests/aurora_failover.test.ts

@@ -27,7 +27,10 @@ import { features, instanceCount } from "./config";
 import { TestEnvironmentFeatures } from "./utils/test_environment_features";
 
 const itIf =
-  features.includes(TestEnvironmentFeatures.FAILOVER_SUPPORTED) && !features.includes(TestEnvironmentFeatures.PERFORMANCE) && instanceCount >= 2
+  features.includes(TestEnvironmentFeatures.FAILOVER_SUPPORTED) &&
+  !features.includes(TestEnvironmentFeatures.PERFORMANCE) &&
+  !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) &&
+  instanceCount >= 2
     ? it
     : it.skip;
 const itIfTwoInstance = instanceCount == 2 ? itIf : it.skip;

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

@@ -33,6 +33,7 @@ const itIf =
   instanceCount >= 2
     ? it
     : it.skip;
+const itIfMinFiveInstance = instanceCount >= 5 ? itIf : it.skip;
 
 let env: TestEnvironment;
 let driver;
@@ -42,6 +43,7 @@ let newInstance: TestInstanceInfo;
 let newInstanceClient: any;
 let auroraTestUtility: AuroraTestUtility;
 let provider: InternalPooledConnectionProvider | null;
+const instanceId: string = "auto-scaling-instance";
 
 async function initDefaultConfig(host: string, port: number, provider: InternalPooledConnectionProvider): Promise<any> {
   let config: any = {
@@ -55,7 +57,8 @@ async function initDefaultConfig(host: string, port: number, provider: InternalP
     enableTelemetry: true,
     telemetryTracesBackend: "OTLP",
     telemetryMetricsBackend: "OTLP",
-    readerHostSelectorStrategy: "leastConnections"
+    readerHostSelectorStrategy: "leastConnections",
+    clusterTopologyRefreshRateMs: 5000
   };
 
   config = DriverHelper.addDriverSpecificConfiguration(config, env.engine);
@@ -75,7 +78,8 @@ async function initConfigWithFailover(host: string, port: number, provider: Inte
     enableTelemetry: true,
     telemetryTracesBackend: "OTLP",
     telemetryMetricsBackend: "OTLP",
-    readerHostSelectorStrategy: "leastConnections"
+    readerHostSelectorStrategy: "leastConnections",
+    clusterTopologyRefreshRateMs: 5000
   };
 
   config = DriverHelper.addDriverSpecificConfiguration(config, env.engine);
@@ -93,6 +97,7 @@ describe("pooled connection autoscaling", () => {
 
     connectionsSet = new Set();
     provider = null;
+    await auroraTestUtility.deleteInstance(instanceId);
     await TestEnvironment.verifyClusterStatus();
   }, 1320000);
 
@@ -107,16 +112,27 @@ describe("pooled connection autoscaling", () => {
     logger.info(`Test finished: ${expect.getState().currentTestName}`);
   }, 1320000);
 
-  itIf(
+  itIfMinFiveInstance(
     "set read only on old connection",
     async () => {
       // Test setup.
       const totalInstances: number = await auroraTestUtility.getNumberOfInstances();
       const instances: TestInstanceInfo[] = env.databaseInfo.instances;
       const numInstances: number = instances.length;
+      const idleTimeoutMillis = 10 * 60 * 1000; // 10 minutes
+      const poolExpirationNanos = BigInt(3 * 60 * 1000_000_000); // 3 minutes
+      const poolCleanupNanos = BigInt(10 * 60 * 1000_000_000); // 10 minutes
 
       // Set provider.
-      provider = new InternalPooledConnectionProvider(new AwsPoolConfig({ maxConnections: numInstances }));
+      provider = new InternalPooledConnectionProvider(
+        new AwsPoolConfig({
+          maxConnections: numInstances,
+          idleTimeoutMillis: idleTimeoutMillis
+        }),
+        undefined,
+        poolExpirationNanos,
+        poolCleanupNanos
+      );
 
       // Initialize clients.
       try {
@@ -126,18 +142,12 @@ describe("pooled connection autoscaling", () => {
           if (host && port) {
             const config: any = await initDefaultConfig(host, port, provider);
             const client = initClientFunc(config);
-            client.on("error", (error: any): void => {
-              logger.debug(`event emitter threw error: ${error.message}`);
-              logger.debug(error.stack);
-            });
-
             await client.connect();
             connectionsSet.add(client);
           }
         }
 
         // Create new instance.
-        const instanceId: string = "auto-scaling-instance";
         newInstance = await auroraTestUtility.createInstance(instanceId);
         if (!newInstance?.instanceId || !newInstance?.host || !newInstance?.port) {
           fail("Instance not returned.");
@@ -149,32 +159,36 @@ describe("pooled connection autoscaling", () => {
           newInstanceClient = initClientFunc(config);
           await newInstanceClient.connect();
           connectionsSet.add(newInstanceClient);
+          const writerInstance = await auroraTestUtility.queryInstanceId(newInstanceClient);
 
           // Should connect to created instance.
           await newInstanceClient.setReadOnly(true);
           expect(await provider.containsHost(newInstance.host)).toBe(true);
           expect(await auroraTestUtility.queryInstanceId(newInstanceClient)).toBe(newInstance.instanceId);
 
           await newInstanceClient.setReadOnly(false);
+          expect(await auroraTestUtility.queryInstanceId(newInstanceClient)).toBe(writerInstance);
         } finally {
-          await auroraTestUtility.deleteInstance(newInstance.instanceId ? newInstance.instanceId : instanceId);
-        }
-
-        // Ensure instance has deleted.
-        const waitTilTime: number = Date.now() + 5 * 60 * 1000; // 5 minutes
-        while ((await auroraTestUtility.getNumberOfInstances()) !== totalInstances && waitTilTime > Date.now()) {
-          await sleep(5000);
-        }
-        if (await auroraTestUtility.instanceExists(instanceId)) {
-          fail(`The instance ${instanceId} was not deleted.`);
+          const instance = newInstance.instanceId ? newInstance.instanceId : instanceId;
+          let deleted = false;
+          setTimeout(async () => {
+            const stopTime = Date.now() + 5 * 60 * 1000;
+            while (!deleted && Date.now() < stopTime) {
+              await auroraTestUtility.queryInstanceId(newInstanceClient);
+              await sleep(3000);
+            }
+          }, 3000);
+          await auroraTestUtility.deleteInstance(instance);
+          deleted = true;
         }
 
         // Should have removed the pool with the deleted instance.
         await newInstanceClient.setReadOnly(true);
+
         const readerId = await auroraTestUtility.queryInstanceId(newInstanceClient);
         expect(newInstance.instanceId).not.toBe(readerId);
         expect(await provider.containsHost(newInstance.host)).toBe(false);
-        expect(provider.getHostCount()).toBe(instances.length);
+        expect(provider.getHostCount()).toBeLessThanOrEqual(instances.length);
       } finally {
         for (const connection of connectionsSet) {
           try {
@@ -188,15 +202,15 @@ describe("pooled connection autoscaling", () => {
     1320000
   );
 
-  itIf(
+  itIfMinFiveInstance(
     "failover from deleted reader",
     async () => {
       // Test setup.
       const instances: TestInstanceInfo[] = env.databaseInfo.instances;
       const numInstances: number = instances.length;
 
       // Set provider.
-      provider = new InternalPooledConnectionProvider(new AwsPoolConfig({ maxConnections: numInstances }));
+      provider = new InternalPooledConnectionProvider(new AwsPoolConfig({ maxConnections: numInstances * 5 }));
 
       // Initialize clients.
       try {
@@ -206,26 +220,24 @@ describe("pooled connection autoscaling", () => {
           if (host && port) {
             const config: any = await initConfigWithFailover(host, port, provider);
             const client = initClientFunc(config);
-            client.on("error", (error: any): void => {
-              logger.debug(`event emitter threw error: ${error.message}`);
-              logger.debug(error.stack);
-            });
+            const newClient = initClientFunc(config);
+            await newClient.connect();
+            connectionsSet.add(newClient);
 
             await client.connect();
             connectionsSet.add(client);
           }
         }
 
         // Create new instance.
-        const instanceId: string = "auto-scaling-instance";
         newInstance = await auroraTestUtility.createInstance(instanceId);
         if (!newInstance?.instanceId || !newInstance?.host || !newInstance?.port) {
           fail("Instance not returned.");
         }
 
         // Connect to instance.
         try {
-          const config = await initConfigWithFailover(newInstance.host, newInstance.port, provider);
+          const config = await initConfigWithFailover(env.databaseInfo.writerInstanceEndpoint, env.databaseInfo.instanceEndpointPort, provider);
           newInstanceClient = initClientFunc(config);
           await newInstanceClient.connect();
           connectionsSet.add(newInstanceClient);

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

@@ -24,7 +24,8 @@ import { TestEnvironmentFeatures } from "./utils/test_environment_features";
 import { features } from "./config";
 import { DatabaseEngineDeployment } from "./utils/database_engine_deployment";
 
-const itIf = !features.includes(TestEnvironmentFeatures.PERFORMANCE) ? it : it.skip;
+const itIf =
+  !features.includes(TestEnvironmentFeatures.PERFORMANCE) && !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) ? it : it.skip;
 
 let client: any;
 let auroraTestUtility: AuroraTestUtility;

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

@@ -27,7 +27,12 @@ import { logger } from "../../../../common/logutils";
 import { TestEnvironmentFeatures } from "./utils/test_environment_features";
 import { features } from "./config";
 
-const itIf = !features.includes(TestEnvironmentFeatures.PERFORMANCE) && features.includes(TestEnvironmentFeatures.IAM) ? it : it.skip;
+const itIf =
+  !features.includes(TestEnvironmentFeatures.PERFORMANCE) &&
+  !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) &&
+  features.includes(TestEnvironmentFeatures.IAM)
+    ? it
+    : it.skip;
 
 let env: TestEnvironment;
 let driver;

tests/integration/container/tests/performance.test.ts

@@ -30,7 +30,8 @@ import { PerfTestUtility } from "./utils/perf_util";
 const itIf =
   features.includes(TestEnvironmentFeatures.FAILOVER_SUPPORTED) &&
   features.includes(TestEnvironmentFeatures.PERFORMANCE) &&
-  features.includes(TestEnvironmentFeatures.NETWORK_OUTAGES_ENABLED)
+  features.includes(TestEnvironmentFeatures.NETWORK_OUTAGES_ENABLED) &&
+  !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY)
     ? it
     : it.skip;
 

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

@@ -31,7 +31,12 @@ import { InternalPoolMapping } from "../../../../common/lib/utils/internal_pool_
 import { HostInfo } from "../../../../common/lib/host_info";
 
 const itIf =
-  !features.includes(TestEnvironmentFeatures.PERFORMANCE) && features.includes(TestEnvironmentFeatures.IAM) && instanceCount >= 2 ? it : it.skip;
+  !features.includes(TestEnvironmentFeatures.PERFORMANCE) &&
+  features.includes(TestEnvironmentFeatures.IAM) &&
+  !features.includes(TestEnvironmentFeatures.RUN_AUTOSCALING_TESTS_ONLY) &&
+  instanceCount >= 2
+    ? it
+    : it.skip;
 const itIfMinThreeInstance = instanceCount >= 3 ? itIf : it.skip;
 const itIfMinFiveInstance = instanceCount >= 5 ? itIf : it.skip;