test: run multi-az integration tests

.github/workflows/aurora_performance.yml

@@ -55,8 +55,8 @@ jobs:
         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 }}
+          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 }}

.github/workflows/integration_tests.yml

@@ -22,9 +22,6 @@ jobs:
     steps:
       - name: Clone repository
         uses: actions/checkout@v4
-        with:
-          fetch-depth: 50
-
       - name: "Set up JDK 8"
         uses: actions/setup-java@v3
         with:

.github/workflows/multi_az_integration_tests.yml

@@ -0,0 +1,94 @@
+name: Integration Tests
+
+on:
+  workflow_dispatch:
+  push:
+    branches:
+      - main
+      - test/taz-integration-tests
+    paths-ignore:
+      - "**/*.md"
+      - "**/*.jpg"
+      - "**/README.txt"
+      - "**/LICENSE.txt"
+      - "docs/**"
+      - "ISSUE_TEMPLATE/**"
+      - "**/remove-old-artifacts.yml"
+
+jobs:
+  run-integration-tests:
+    name: Run Integration Tests
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        dbEngine: [ "multi-az-mysql", "multi-az-postgres" ]
+
+    steps:
+      - name: Clone repository
+        uses: actions/checkout@v4
+      - 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-${{ matrix.dbEngine }} --info
+        env:
+          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 }}
+
+      - 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 results
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: integration-report
+          path: ./tests/integration/container/reports
+          retention-days: 5

common/lib/database_dialect/database_dialect.ts

@@ -17,6 +17,7 @@
 import { HostListProvider } from "../host_list_provider/host_list_provider";
 import { HostListProviderService } from "../host_list_provider_service";
 import { ClientWrapper } from "../client_wrapper";
+import { FailoverRestriction } from "../plugins/failover/failover_restriction";
 
 export enum DatabaseType {
   MYSQL,
@@ -37,6 +38,7 @@ export interface DatabaseDialect {
   getConnectFunc(targetClient: any): () => Promise<any>;
   getDatabaseType(): DatabaseType;
   getDialectName(): string;
+  getFailoverRestrictions(): FailoverRestriction[];
   doesStatementSetReadOnly(statement: string): boolean | undefined;
   doesStatementSetTransactionIsolation(statement: string): number | undefined;
   doesStatementSetAutoCommit(statement: string): boolean | undefined;

common/lib/plugins/failover/failover_restriction.ts

@@ -15,5 +15,6 @@
 */
 
 export enum FailoverRestriction {
-  DISABLE_TASK_A
+  DISABLE_TASK_A,
+  ENABLE_WRITER_IN_TASK_B
 }

common/lib/plugins/failover/reader_failover_handler.ts

@@ -26,6 +26,7 @@ import { Messages } from "../../utils/messages";
 import { WrapperProperties } from "../../wrapper_property";
 import { ReaderTaskSelectorHandler } from "./reader_task_selector";
 import { uniqueId } from "lodash";
+import { FailoverRestriction } from "./failover_restriction";
 
 export interface ReaderFailoverHandler {
   failover(hosts: HostInfo[], currentHost: HostInfo): Promise<ReaderFailoverResult>;
@@ -216,9 +217,11 @@ export class ClusterAwareReaderFailoverHandler implements ReaderFailoverHandler
   getReaderHostsByPriority(hosts: HostInfo[]): HostInfo[] {
     const activeReaders: HostInfo[] = [];
     const downHostList: HostInfo[] = [];
+    let writerHost: HostInfo | null = null;
 
     hosts.forEach((host) => {
       if (host.role === HostRole.WRITER) {
+        writerHost = host;
         return;
       }
 
@@ -232,8 +235,15 @@ export class ClusterAwareReaderFailoverHandler implements ReaderFailoverHandler
     shuffleList(activeReaders);
     shuffleList(downHostList);
 
+    const numOfReaders = downHostList.length + activeReaders.length;
     const hostsByPriority: HostInfo[] = [...activeReaders];
     hostsByPriority.push(...downHostList);
+    if (
+      writerHost !== null &&
+      (numOfReaders === 0 || this.pluginService.getDialect().getFailoverRestrictions().includes(FailoverRestriction.ENABLE_WRITER_IN_TASK_B))
+    ) {
+      hostsByPriority.push(writerHost);
+    }
 
     return hostsByPriority;
   }

common/lib/plugins/failover/writer_failover_handler.ts

@@ -20,12 +20,13 @@ import { ClusterAwareReaderFailoverHandler } from "./reader_failover_handler";
 import { PluginService } from "../../plugin_service";
 import { HostAvailability } from "../../host_availability/host_availability";
 import { AwsWrapperError } from "../../utils/errors";
-import { getWriter, maskProperties } from "../../utils/utils";
+import { getWriter, logTopology, maskProperties } from "../../utils/utils";
 import { ReaderFailoverResult } from "./reader_failover_result";
 import { Messages } from "../../utils/messages";
 import { logger } from "../../../logutils";
 import { WrapperProperties } from "../../wrapper_property";
 import { ClientWrapper } from "../../client_wrapper";
+import { FailoverRestriction } from "./failover_restriction";
 
 export interface WriterFailoverHandler {
   failover(currentTopology: HostInfo[]): Promise<WriterFailoverResult>;
@@ -104,11 +105,14 @@ export class ClusterAwareWriterFailoverHandler implements WriterFailoverHandler
     const taskB = waitForNewWriterHandlerTask.call();
 
     let selectedTask = "";
-    const failoverTask = Promise.any([taskA, taskB])
+    const singleTask: boolean = this.pluginService.getDialect().getFailoverRestrictions().includes(FailoverRestriction.DISABLE_TASK_A);
+    const failoverTaskPromise = singleTask ? taskB : Promise.any([taskA, taskB]);
+
+    const failoverTask = failoverTaskPromise
       .then((result) => {
         selectedTask = result.taskName;
         // If the first resolved promise is connected or has an error, return it.
-        if (result.isConnected || result.exception) {
+        if (result.isConnected || result.exception || singleTask) {
           return result;
         }
 
@@ -369,6 +373,8 @@ class WaitForNewWriterHandlerTask {
   }
 
   async refreshTopologyAndConnectToNewWriter(): Promise<boolean> {
+    const allowOldWriter: boolean = this.pluginService.getDialect().getFailoverRestrictions().includes(FailoverRestriction.ENABLE_WRITER_IN_TASK_B);
+
     while (this.pluginService.getCurrentClient() && Date.now() < this.endTime && !this.failoverCompleted) {
       try {
         if (this.currentReaderTargetClient) {
@@ -390,8 +396,9 @@ class WaitForNewWriterHandlerTask {
           } else {
             this.currentTopology = topology;
             const writerCandidate = getWriter(this.currentTopology);
-            if (writerCandidate && !this.isSame(writerCandidate, this.originalWriterHost)) {
+            if (writerCandidate && (allowOldWriter || !this.isSame(writerCandidate, this.originalWriterHost))) {
               // new writer is available, and it's different from the previous writer
+              logger.debug(logTopology(this.currentTopology, "[Task B] "));
               if (await this.connectToWriter(writerCandidate)) {
                 return true;
               }

mysql/lib/dialect/mysql_database_dialect.ts

@@ -23,6 +23,7 @@ import { DatabaseDialectCodes } from "../../../common/lib/database_dialect/datab
 import { TransactionIsolationLevel } from "../../../common/lib/utils/transaction_isolation_level";
 import { ClientWrapper } from "../../../common/lib/client_wrapper";
 import { ClientUtils } from "../../../common/lib/utils/client_utils";
+import { FailoverRestriction } from "../../../common/lib/plugins/failover/failover_restriction";
 
 export class MySQLDatabaseDialect implements DatabaseDialect {
   protected dialectName: string = this.constructor.name;
@@ -118,6 +119,10 @@ export class MySQLDatabaseDialect implements DatabaseDialect {
     return this.dialectName;
   }
 
+  getFailoverRestrictions(): FailoverRestriction[] {
+    return [];
+  }
+
   doesStatementSetReadOnly(statement: string): boolean | undefined {
     if (statement.includes("set session transaction read only")) {
       return true;

mysql/lib/dialect/rds_multi_az_mysql_database_dialect.ts

@@ -24,10 +24,8 @@ import { Messages } from "../../../common/lib/utils/messages";
 import { logger } from "../../../common/logutils";
 import { AwsWrapperError } from "../../../common/lib/utils/errors";
 import { TopologyAwareDatabaseDialect } from "../../../common/lib/topology_aware_database_dialect";
-import { HostAvailability } from "../../../common/lib/host_availability/host_availability";
-import { HostInfoBuilder } from "../../../common/lib/host_info_builder";
-import { SimpleHostAvailabilityStrategy } from "../../../common/lib/host_availability/simple_host_availability_strategy";
 import { RdsHostListProvider } from "../../../common/lib/host_list_provider/rds_host_list_provider";
+import { FailoverRestriction } from "../../../common/lib/plugins/failover/failover_restriction";
 
 export class RdsMultiAZMySQLDatabaseDialect extends MySQLDatabaseDialect implements TopologyAwareDatabaseDialect {
   private static readonly TOPOLOGY_QUERY: string = "SELECT id, endpoint, port FROM mysql.rds_topology";
@@ -137,4 +135,8 @@ export class RdsMultiAZMySQLDatabaseDialect extends MySQLDatabaseDialect impleme
   getDialectUpdateCandidates(): string[] {
     return [];
   }
+
+  getFailoverRestrictions(): FailoverRestriction[] {
+    return [FailoverRestriction.DISABLE_TASK_A, FailoverRestriction.ENABLE_WRITER_IN_TASK_B];
+  }
 }

pg/lib/dialect/pg_database_dialect.ts

@@ -22,6 +22,7 @@ import { AwsWrapperError } from "../../../common/lib/utils/errors";
 import { DatabaseDialectCodes } from "../../../common/lib/database_dialect/database_dialect_codes";
 import { TransactionIsolationLevel } from "../../../common/lib/utils/transaction_isolation_level";
 import { ClientWrapper } from "../../../common/lib/client_wrapper";
+import { FailoverRestriction } from "../../../common/lib/plugins/failover/failover_restriction";
 
 export class PgDatabaseDialect implements DatabaseDialect {
   protected dialectName: string = this.constructor.name;
@@ -104,6 +105,10 @@ export class PgDatabaseDialect implements DatabaseDialect {
     return this.dialectName;
   }
 
+  getFailoverRestrictions(): FailoverRestriction[] {
+    return [];
+  }
+
   doesStatementSetAutoCommit(statement: string): boolean | undefined {
     return undefined;
   }

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

@@ -58,16 +58,15 @@ describe("aurora failover", () => {
     logger.info(`Test started: ${expect.getState().currentTestName}`);
     env = await TestEnvironment.getCurrent();
 
-    auroraTestUtility = new AuroraTestUtility(env.auroraRegion);
+    auroraTestUtility = new AuroraTestUtility(env.region);
     driver = DriverHelper.getDriverForDatabaseEngine(env.engine);
     initClientFunc = DriverHelper.getClient(driver);
     await ProxyHelper.enableAllConnectivity();
-    await TestEnvironment.updateWriter();
+    await TestEnvironment.verifyClusterStatus();
 
     client = null;
     secondaryClient = null;
-    await TestEnvironment.updateWriter();
-  });
+  }, 1000000);
 
   afterEach(async () => {
     if (client !== null) {

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

@@ -22,27 +22,29 @@ import { logger } from "../../../../common/logutils";
 import { DatabaseEngine } from "./utils/database_engine";
 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) ? it : it;
 
 let client: any;
 let auroraTestUtility: AuroraTestUtility;
 
-async function executeInstanceQuery(client: any, engine: DatabaseEngine, props: any): Promise<void> {
+async function executeInstanceQuery(client: any, engine: DatabaseEngine, deployment: DatabaseEngineDeployment, props: any): Promise<void> {
   client.on("error", (error: any) => {
     logger.debug(error.message);
   });
   await client.connect();
 
-  const res = await DriverHelper.executeInstanceQuery(engine, client);
+  const res = await DriverHelper.executeInstanceQuery(engine, deployment, client);
 
   expect(res).not.toBeNull();
 }
 
 beforeEach(async () => {
   logger.info(`Test started: ${expect.getState().currentTestName}`);
-  auroraTestUtility = new AuroraTestUtility((await TestEnvironment.getCurrent()).auroraRegion);
+  auroraTestUtility = new AuroraTestUtility((await TestEnvironment.getCurrent()).region);
   await ProxyHelper.enableAllConnectivity();
+  await TestEnvironment.verifyClusterStatus();
   client = null;
 });
 
@@ -76,7 +78,7 @@ describe("basic_connectivity", () => {
       props = DriverHelper.addDriverSpecificConfiguration(props, env.engine);
       client = initClientFunc(props);
 
-      await executeInstanceQuery(client, env.engine, props);
+      await executeInstanceQuery(client, env.engine, env.deployment, props);
     },
     1000000
   );
@@ -99,7 +101,7 @@ describe("basic_connectivity", () => {
       props = DriverHelper.addDriverSpecificConfiguration(props, env.engine);
 
       client = initClientFunc(props);
-      await executeInstanceQuery(client, env.engine, props);
+      await executeInstanceQuery(client, env.engine, env.deployment, props);
     },
     1000000
   );
@@ -122,7 +124,7 @@ describe("basic_connectivity", () => {
       props = DriverHelper.addDriverSpecificConfiguration(props, env.engine);
 
       client = initClientFunc(props);
-      await executeInstanceQuery(client, env.engine, props);
+      await executeInstanceQuery(client, env.engine, env.deployment, props);
     },
     1000000
   );
@@ -150,7 +152,7 @@ describe("basic_connectivity", () => {
       });
       await client.connect();
 
-      const res = await DriverHelper.executeInstanceQuery(env.engine, client);
+      const res = await DriverHelper.executeInstanceQuery(env.engine, env.deployment, client);
 
       expect(res).not.toBeNull();
     },

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

@@ -81,7 +81,7 @@ describe("iam authentication", () => {
   });
 
   afterEach(async () => {
-    await TestEnvironment.updateWriter();
+    await TestEnvironment.verifyClusterStatus();
     logger.info(`Test finished: ${expect.getState().currentTestName}`);
   }, 1000000);