diff --git a/it/jdbc/src/main/java/org/apache/beam/it/jdbc/SSLMySQLResourceManager.java b/it/jdbc/src/main/java/org/apache/beam/it/jdbc/SSLMySQLResourceManager.java new file mode 100644 index 0000000000..ad50a16c9e --- /dev/null +++ b/it/jdbc/src/main/java/org/apache/beam/it/jdbc/SSLMySQLResourceManager.java @@ -0,0 +1,212 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.it.jdbc; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Comparator; +import java.util.stream.Stream; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.Container.ExecResult; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * Custom MySQL resource manager that enables SSL and dynamically extracts the ephemeral + * certificates generated by the MySQL daemon at startup, converting them into JKS files that can be + * uploaded to GCS for the Dataflow pipeline to use. + */ +public class SSLMySQLResourceManager extends AbstractJDBCResourceManager> { + + private static final Logger LOG = LoggerFactory.getLogger(SSLMySQLResourceManager.class); + private static final String DEFAULT_MYSQL_CONTAINER_NAME = "mysql"; + private static final String DEFAULT_MYSQL_CONTAINER_TAG = "8.0.30"; + + private final MySQLContainer container; + + private String keystorePath; + private String truststorePath; + private Path tempDir; + + private SSLMySQLResourceManager(MySQLContainer container, Builder builder) { + super(container, builder); + this.container = container; + try { + generateAndExtractJKS(); + } catch (Exception e) { + throw new RuntimeException("Failed to generate and extract SSL certificates", e); + } + } + + public static Builder builder(String testId) { + return new Builder(testId); + } + + @Override + protected int getJDBCPort() { + return MySQLContainer.MYSQL_PORT; + } + + @Override + public String getJDBCPrefix() { + return "mysql"; + } + + @Override + public void cleanupAll() { + try { + super.cleanupAll(); + } finally { + if (tempDir != null && Files.exists(tempDir)) { + try (Stream walk = Files.walk(tempDir)) { + walk.sorted(Comparator.reverseOrder()).map(Path::toFile).forEach(File::delete); + LOG.info("Cleaned up temporary directory: {}", tempDir); + } catch (IOException e) { + LOG.warn("Failed to clean up temporary directory: {}", tempDir, e); + } + } + } + } + + private void generateAndExtractJKS() throws IOException, InterruptedException { + String password = getPassword(); + LOG.info("Generating PKCS12 keystore inside MySQL container..."); + ExecResult result = + container.execInContainer( + "openssl", + "pkcs12", + "-export", + "-in", + "/var/lib/mysql/client-cert.pem", + "-inkey", + "/var/lib/mysql/client-key.pem", + "-out", + "/var/lib/mysql/keystore.p12", + "-name", + "mysqlclient", + "-password", + "pass:" + password); + + if (result.getExitCode() != 0) { + throw new RuntimeException( + "Failed to generate pkcs12 inside container: " + result.getStderr()); + } + + this.tempDir = Files.createTempDirectory("mysql-ssl-"); + + LOG.info("Copying certificates out of container to {}", tempDir); + container.copyFileFromContainer("/var/lib/mysql/ca.pem", tempDir.resolve("ca.pem").toString()); + container.copyFileFromContainer( + "/var/lib/mysql/keystore.p12", tempDir.resolve("keystore.p12").toString()); + + String keytool = Paths.get(System.getProperty("java.home"), "bin", "keytool").toString(); + + // Create Truststore JKS + Path truststore = tempDir.resolve("truststore.jks"); + LOG.info("Generating truststore.jks"); + ProcessBuilder pbTrust = + new ProcessBuilder( + keytool, + "-importcert", + "-alias", + "mysqlca", + "-file", + tempDir.resolve("ca.pem").toString(), + "-keystore", + truststore.toString(), + "-storepass", + password, + "-noprompt"); + runProcess(pbTrust, "Failed to generate truststore.jks"); + + // Convert PKCS12 to Keystore JKS + Path keystore = tempDir.resolve("keystore.jks"); + LOG.info("Generating keystore.jks"); + ProcessBuilder pbKey = + new ProcessBuilder( + keytool, + "-importkeystore", + "-srckeystore", + tempDir.resolve("keystore.p12").toString(), + "-srcstoretype", + "PKCS12", + "-srcstorepass", + password, + "-destkeystore", + keystore.toString(), + "-deststoretype", + "JKS", + "-deststorepass", + password, + "-noprompt"); + runProcess(pbKey, "Failed to generate keystore.jks"); + + this.keystorePath = keystore.toAbsolutePath().toString(); + this.truststorePath = truststore.toAbsolutePath().toString(); + LOG.info( + "Successfully generated Keystore: {} and Truststore: {}", keystorePath, truststorePath); + } + + private void runProcess(ProcessBuilder pb, String errorMessage) + throws IOException, InterruptedException { + pb.redirectErrorStream(true); + Process process = null; + try { + process = pb.start(); + int exitCode = process.waitFor(); + if (exitCode != 0) { + String output = + new String( + process.getInputStream().readAllBytes(), java.nio.charset.StandardCharsets.UTF_8); + throw new RuntimeException( + errorMessage + ". Exit code: " + exitCode + ", Output: " + output); + } + } finally { + if (process != null) { + process.destroy(); + } + } + } + + public String getKeystorePath() { + return keystorePath; + } + + public String getTruststorePath() { + return truststorePath; + } + + public static final class Builder extends AbstractJDBCResourceManager.Builder> { + + public Builder(String testId) { + super(testId, DEFAULT_MYSQL_CONTAINER_NAME, DEFAULT_MYSQL_CONTAINER_TAG); + } + + @Override + public SSLMySQLResourceManager build() { + MySQLContainer container = + new MySQLContainer<>( + DockerImageName.parse(this.containerImageName).withTag(this.containerImageTag)); + return new SSLMySQLResourceManager(container, this); + } + } +} diff --git a/it/jdbc/src/main/java/org/apache/beam/it/jdbc/package-info.java b/it/jdbc/src/main/java/org/apache/beam/it/jdbc/package-info.java new file mode 100644 index 0000000000..cb7b595625 --- /dev/null +++ b/it/jdbc/src/main/java/org/apache/beam/it/jdbc/package-info.java @@ -0,0 +1,20 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** Package for managing JDBC resources. */ +package org.apache.beam.it.jdbc; diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbIT.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbIT.java index 339ab8f14b..b084bf67cd 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbIT.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbIT.java @@ -27,10 +27,16 @@ import com.google.cloud.spanner.Value; import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; +import com.google.cloud.teleport.v2.spanner.migrations.source.config.JdbcShardConfig; import com.google.cloud.teleport.v2.templates.utils.SpannerGeneratedColumnUtils; import com.google.common.io.Resources; +import com.google.gson.Gson; +import com.google.gson.JsonObject; import com.google.pubsub.v1.SubscriptionName; import java.io.IOException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; import java.util.HashMap; @@ -43,7 +49,7 @@ import org.apache.beam.it.gcp.pubsub.PubsubResourceManager; import org.apache.beam.it.gcp.spanner.SpannerResourceManager; import org.apache.beam.it.gcp.storage.GcsResourceManager; -import org.apache.beam.it.jdbc.MySQLResourceManager; +import org.apache.beam.it.jdbc.SSLMySQLResourceManager; import org.apache.beam.sdk.io.gcp.spanner.SpannerAccessor; import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig; import org.junit.AfterClass; @@ -84,7 +90,7 @@ public class SpannerToSourceDbIT extends SpannerToSourceDbITBase { private static PipelineLauncher.LaunchInfo jobInfo; public static SpannerResourceManager spannerResourceManager; private static SpannerResourceManager spannerMetadataResourceManager; - private static MySQLResourceManager jdbcResourceManager; + private static SSLMySQLResourceManager jdbcResourceManager; private static GcsResourceManager gcsResourceManager; private static PubsubResourceManager pubsubResourceManager; private SubscriptionName subscriptionName; @@ -103,12 +109,36 @@ public void setUp() throws IOException { spannerResourceManager = createSpannerDatabase(SpannerToSourceDbIT.SPANNER_DDL_RESOURCE); spannerMetadataResourceManager = createSpannerMetadataDatabase(); - jdbcResourceManager = MySQLResourceManager.builder(testName).build(); + jdbcResourceManager = SSLMySQLResourceManager.builder(testName).build(); createMySQLSchema(jdbcResourceManager, SpannerToSourceDbIT.MYSQL_SCHEMA_FILE_RESOURCE); gcsResourceManager = setUpSpannerITGcsResourceManager(); - createAndUploadShardConfigToGcs(gcsResourceManager, jdbcResourceManager); + gcsResourceManager.uploadArtifact( + "input/truststore_Shard1.jks", jdbcResourceManager.getTruststorePath()); + String truststoreGcsUrl = getGcsPath("input/truststore_Shard1.jks", gcsResourceManager); + + String truststoreLocalUrl = "file:///extra_files/truststore_Shard1.jks"; + + String props = + String.format( + "sslMode=VERIFY_CA&allowPublicKeyRetrieval=true&trustCertificateKeyStoreUrl=%s&trustCertificateKeyStorePassword=%s", + URLEncoder.encode(truststoreLocalUrl, StandardCharsets.UTF_8), + URLEncoder.encode(jdbcResourceManager.getPassword(), StandardCharsets.UTF_8)); + + Shard shard = new Shard(); + shard.setLogicalShardId("Shard1"); + shard.setUser(jdbcResourceManager.getUsername()); + shard.setPassword(jdbcResourceManager.getPassword()); + shard.setHost(jdbcResourceManager.getHost()); + shard.setPort(String.valueOf(jdbcResourceManager.getPort())); + shard.setDbName(jdbcResourceManager.getDatabaseName()); + shard.setConnectionProperties(props); + + JdbcShardConfig jdbcShardConfig = new JdbcShardConfig(); + jdbcShardConfig.setShardConfigs(java.util.Collections.singletonList(shard)); + JsonObject jsObj = new Gson().toJsonTree(jdbcShardConfig).getAsJsonObject(); + gcsResourceManager.createArtifact("input/shard.json", jsObj.toString()); gcsResourceManager.uploadArtifact( "input/session.json", Resources.getResource(SESSION_FILE_RESOURCE).getPath()); pubsubResourceManager = setUpPubSubResourceManager(); @@ -123,6 +153,7 @@ public void setUp() throws IOException { new HashMap<>() { { put("sessionFilePath", getGcsPath("input/session.json", gcsResourceManager)); + put("extraFilesToStage", truststoreGcsUrl); } }; jobInfo = diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbITBase.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbITBase.java index acae1ad326..9de5974483 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbITBase.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbITBase.java @@ -389,7 +389,7 @@ protected void loadSQLFileResource(JDBCResourceManager jdbcResourceManager, Stri } } - protected void createMySQLSchema(MySQLResourceManager jdbcResourceManager, String mySqlSchemaFile) + protected void createMySQLSchema(JDBCResourceManager jdbcResourceManager, String mySqlSchemaFile) throws IOException { HashMap columns = new HashMap<>(); columns.put("id", "INT NOT NULL"); diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/utils/SpannerGeneratedColumnUtils.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/utils/SpannerGeneratedColumnUtils.java index e4735ca912..79f01e202e 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/utils/SpannerGeneratedColumnUtils.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/utils/SpannerGeneratedColumnUtils.java @@ -27,7 +27,7 @@ import java.util.Map; import org.apache.beam.it.conditions.ConditionCheck; import org.apache.beam.it.gcp.spanner.SpannerResourceManager; -import org.apache.beam.it.jdbc.MySQLResourceManager; +import org.apache.beam.it.jdbc.JDBCResourceManager; import org.checkerframework.checker.initialization.qual.Initialized; import org.checkerframework.checker.nullness.qual.NonNull; import org.checkerframework.checker.nullness.qual.UnknownKeyFor; @@ -39,7 +39,7 @@ public class SpannerGeneratedColumnUtils { public static ConditionCheck buildConditionCheck( Map>> spannerTableData, - MySQLResourceManager jdbcResourceManager) { + JDBCResourceManager jdbcResourceManager) { ConditionCheck combinedCondition = null; for (Map.Entry>> entry : spannerTableData.entrySet()) { String tableName = getTableName(entry.getKey()); @@ -69,7 +69,7 @@ public static ConditionCheck buildConditionCheck( public static void assertRowInMySQL( Map>> expectedData, - MySQLResourceManager jdbcResourceManager) { + JDBCResourceManager jdbcResourceManager) { for (Map.Entry>> expectedTableData : expectedData.entrySet()) { String type = expectedTableData.getKey(); String tableName = getTableName(type);