forked from airbytehq/airbyte-platform
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
use cloudsql on stage acceptance tests (#11894)
- Loading branch information
Showing
15 changed files
with
357 additions
and
95 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,20 @@ | ||
| # airbyte-test-utils | ||
|
|
||
| Shared Java code for executing TestContainers and other helpers. | ||
|
|
||
| ## Stage databases setup | ||
|
|
||
| When we run acceptance tests on an environment that is not `stage`, a test container will be used for each connector that requires a database. Each test container will be used for only one test case, and it will be deleted once the test case completes. | ||
|
|
||
| When we run acceptance tests on stage, things are slightly more complex, but we try to have the same behavior. Instead of using a test container for each connector that requires a database, we will use a CloudSQL database for each connector. Similarly to the test containers, each CloudSQL database will be used for only one test case, and it will be deleted once the test case completes. | ||
|
|
||
| It's important to understand how are the different components communicating when running on stage. | ||
|
|
||
|  | ||
|
|
||
| - It is possible to communicate with the `CloudSQL Instance` from both private IP and public ip | ||
| - One same `CloudSQL Instance` is use for all the tests, but each test case will create their own databases inside this instance. | ||
| - We run the acceptance tests from a `AWS Test Runner` (EC2 instances), which are behind Tailscale, so they can communicate with the CloudSQL instance using its private IP. We need to be able to access the CloudSQL instance from these test runners since the tests will access these databases to validate their content. | ||
| - The only IPs that are allowed to connect to the CloudSQL instance via its public IP are the ones that belong to stage Dataplanes (both `GCP Dataplane` and `AWS Dataplane`). Note that this is not a workaround for the sake of our tests, this is the same setup that real users have. | ||
|
|
||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
109 changes: 109 additions & 0 deletions
109
airbyte-test-utils/src/main/java/io/airbyte/test/utils/CloudSqlDatabaseProvisioner.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| /* | ||
| * Copyright (c) 2020-2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.test.utils; | ||
|
|
||
| import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport; | ||
| import com.google.api.client.googleapis.json.GoogleJsonResponseException; | ||
| import com.google.api.client.json.gson.GsonFactory; | ||
| import com.google.api.services.sqladmin.SQLAdmin; | ||
| import com.google.api.services.sqladmin.model.Database; | ||
| import com.google.api.services.sqladmin.model.Operation; | ||
| import com.google.auth.http.HttpCredentialsAdapter; | ||
| import com.google.auth.oauth2.GoogleCredentials; | ||
| import com.google.common.annotations.VisibleForTesting; | ||
| import java.io.IOException; | ||
| import java.security.GeneralSecurityException; | ||
| import java.util.concurrent.Callable; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * Creates and deletes GCP CloudSQL databases. | ||
| */ | ||
| public class CloudSqlDatabaseProvisioner { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(CloudSqlDatabaseProvisioner.class); | ||
|
|
||
| private static final String SQL_OPERATION_DONE_STATUS = "DONE"; | ||
| private static final int DEFAULT_MAX_POLL_ATTEMPTS = 10; | ||
| private static final int DEFAULT_MAX_API_CALL_ATTEMPTS = 10; | ||
| private static final String APPLICATION_NAME = "cloud-sql-database-provisioner"; | ||
|
|
||
| private final SQLAdmin sqlAdmin; | ||
| private final int maxPollAttempts; | ||
| private final int maxApiCallAttempts; | ||
|
|
||
| @VisibleForTesting | ||
| CloudSqlDatabaseProvisioner(SQLAdmin sqlAdmin, int maxPollAttempts, int maxApiCallAttempts) { | ||
| this.sqlAdmin = sqlAdmin; | ||
| this.maxPollAttempts = maxPollAttempts; | ||
| this.maxApiCallAttempts = maxApiCallAttempts; | ||
| } | ||
|
|
||
| public CloudSqlDatabaseProvisioner() throws GeneralSecurityException, IOException { | ||
| this.sqlAdmin = new SQLAdmin.Builder( | ||
| GoogleNetHttpTransport.newTrustedTransport(), | ||
| GsonFactory.getDefaultInstance(), | ||
| new HttpCredentialsAdapter(GoogleCredentials.getApplicationDefault())).setApplicationName(APPLICATION_NAME).build(); | ||
| this.maxPollAttempts = DEFAULT_MAX_POLL_ATTEMPTS; | ||
| this.maxApiCallAttempts = DEFAULT_MAX_API_CALL_ATTEMPTS; | ||
| } | ||
|
|
||
| public synchronized String createDatabase(String projectId, String instanceId, String databaseName) throws IOException, InterruptedException { | ||
| Database database = new Database().setName(databaseName); | ||
| Operation operation = runWithRetry(() -> sqlAdmin.databases().insert(projectId, instanceId, database).execute()); | ||
| pollOperation(projectId, operation.getName()); | ||
|
|
||
| return databaseName; | ||
| } | ||
|
|
||
| public synchronized void deleteDatabase(String projectId, String instanceId, String databaseName) throws IOException, InterruptedException { | ||
| Operation operation = runWithRetry(() -> sqlAdmin.databases().delete(projectId, instanceId, databaseName).execute()); | ||
| pollOperation(projectId, operation.getName()); | ||
| } | ||
|
|
||
| /** | ||
| * Database operations are asynchronous. This method polls the operation until it is done. | ||
| */ | ||
| @VisibleForTesting | ||
| void pollOperation(String projectId, String operationName) throws IOException, InterruptedException { | ||
| int pollAttempts = 0; | ||
| while (pollAttempts < maxPollAttempts) { | ||
| Operation operation = sqlAdmin.operations().get(projectId, operationName).execute(); | ||
| if (operation.getStatus().equals(SQL_OPERATION_DONE_STATUS)) { | ||
| return; | ||
| } | ||
| Thread.sleep(1000); | ||
| pollAttempts += 1; | ||
| } | ||
|
|
||
| throw new RuntimeException("Operation " + operationName + " did not complete successfully"); | ||
| } | ||
|
|
||
| /** | ||
| * If there's another operation already in progress in one same cloudsql instance then the api will | ||
| * return a 409 error. This method will retry api calls that return a 409 error. | ||
| */ | ||
| @VisibleForTesting | ||
| Operation runWithRetry(Callable<Operation> callable) throws InterruptedException { | ||
| int attempts = 0; | ||
| while (attempts < maxApiCallAttempts) { | ||
| try { | ||
| return callable.call(); | ||
| } catch (Exception e) { | ||
| if (e instanceof GoogleJsonResponseException && ((GoogleJsonResponseException) e).getStatusCode() == 409) { | ||
| attempts++; | ||
| LOGGER.info("Attempt " + attempts + " failed with 409 error"); | ||
| LOGGER.info("Exception thrown by API: " + e.getMessage()); | ||
| Thread.sleep(1000); | ||
| } else { | ||
| throw new RuntimeException(e); | ||
| } | ||
| } | ||
| } | ||
| throw new RuntimeException("Max retries exceeded. Could not complete operation."); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.