test: aurora failover performance tests

.github/workflows/aurora_performance.yml

@@ -0,0 +1,89 @@
+name: Aurora Performance Tests
+
+on:
+  workflow_dispatch:
+  push:
+    branches:
+      - performance-tests
+
+jobs:
+  run-integration-tests:
+    strategy:
+      matrix:
+        db: [mysql, pg]
+
+    name: Run Aurora ${{ matrix.db }} container performance tests
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Clone repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 50
+
+      - name: "Set up JDK 8"
+        uses: actions/setup-java@v3
+        with:
+          distribution: "corretto"
+          java-version: 8
+      - name: Set up Node.js
+        uses: actions/setup-node@v3
+        with:
+          node-version: "20.x"
+      - name: Install dependencies
+        run: npm install --no-save
+
+      - name: Configure AWS Credentials
+        uses: aws-actions/configure-aws-credentials@v4
+        with:
+          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
+          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+          aws-region: ${{ secrets.AWS_DEFAULT_REGION }}
+
+      - name: Set up Temp AWS Credentials
+        run: |
+          creds=($(aws sts get-session-token \
+            --duration-seconds 21600 \
+            --query 'Credentials.[AccessKeyId, SecretAccessKey, SessionToken]' \
+            --output text \
+          | xargs));
+          echo "::add-mask::${creds[0]}"
+          echo "::add-mask::${creds[1]}"
+          echo "::add-mask::${creds[2]}"
+          echo "TEMP_AWS_ACCESS_KEY_ID=${creds[0]}" >> $GITHUB_ENV
+          echo "TEMP_AWS_SECRET_ACCESS_KEY=${creds[1]}" >> $GITHUB_ENV
+          echo "TEMP_AWS_SESSION_TOKEN=${creds[2]}" >> $GITHUB_ENV
+
+      - name: Run Integration Tests
+        run: |
+          ./gradlew --no-parallel --no-daemon test-aurora-${{ matrix.db }}-performance --info
+        env:
+          AURORA_CLUSTER_DOMAIN: ${{ secrets.DB_CONN_SUFFIX }}
+          AURORA_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
+
+      - name: "Get Github Action IP"
+        if: always()
+        id: ip
+        uses: haythem/[email protected]
+
+      - name: "Remove Github Action IP"
+        if: always()
+        run: |
+          aws ec2 revoke-security-group-ingress \
+            --group-name default \
+            --protocol -1 \
+            --port -1 \
+            --cidr ${{ steps.ip.outputs.ipv4 }}/32 \
+          2>&1 > /dev/null;
+
+      - name: Archive Performance results
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: ${{ matrix.db }}-performance-results
+          path: ./tests/integration/container/reports
+          retention-days: 5

.github/workflows/integration_tests.yml

@@ -86,7 +86,7 @@ jobs:
 
       - name: Archive results
         if: always()
-        uses: actions/upload-artifact@v3
+        uses: actions/upload-artifact@v4
         with:
           name: integration-report
           path: ./tests/integration/container/reports

common/lib/plugin_service.ts

@@ -282,9 +282,8 @@ export class PluginService implements ErrorHandler, HostListProviderService {
 
   createTargetClient(props: Map<string, any>): any {
     const createClientFunc = this.getCurrentClient().getCreateClientFunc();
-    const copy = WrapperProperties.removeWrapperProperties(Object.fromEntries(props.entries()));
     if (createClientFunc) {
-      return createClientFunc(Object.fromEntries(new Map(Object.entries(copy))));
+      return createClientFunc(props);
     }
     throw new AwsWrapperError("AwsClient is missing create target client function."); // This should not be reached
   }

common/lib/plugins/efm/host_monitoring_connection_plugin.ts

@@ -111,7 +111,14 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp
         failureDetectionIntervalMillis,
         failureDetectionCount
       );
-      result = await methodFunc();
+
+      result = await Promise.race([monitorContext.trackHealthStatus(), methodFunc()])
+        .then((result: any) => {
+          return result;
+        })
+        .catch((error: any) => {
+          throw error;
+        });
     } finally {
       if (monitorContext != null) {
         await this.monitorService.stopMonitoring(monitorContext);
@@ -123,13 +130,15 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp
           }
 
           const targetClient = this.pluginService.getCurrentClient().targetClient;
-          let isClientClosed = false;
+          let isClientValid = false;
           if (targetClient) {
-            isClientClosed = await this.pluginService.isClientValid(targetClient);
+            isClientValid = await this.pluginService.isClientValid(targetClient);
           }
 
-          if (!targetClient || !isClientClosed) {
-            await this.pluginService.getCurrentClient().end();
+          if (!targetClient || !isClientValid) {
+            if (targetClient) {
+              await this.pluginService.tryClosingTargetClient(targetClient);
+            }
             // eslint-disable-next-line no-unsafe-finally
             throw new AwsWrapperError(Messages.get("HostMonitoringConnectionPlugin.unavailableHost", monitoringHostInfo.host));
           }
@@ -192,7 +201,6 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp
   }
 
   async releaseResources(): Promise<void> {
-    const hostKeys = (await this.getMonitoringHostInfo())?.allAliases;
-    return this.monitorService.releaseResources(hostKeys);
+    return this.monitorService.releaseResources();
   }
 }

common/lib/plugins/efm/monitor.ts

@@ -21,6 +21,7 @@ import { logger } from "../../../logutils";
 import { Messages } from "../../utils/messages";
 import { ClientWrapper } from "../../client_wrapper";
 import { sleep } from "../../utils/utils";
+import { WrapperProperties } from "../../wrapper_property";
 
 export interface Monitor {
   startMonitoring(context: MonitorConnectionContext): void;
@@ -51,7 +52,6 @@ class ConnectionStatus {
 export class MonitorImpl implements Monitor {
   private readonly SLEEP_WHEN_INACTIVE_MILLIS: number = 100;
   private readonly MIN_CONNECTION_CHECK_TIMEOUT_MILLIS: number = 3000;
-  private readonly MONITORING_PROPERTY_PREFIX: string = "monitoring_";
 
   private readonly activeContexts: MonitorConnectionContext[] = [];
   private readonly newContexts: MonitorConnectionContext[] = [];
@@ -63,7 +63,7 @@ export class MonitorImpl implements Monitor {
   private contextLastUsedTimestampNanos: number;
   private started = false;
   private stopped: boolean = false;
-  private cancel: boolean = false;
+  private cancelled: boolean = false;
   private monitoringClient: ClientWrapper | null = null;
   private delayMillisTimeoutId: any;
   private sleepWhenInactiveTimeoutId: any;
@@ -110,8 +110,7 @@ export class MonitorImpl implements Monitor {
     logger.debug(Messages.get("MonitorImpl.startMonitoring", this.hostInfo.host));
 
     try {
-      this.stopped = false;
-      while (!this.cancel) {
+      while (!this.cancelled) {
         try {
           let newMonitorContext: MonitorConnectionContext | undefined;
           let firstAddedNewMonitorContext: MonitorConnectionContext | null = null;
@@ -234,11 +233,11 @@ export class MonitorImpl implements Monitor {
       const monitoringConnProperties: Map<string, any> = new Map(this.properties);
 
       for (const key of this.properties.keys()) {
-        if (!key.startsWith(this.MONITORING_PROPERTY_PREFIX)) {
+        if (!key.startsWith(WrapperProperties.MONITORING_PROPERTY_PREFIX)) {
           continue;
         }
 
-        monitoringConnProperties.set(key.substring(this.MONITORING_PROPERTY_PREFIX.length), this.properties.get(key));
+        monitoringConnProperties.set(key.substring(WrapperProperties.MONITORING_PROPERTY_PREFIX.length), this.properties.get(key));
         monitoringConnProperties.delete(key);
       }
 
@@ -258,15 +257,15 @@ export class MonitorImpl implements Monitor {
   }
 
   isStopped(): boolean {
-    return this.stopped;
+    return this.stopped || this.cancelled;
   }
 
   protected getCurrentTimeNano() {
     return Number(process.hrtime.bigint());
   }
 
   async releaseResources() {
-    this.cancel = true;
+    this.cancelled = true;
     clearTimeout(this.delayMillisTimeoutId);
     clearTimeout(this.sleepWhenInactiveTimeoutId);
     await this.endMonitoringClient();

common/lib/plugins/efm/monitor_connection_context.ts

@@ -18,13 +18,18 @@ import { Monitor } from "./monitor";
 import { logger } from "../../../logutils";
 import { Messages } from "../../utils/messages";
 import { ClientWrapper } from "../../client_wrapper";
+import { sleep } from "../../utils/utils";
+import { AwsWrapperError } from "../../utils/errors";
+import { uniqueId } from "lodash";
+import { PluginService } from "../../plugin_service";
 
 export class MonitorConnectionContext {
   readonly failureDetectionIntervalMillis: number;
   private readonly failureDetectionTimeMillis: number;
   private readonly failureDetectionCount: number;
   readonly clientToAbort: ClientWrapper;
   readonly monitor: Monitor;
+  readonly pluginService: PluginService;
 
   isActiveContext: boolean = true;
   isHostUnhealthy: boolean = false;
@@ -33,19 +38,22 @@ export class MonitorConnectionContext {
   private invalidHostStartTimeNano: number = 0;
   private abortedConnectionCounter: number = 0;
   failureCount: number = 0;
+  id: string = uniqueId("_monitorContext");
 
   constructor(
     monitor: Monitor,
     clientToAbort: any,
     failureDetectionTimeMillis: number,
     failureDetectionIntervalMillis: number,
-    failureDetectionCount: number
+    failureDetectionCount: number,
+    pluginService: PluginService
   ) {
     this.monitor = monitor;
     this.clientToAbort = clientToAbort;
     this.failureDetectionTimeMillis = failureDetectionTimeMillis;
     this.failureDetectionIntervalMillis = failureDetectionIntervalMillis;
     this.failureDetectionCount = failureDetectionCount;
+    this.pluginService = pluginService;
   }
 
   resetInvalidHostStartTimeNano(): void {
@@ -62,7 +70,7 @@ export class MonitorConnectionContext {
     }
 
     try {
-      await this.clientToAbort.client.end();
+      await this.pluginService.tryClosingTargetClient(this.clientToAbort);
     } catch (error: any) {
       // ignore
       logger.debug(Messages.get("MonitorConnectionContext.exceptionAbortingConnection", error.message));
@@ -116,7 +124,7 @@ export class MonitorConnectionContext {
       const invalidHostDurationNano: number = statusCheckEndNano - this.invalidHostStartTimeNano;
       const maxInvalidHostDurationMillis: number = this.failureDetectionIntervalMillis * Math.max(0, this.failureDetectionCount);
 
-      if (invalidHostDurationNano >= maxInvalidHostDurationMillis * 1_000_000) {
+      if (this.failureCount >= this.failureDetectionCount || invalidHostDurationNano >= maxInvalidHostDurationMillis * 1_000_000) {
         logger.debug(Messages.get("MonitorConnectionContext.hostDead", hostName));
         this.isHostUnhealthy = true;
         await this.abortConnection();
@@ -133,4 +141,12 @@ export class MonitorConnectionContext {
 
     logger.debug(Messages.get("MonitorConnectionContext.hostAlive", hostName));
   }
+
+  async trackHealthStatus() {
+    while (!this.isHostUnhealthy && this.isActiveContext) {
+      await sleep(100);
+    }
+
+    throw new AwsWrapperError("trackHealthStatus stopped");
+  }
 }

common/lib/plugins/efm/monitor_service.ts

@@ -38,7 +38,7 @@ export interface MonitorService {
 
   stopMonitoringForAllConnections(hostKeys: Set<string>): void;
 
-  releaseResources(hostKeys: Set<string> | undefined): Promise<void>;
+  releaseResources(): Promise<void>;
 }
 
 export class MonitorServiceImpl implements MonitorService {
@@ -87,7 +87,8 @@ export class MonitorServiceImpl implements MonitorService {
         clientToAbort,
         failureDetectionTimeMillis,
         failureDetectionIntervalMillis,
-        failureDetectionCount
+        failureDetectionCount,
+        this.pluginService
       );
       monitor.startMonitoring(context);
       return context;
@@ -152,7 +153,7 @@ export class MonitorServiceImpl implements MonitorService {
     }
   }
 
-  async releaseResources(hostKeys: Set<string>) {
+  async releaseResources() {
     for (const [key, monitor] of MonitorServiceImpl.monitors.entries) {
       if (monitor.item) {
         await monitor.item.releaseResources();

common/lib/plugins/failover/failover_plugin.ts

@@ -15,7 +15,7 @@
 */
 
 import { AbstractConnectionPlugin } from "../../abstract_connection_plugin";
-import { uniqueId } from "lodash";
+import { method, uniqueId } from "lodash";
 import { logger } from "../../../logutils";
 import { HostInfo } from "../../host_info";
 import { OldConnectionSuggestionAction } from "../../old_connection_suggestion_action";

common/lib/utils/client_utils.ts

@@ -21,9 +21,13 @@ import { logger } from "../../logutils";
 import { WrapperProperties } from "../wrapper_property";
 
 export class ClientUtils {
-  static async queryWithTimeout(newPromise: Promise<any>, props: Map<string, any>): Promise<any> {
+  static async queryWithTimeout(newPromise: Promise<any>, props: Map<string, any>, timeValue?: number): Promise<any> {
     const timer: any = {};
-    const timeoutTask = getTimeoutTask(timer, Messages.get("ClientUtils.queryTaskTimeout"), WrapperProperties.INTERNAL_QUERY_TIMEOUT.get(props));
+    const timeoutTask = getTimeoutTask(
+      timer,
+      Messages.get("ClientUtils.queryTaskTimeout"),
+      timeValue ?? WrapperProperties.INTERNAL_QUERY_TIMEOUT.get(props)
+    );
     return await Promise.race([timeoutTask, newPromise])
       .then((result) => {
         if (result) {

common/lib/wrapper_property.ts

@@ -40,6 +40,7 @@ export class WrapperProperty<T> {
 }
 
 export class WrapperProperties {
+  static readonly MONITORING_PROPERTY_PREFIX: string = "monitoring_";
   static readonly DEFAULT_PLUGINS = "auroraConnectionTracker,failover,hostMonitoring";
   static readonly DEFAULT_TOKEN_EXPIRATION_SEC = 15 * 60;
 
@@ -240,8 +241,7 @@ export class WrapperProperties {
     600000 // 10 minutes
   );
 
-  static removeWrapperProperties<T>(config: T): T {
-    const copy = Object.assign({}, config);
+  static removeWrapperProperties(props: Map<string, any>): any {
     const persistingProperties = [
       WrapperProperties.USER.name,
       WrapperProperties.PASSWORD.name,
@@ -250,16 +250,25 @@ export class WrapperProperties {
       WrapperProperties.HOST.name
     ];
 
+    const copy = new Map(props);
+
+    for (const key of props.keys()) {
+      if (!key.startsWith(WrapperProperties.MONITORING_PROPERTY_PREFIX)) {
+        continue;
+      }
+
+      copy.delete(key);
+    }
+
     Object.values(WrapperProperties).forEach((prop) => {
       if (prop instanceof WrapperProperty) {
         const propertyName = (prop as WrapperProperty<any>).name;
-        if (!persistingProperties.includes(propertyName) && Object.hasOwn(config as object, propertyName)) {
-          // @ts-expect-error
-          delete copy[propertyName];
+        if (!persistingProperties.includes(propertyName) && copy.has(propertyName)) {
+          copy.delete(propertyName);
         }
       }
     });
 
-    return copy;
+    return Object.fromEntries(copy.entries());
   }
 }