diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/ca81ee7c-3163-4246-af40-094cc31e5e42.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/ca81ee7c-3163-4246-af40-094cc31e5e42.json index 90eec451a613..33b45e554964 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/ca81ee7c-3163-4246-af40-094cc31e5e42.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/ca81ee7c-3163-4246-af40-094cc31e5e42.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "ca81ee7c-3163-4246-af40-094cc31e5e42", "name": "MySQL", "dockerRepository": "airbyte/destination-mysql", - "dockerImageTag": "0.1.11", + "dockerImageTag": "0.1.12", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/mysql" } diff --git a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml index e8d413893af7..88a6b662c225 100644 --- a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml +++ b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml @@ -63,7 +63,7 @@ - destinationDefinitionId: ca81ee7c-3163-4246-af40-094cc31e5e42 name: MySQL dockerRepository: airbyte/destination-mysql - dockerImageTag: 0.1.11 + dockerImageTag: 0.1.12 documentationUrl: https://docs.airbyte.io/integrations/destinations/mysql - destinationDefinitionId: d4353156-9217-4cad-8dd7-c108fd4f74cf name: MS SQL Server diff --git a/airbyte-integrations/bases/base-normalization/Dockerfile b/airbyte-integrations/bases/base-normalization/Dockerfile index f899ec458712..eaefd09570d2 100644 --- a/airbyte-integrations/bases/base-normalization/Dockerfile +++ b/airbyte-integrations/bases/base-normalization/Dockerfile @@ -45,5 +45,5 @@ WORKDIR /airbyte ENV AIRBYTE_ENTRYPOINT "/airbyte/entrypoint.sh" ENTRYPOINT ["/airbyte/entrypoint.sh"] -LABEL io.airbyte.version=0.1.46 +LABEL io.airbyte.version=0.1.47 LABEL io.airbyte.name=airbyte/normalization diff --git a/airbyte-integrations/bases/base-normalization/normalization/transform_config/transform.py b/airbyte-integrations/bases/base-normalization/normalization/transform_config/transform.py index daeb24df4a28..af0b6006347b 100644 --- a/airbyte-integrations/bases/base-normalization/normalization/transform_config/transform.py +++ b/airbyte-integrations/bases/base-normalization/normalization/transform_config/transform.py @@ -217,6 +217,10 @@ def transform_snowflake(config: Dict[str, Any]): @staticmethod def transform_mysql(config: Dict[str, Any]): print("transform_mysql") + + if TransformConfig.is_ssh_tunnelling(config): + config = TransformConfig.get_ssh_altered_config(config, port_key="port", host_key="host") + # https://github.com/dbeatty10/dbt-mysql#configuring-your-profile dbt_config = { # MySQL 8.x - type: mysql diff --git a/airbyte-integrations/connectors/destination-mysql/Dockerfile b/airbyte-integrations/connectors/destination-mysql/Dockerfile index 0d287509faff..ec5a103b4228 100644 --- a/airbyte-integrations/connectors/destination-mysql/Dockerfile +++ b/airbyte-integrations/connectors/destination-mysql/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.1.11 +LABEL io.airbyte.version=0.1.12 LABEL io.airbyte.name=airbyte/destination-mysql diff --git a/airbyte-integrations/connectors/destination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java b/airbyte-integrations/connectors/destination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java index cd46edaeec78..b08dcb49cd55 100644 --- a/airbyte-integrations/connectors/destination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java +++ b/airbyte-integrations/connectors/destination-mysql/src/main/java/io/airbyte/integrations/destination/mysql/MySQLDestination.java @@ -11,16 +11,20 @@ import io.airbyte.db.jdbc.JdbcDatabase; import io.airbyte.integrations.base.Destination; import io.airbyte.integrations.base.IntegrationRunner; +import io.airbyte.integrations.base.ssh.SshWrappedDestination; import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination; import io.airbyte.integrations.destination.mysql.MySQLSqlOperations.VersionCompatibility; import io.airbyte.protocol.models.AirbyteConnectionStatus; import io.airbyte.protocol.models.AirbyteConnectionStatus.Status; +import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class MySQLDestination extends AbstractJdbcDestination implements Destination { private static final Logger LOGGER = LoggerFactory.getLogger(MySQLDestination.class); + public static final List HOST_KEY = List.of("host"); + public static final List PORT_KEY = List.of("port"); public static final String DRIVER_CLASS = "com.mysql.cj.jdbc.Driver"; @@ -69,14 +73,12 @@ protected JdbcDatabase getDatabase(JsonNode config) { @Override public JsonNode toJdbcConfig(JsonNode config) { - final StringBuilder jdbcUrl = new StringBuilder(String.format("jdbc:mysql://%s:%s/%s", - config.get("host").asText(), - config.get("port").asText(), - config.get("database").asText())); - ImmutableMap.Builder configBuilder = ImmutableMap.builder() .put("username", config.get("username").asText()) - .put("jdbc_url", jdbcUrl.toString()); + .put("jdbc_url", String.format("jdbc:mysql://%s:%s/%s", + config.get("host").asText(), + config.get("port").asText(), + config.get("database").asText())); if (config.has("password")) { configBuilder.put("password", config.get("password").asText()); @@ -86,7 +88,7 @@ public JsonNode toJdbcConfig(JsonNode config) { } public static void main(String[] args) throws Exception { - final Destination destination = new MySQLDestination(); + final Destination destination = new SshWrappedDestination(new MySQLDestination(), HOST_KEY, PORT_KEY); LOGGER.info("starting destination: {}", MySQLDestination.class); new IntegrationRunner(destination).run(args); LOGGER.info("completed destination: {}", MySQLDestination.class); diff --git a/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshKeyMySQLDestinationAcceptanceTest.java b/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshKeyMySQLDestinationAcceptanceTest.java new file mode 100644 index 000000000000..7b62d89d2383 --- /dev/null +++ b/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshKeyMySQLDestinationAcceptanceTest.java @@ -0,0 +1,36 @@ +/* + * MIT License + * + * Copyright (c) 2020 Airbyte + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package io.airbyte.integrations.destination.mysql; + +import java.nio.file.Path; + +public class SshKeyMySQLDestinationAcceptanceTest extends SshMySQLDestinationAcceptanceTest { + + @Override + public Path getConfigFilePath() { + return Path.of("secrets/ssh-key-config.json"); + } + +} diff --git a/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshMySQLDestinationAcceptanceTest.java b/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshMySQLDestinationAcceptanceTest.java new file mode 100644 index 000000000000..a2a6acc5ecb6 --- /dev/null +++ b/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshMySQLDestinationAcceptanceTest.java @@ -0,0 +1,192 @@ +/* + * MIT License + * + * Copyright (c) 2020 Airbyte + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package io.airbyte.integrations.destination.mysql; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import io.airbyte.commons.functional.CheckedFunction; +import io.airbyte.commons.io.IOs; +import io.airbyte.commons.json.Jsons; +import io.airbyte.db.Database; +import io.airbyte.db.Databases; +import io.airbyte.integrations.base.JavaBaseConstants; +import io.airbyte.integrations.base.ssh.SshTunnel; +import io.airbyte.integrations.destination.ExtendedNameTransformer; +import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.commons.lang3.RandomStringUtils; +import org.jooq.JSONFormat; + +/** + * Abstract class that allows us to avoid duplicating testing logic for testing SSH with a key file + * or with a password. + */ +public abstract class SshMySQLDestinationAcceptanceTest extends DestinationAcceptanceTest { + + private static final JSONFormat JSON_FORMAT = new JSONFormat().recordFormat(JSONFormat.RecordFormat.OBJECT); + + private final ExtendedNameTransformer namingResolver = new MySQLNameTransformer(); + private String schemaName; + + public abstract Path getConfigFilePath(); + + @Override + protected String getImageName() { + return "airbyte/destination-mysql:dev"; + } + + @Override + protected JsonNode getConfig() { + var config = getConfigFromSecretsFile(); + ((ObjectNode) config).put("database", schemaName); + return config; + } + + private JsonNode getConfigFromSecretsFile() { + return Jsons.deserialize(IOs.readFile(getConfigFilePath())); + } + + @Override + protected JsonNode getFailCheckConfig() { + final JsonNode clone = Jsons.clone(getConfig()); + ((ObjectNode) clone).put("password", "wrong password"); + return clone; + } + + @Override + protected List retrieveRecords(final TestDestinationEnv env, + final String streamName, + final String namespace, + final JsonNode streamSchema) + throws Exception { + return retrieveRecordsFromTable(namingResolver.getRawTableName(streamName), namespace) + .stream() + .map(r -> Jsons.deserialize(r.get(JavaBaseConstants.COLUMN_NAME_DATA).asText())) + .collect(Collectors.toList()); + } + + @Override + protected boolean supportsNormalization() { + return true; + } + + @Override + protected boolean supportsDBT() { + return true; + } + + @Override + protected boolean implementsNamespaces() { + return true; + } + + @Override + protected List retrieveNormalizedRecords(final TestDestinationEnv env, + final String streamName, + final String namespace) + throws Exception { + var tableName = namingResolver.getIdentifier(streamName); + String schema = namespace != null ? namingResolver.getIdentifier(namespace) : namingResolver.getIdentifier(schemaName); + return retrieveRecordsFromTable(tableName, schema); + } + + @Override + protected List resolveIdentifier(final String identifier) { + final List result = new ArrayList<>(); + final String resolved = namingResolver.getIdentifier(identifier); + result.add(identifier); + result.add(resolved); + if (!resolved.startsWith("\"")) { + result.add(resolved.toLowerCase()); + result.add(resolved.toUpperCase()); + } + return result; + } + + private static Database getDatabaseFromConfig(final JsonNode config) { + return Databases.createMySqlDatabase( + config.get("username").asText(), + config.get("password").asText(), + String.format("jdbc:mysql://%s:%s", + config.get("host").asText(), + config.get("port").asText())); + } + + private List retrieveRecordsFromTable(String tableName, String schemaName) throws Exception { + var schema = schemaName == null ? this.schemaName : schemaName; + return SshTunnel.sshWrap( + getConfig(), + MySQLDestination.HOST_KEY, + MySQLDestination.PORT_KEY, + (CheckedFunction, Exception>) mangledConfig -> getDatabaseFromConfig(mangledConfig) + .query( + ctx -> ctx + .fetch(String.format("SELECT * FROM %s.%s ORDER BY %s ASC;", schema, tableName.toLowerCase(), + JavaBaseConstants.COLUMN_NAME_EMITTED_AT)) + .stream() + .map(r -> r.formatJSON(JSON_FORMAT)) + .map(Jsons::deserialize) + .collect(Collectors.toList()))); + } + + @Override + protected void setup(final TestDestinationEnv testEnv) throws Exception { + schemaName = RandomStringUtils.randomAlphabetic(8).toLowerCase(); + var config = getConfig(); + SshTunnel.sshWrap( + config, + MySQLDestination.HOST_KEY, + MySQLDestination.PORT_KEY, + mangledConfig -> { + getDatabaseFromConfig(mangledConfig).query(ctx -> ctx.fetch(String.format("CREATE DATABASE %s;", schemaName))); + }); + } + + @Override + protected void tearDown(final TestDestinationEnv testEnv) throws Exception { + SshTunnel.sshWrap( + getConfig(), + MySQLDestination.HOST_KEY, + MySQLDestination.PORT_KEY, + mangledConfig -> { + getDatabaseFromConfig(mangledConfig).query(ctx -> ctx.fetch(String.format("DROP DATABASE %s", schemaName))); + }); + } + + protected void assertSameValue(JsonNode expectedValue, JsonNode actualValue) { + if (expectedValue.isBoolean()) { + // Boolean in MySQL are stored as TINYINT (0 or 1) so we force them to boolean values here + assertEquals(expectedValue.asBoolean(), actualValue.asBoolean()); + } else { + assertEquals(expectedValue, actualValue); + } + } + +} diff --git a/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshPasswordMySQLDestinationAcceptanceTest.java b/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshPasswordMySQLDestinationAcceptanceTest.java new file mode 100644 index 000000000000..5f38ad436316 --- /dev/null +++ b/airbyte-integrations/connectors/destination-mysql/src/test-integration/java/io/airbyte/integrations/destination/mysql/SshPasswordMySQLDestinationAcceptanceTest.java @@ -0,0 +1,36 @@ +/* + * MIT License + * + * Copyright (c) 2020 Airbyte + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package io.airbyte.integrations.destination.mysql; + +import java.nio.file.Path; + +public class SshPasswordMySQLDestinationAcceptanceTest extends SshMySQLDestinationAcceptanceTest { + + @Override + public Path getConfigFilePath() { + return Path.of("secrets/ssh-pwd-config.json"); + } + +} diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java b/airbyte-workers/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java index 21afb0a03750..5193303ea721 100644 --- a/airbyte-workers/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java +++ b/airbyte-workers/src/main/java/io/airbyte/workers/normalization/DefaultNormalizationRunner.java @@ -28,7 +28,7 @@ public class DefaultNormalizationRunner implements NormalizationRunner { private static final Logger LOGGER = LoggerFactory.getLogger(DefaultNormalizationRunner.class); - public static final String NORMALIZATION_IMAGE_NAME = "airbyte/normalization:0.1.46"; + public static final String NORMALIZATION_IMAGE_NAME = "airbyte/normalization:0.1.47"; private final DestinationType destinationType; private final ProcessFactory processFactory; diff --git a/docs/integrations/destinations/mysql.md b/docs/integrations/destinations/mysql.md index 2796c8f7d738..b37382e92f2f 100644 --- a/docs/integrations/destinations/mysql.md +++ b/docs/integrations/destinations/mysql.md @@ -70,10 +70,28 @@ One of their recommendations is: As a result, Airbyte MySQL destination forces all identifier (table, schema and columns) names to be lowercase. +## Connection to MySQL via an SSH Tunnel + +Airbyte has the ability to connect to a MySQl instance via an SSH Tunnel. The reason you might want to do this because it is not possible (or against security policy) to connect to the database directly (e.g. it does not have a public IP address). + +When using an SSH tunnel, you are configuring Airbyte to connect to an intermediate server (a.k.a. a bastion sever) that _does_ have direct access to the database. Airbyte connects to the bastion and then asks the bastion to connect directly to the server. + +Using this feature requires additional configuration, when creating the destination. We will talk through what each piece of configuration means. +1. Configure all fields for the destination as you normally would, except `SSH Tunnel Method`. +2. `SSH Tunnel Method` defaults to `No Tunnel` (meaning a direct connection). If you want to use an SSH Tunnel choose `SSH Key Authentication` or `Password Authentication`. + 1. Choose `Key Authentication` if you will be using an RSA private key as your secret for establishing the SSH Tunnel (see below for more information on generating this key). + 2. Choose `Password Authentication` if you will be using a password as your secret for establishing the SSH Tunnel. +3. `SSH Tunnel Jump Server Host` refers to the intermediate (bastion) server that Airbyte will connect to. This should be a hostname or an IP Address. +4. `SSH Connection Port` is the port on the bastion server with which to make the SSH connection. The default port for SSH connections is `22`, so unless you have explicitly changed something, go with the default. +5. `SSH Login Username` is the username that Airbyte should use when connection to the bastion server. This is NOT the MySQl username. +6. If you are using `Password Authentication`, then `SSH Login Username` should be set to the password of the User from the previous step. If you are using `SSH Key Authentication` leave this blank. Again, this is not the MySQl password, but the password for the OS-user that Airbyte is using to perform commands on the bastion. +7. If you are using `SSH Key Authentication`, then `SSH Private Key` should be set to the RSA Private Key that you are using to create the SSH connection. This should be the full contents of the key file starting with `-----BEGIN RSA PRIVATE KEY-----` and ending with `-----END RSA PRIVATE KEY-----`. + ## CHANGELOG | Version | Date | Pull Request | Subject | | :--- | :--- | :--- | :--- | +| 0.1.12 | 2021-09-24 | [#6317](https://github.com/airbytehq/airbyte/pull/6317) | Added option to connect to DB via SSH | | 0.1.11 | 2021-07-30 | [#5125](https://github.com/airbytehq/airbyte/pull/5125) | Enable `additionalPropertities` in spec.json | | 0.1.10 | 2021-07-28 | [#5026](https://github.com/airbytehq/airbyte/pull/5026) | Add sanitized json fields in raw tables to handle quotes in column names | | 0.1.7 | 2021-07-09 | [#4651](https://github.com/airbytehq/airbyte/pull/4651) | Switch normalization flag on so users can use normalization. | diff --git a/docs/understanding-airbyte/basic-normalization.md b/docs/understanding-airbyte/basic-normalization.md index 47e0f004aa8b..ad4d4c27587b 100644 --- a/docs/understanding-airbyte/basic-normalization.md +++ b/docs/understanding-airbyte/basic-normalization.md @@ -294,6 +294,7 @@ Therefore, in order to "upgrade" to the desired normalization version, you need | Airbyte Version | Normalization Version | Date | Pull Request | Subject | | :--- | :--- | :--- | :--- | :--- | +| 0.29.17-alpha | 0.1.47 | 2021-09-20 | [#6317](https://github.com/airbytehq/airbyte/pull/6317) | MySQL: updated MySQL normalization with using SSH tunnel | | 0.29.17-alpha | 0.1.45 | 2021-09-18 | [#6052](https://github.com/airbytehq/airbyte/pull/6052) | Snowflake: accept any date-time format | | 0.29.8-alpha | 0.1.40 | 2021-08-18 | [#5433](https://github.com/airbytehq/airbyte/pull/5433) | Allow optional credentials_json for BigQuery | | 0.29.5-alpha | 0.1.39 | 2021-08-11 | [#4557](https://github.com/airbytehq/airbyte/pull/4557) | Handle date times and solve conflict name btw stream/field | diff --git a/tools/bin/ci_credentials.sh b/tools/bin/ci_credentials.sh index 324271012f20..652dd7aa9afb 100755 --- a/tools/bin/ci_credentials.sh +++ b/tools/bin/ci_credentials.sh @@ -30,6 +30,8 @@ write_standard_creds destination-kvdb "$DESTINATION_KVDB_TEST_CREDS" write_standard_creds destination-keen "$DESTINATION_KEEN_TEST_CREDS" write_standard_creds destination-postgres "$DESTINATION_PUBSUB_TEST_CREDS" "credentials.json" +write_standard_creds destination-mysql "$MYSQL_SSH_KEY_TEST_CREDS" "ssh-key-config.json" +write_standard_creds destination-mysql "$MYSQL_SSH_PWD_TEST_CREDS" "ssh-pwd-config.json" write_standard_creds destination-pubsub "$DESTINATION_PUBSUB_TEST_CREDS" "credentials.json" write_standard_creds destination-redshift "$AWS_REDSHIFT_INTEGRATION_TEST_CREDS" write_standard_creds destination-dynamodb "$DESTINATION_DYNAMODB_TEST_CREDS"