Add datastream to sql IT's

Signed-off-by: Jeffrey Kinard <[email protected]>
GoogleCloudPlatform · Jul 3, 2024 · 7e440c4 · 7e440c4
1 parent 0bcf179
commit 7e440c4
Show file tree

Hide file tree

Showing 11 changed files with 820 additions and 39 deletions.
diff --git a/cicd/cmd/run-it-smoke-tests/main.go b/cicd/cmd/run-it-smoke-tests/main.go
@@ -67,9 +67,11 @@ func main() {
 		flags.SpannerHost(),
 		flags.FailureMode(),
 		flags.RetryFailures(),
-		flags.StaticOracleInstance(),
+		flags.StaticOracleHost(),
+		flags.StaticOracleSysPassword(),
 		flags.CloudProxyHost(),
-		flags.CloudProxyPort(),
+		flags.CloudProxyMySqlPort(),
+		flags.CloudProxyPostgresPort(),
 		flags.CloudProxyPassword())
 	if err != nil {
 		log.Fatalf("%v\n", err)

diff --git a/cicd/cmd/run-it-tests/main.go b/cicd/cmd/run-it-tests/main.go
@@ -68,9 +68,11 @@ func main() {
 		flags.SpannerHost(),
 		flags.FailureMode(),
 		flags.RetryFailures(),
-		flags.StaticOracleInstance(),
+		flags.StaticOracleHost(),
+		flags.StaticOracleSysPassword(),
 		flags.CloudProxyHost(),
-		flags.CloudProxyPort(),
+		flags.CloudProxyMySqlPort(),
+		flags.CloudProxyPostgresPort(),
 		flags.CloudProxyPassword())
 	if err != nil {
 		log.Fatalf("%v\n", err)

diff --git a/cicd/cmd/run-load-tests/main.go b/cicd/cmd/run-load-tests/main.go
@@ -67,9 +67,11 @@ func main() {
 		flags.ExportProject(),
 		flags.ExportDataset(),
 		flags.ExportTable(),
-		flags.StaticOracleInstance(),
+		flags.StaticOracleHost(),
+		flags.StaticOracleSysPassword(),
 		flags.CloudProxyHost(),
-		flags.CloudProxyPort(),
+		flags.CloudProxyMySqlPort(),
+		flags.CloudProxyPostgresPort(),
 		flags.CloudProxyPassword())
 	if err != nil {
 		log.Fatalf("%v\n", err)

diff --git a/cicd/internal/flags/it-flags.go b/cicd/internal/flags/it-flags.go
@@ -24,19 +24,21 @@ import (
 
 // Avoid making these vars public.
 var (
-	dRegion              string
-	dProject             string
-	dArtifactBucket      string
-	dStageBucket         string
-	dHostIp              string
-	dPrivateConnectivity string
-	dSpannerHost         string
-	dReleaseMode         bool
-	dRetryFailures       string
-	dCloudProxyHost      string
-	dCloudProxyPort      string
-	dCloudProxyPassword  string
-	dOracleInstance      string
+	dRegion                 string
+	dProject                string
+	dArtifactBucket         string
+	dStageBucket            string
+	dHostIp                 string
+	dPrivateConnectivity    string
+	dSpannerHost            string
+	dReleaseMode            bool
+	dRetryFailures          string
+	dCloudProxyHost         string
+	dCloudProxyMySqlPort    string
+	dCloudProxyPostgresPort string
+	dCloudProxyPassword     string
+	dOracleHost             string
+	dCloudOracleSysPassword string
 )
 
 // Registers all common flags. Must be called before flag.Parse().
@@ -51,9 +53,11 @@ func RegisterItFlags() {
 	flag.BoolVar(&dReleaseMode, "it-release", false, "(optional) Set if tests are being executed for a release")
 	flag.StringVar(&dRetryFailures, "it-retry-failures", "0", "Number of retries attempts for failing tests")
 	flag.StringVar(&dCloudProxyHost, "it-cloud-proxy-host", "10.128.0.34", "Hostname or IP address of static Cloud Auth Proxy")
-	flag.StringVar(&dCloudProxyPort, "it-cloud-proxy-port", "33134", "Port number of static Cloud Auth Proxy")
+	flag.StringVar(&dCloudProxyMySqlPort, "it-cloud-proxy-mysql-port", "33134", "MySql port number on static Cloud Auth Proxy")
+	flag.StringVar(&dCloudProxyPostgresPort, "it-cloud-proxy-postgres-port", "33136", "Postgres port number on static Cloud Auth Proxy")
 	flag.StringVar(&dCloudProxyPassword, "it-cloud-proxy-password", "t>5xl%J(&qTK6?FaZ", "Password of static Cloud Auth Proxy")
-	flag.StringVar(&dOracleInstance, "it-oracle-host", "10.128.0.90", "Hostname or IP address of static Oracle DB")
+	flag.StringVar(&dOracleHost, "it-oracle-host", "10.128.0.90", "Hostname or IP address of static Oracle DB")
+	flag.StringVar(&dCloudOracleSysPassword, "it-oracle-sys-password", "oracle", "sys password of static Oracle DB")
 }
 
 func Region() string {
@@ -119,14 +123,22 @@ func CloudProxyHost() string {
 	return "-DcloudProxyHost=" + dCloudProxyHost
 }
 
-func CloudProxyPort() string {
-	return "-DcloudProxyPort=" + dCloudProxyPort
+func CloudProxyMySqlPort() string {
+	return "-DcloudProxyMySqlPort=" + dCloudProxyMySqlPort
+}
+
+func CloudProxyPostgresPort() string {
+	return "-DcloudProxyPostgresPort=" + dCloudProxyPostgresPort
 }
 
 func CloudProxyPassword() string {
 	return "-DcloudProxyPassword=" + dCloudProxyPassword
 }
 
-func StaticOracleInstance() string {
-	return "-DcloudOracleHost=" + dOracleInstance
+func StaticOracleHost() string {
+	return "-DcloudOracleHost=" + dOracleHost
+}
+
+func StaticOracleSysPassword() string {
+	return "-DcloudOracleSysPassword=" + dCloudOracleSysPassword
 }
diff --git a/...oud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudMySQLResourceManager.java b/...oud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudMySQLResourceManager.java
@@ -18,6 +18,8 @@
 package org.apache.beam.it.gcp.cloudsql;
 
 import org.checkerframework.checker.nullness.qual.NonNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Custom class for the MySQL implementation of {@link CloudSqlResourceManager} abstract class.
@@ -29,6 +31,8 @@
  */
 public class CloudMySQLResourceManager extends CloudSqlResourceManager {
 
+  private static final Logger LOG = LoggerFactory.getLogger(CloudMySQLResourceManager.class);
+
   private CloudMySQLResourceManager(Builder builder) {
     super(builder);
   }
@@ -49,6 +53,15 @@ public Builder(String testId) {
       super(testId);
     }
 
+    @Override
+    protected void configurePort() {
+      if (System.getProperty("cloudProxyMySqlPort") != null) {
+        this.setPort(Integer.parseInt(System.getProperty("cloudProxyMySqlPort")));
+      } else {
+        LOG.warn("Missing -DcloudProxyMySqlPort.");
+      }
+    }
+
     @Override
     public @NonNull CloudMySQLResourceManager build() {
       return new CloudMySQLResourceManager(this);

diff --git a/...ud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudOracleResourceManager.java b/...ud-platform/src/main/java/org/apache/beam/it/gcp/cloudsql/CloudOracleResourceManager.java
@@ -36,11 +36,16 @@ public class CloudOracleResourceManager extends CloudSqlResourceManager {
 
   private static final Logger LOG = LoggerFactory.getLogger(CloudOracleResourceManager.class);
 
+  private static final String DEFAULT_SYSTEM_IDENTIFIER = "xe";
+
   private static final int DEFAULT_ORACLE_PORT = 1521;
 
-  private CloudOracleResourceManager(Builder builder) {
+  private String systemIdentifier;
+
+  protected CloudOracleResourceManager(Builder builder) {
     super(builder);
 
+    this.systemIdentifier = builder.systemIdentifier;
     System.setProperty("oracle.jdbc.timezoneAsRegion", "false");
   }
 
@@ -65,27 +70,65 @@ public static Builder builder(String testId) {
     return "SELECT * FROM " + tableName + " WHERE ROWNUM <= 1";
   }
 
+  /**
+   * Return the SID of the connected DB.
+   *
+   * @return the SID.
+   */
+  public String getSystemIdentifier() {
+    return this.systemIdentifier;
+  }
+
   /** Builder for {@link CloudOracleResourceManager}. */
   public static final class Builder extends CloudSqlResourceManager.Builder {
 
+    private String systemIdentifier;
+
     public Builder(String testId) {
       super(testId);
 
-      this.setDatabaseName("xe");
-      this.setPort(DEFAULT_ORACLE_PORT);
-
-      // Currently only supports static Oracle instance on GCE
-      this.maybeUseStaticInstance();
+      this.setSystemIdentifier(DEFAULT_SYSTEM_IDENTIFIER);
+      this.setDatabaseName(this.systemIdentifier);
     }
 
-    public Builder maybeUseStaticInstance() {
+    @Override
+    protected void configureHost() {
       if (System.getProperty("cloudOracleHost") != null) {
         this.setHost(System.getProperty("cloudOracleHost"));
       } else {
         LOG.warn("Missing -DcloudOracleHost.");
       }
-      this.useStaticContainer();
+    }
+
+    @Override
+    protected void configurePort() {
+      if (System.getProperty("cloudOraclePort") != null) {
+        this.setPort(Integer.parseInt(System.getProperty("cloudOraclePort")));
+      } else {
+        this.setPort(DEFAULT_ORACLE_PORT);
+      }
+    }
+
+    @Override
+    protected void configureUsername() {
+      if (System.getProperty("cloudOracleUsername") != null) {
+        this.setUsername(System.getProperty("cloudOracleUsername"));
+      } else {
+        super.configureUsername();
+      }
+    }
+
+    @Override
+    protected void configurePassword() {
+      if (System.getProperty("cloudOraclePassword") != null) {
+        this.setPassword(System.getProperty("cloudOraclePassword"));
+      } else {
+        super.configurePassword();
+      }
+    }
 
+    public Builder setSystemIdentifier(String systemIdentifier) {
+      this.systemIdentifier = systemIdentifier;
       return this;
     }