Merge pull request #181 from data-integrations/connection-management

Adding CloudSqlMySql  and CloudSQLPostgreSQL connector

Author: itsankit-google

cloudsql-mysql-plugin/docs/CloudSQLMySQL-connector.md

@@ -0,0 +1,125 @@
+# CloudSQLMySQL Connection
+
+
+Description
+-----------
+Use this connection to access data in a CloudSQLMySQL database using JDBC.
+
+Properties
+----------
+**Name:** Name of the connection. Connection names must be unique in a namespace.
+
+**Description:** Description of the connection.
+
+**JDBC Driver name:** Select the JDBC driver to use.
+
+**CloudSQL Instance Type:** Whether the CloudSQL instance to connect to is private or public. Defaults to 'Public'.
+
+**Connection Name:** The CloudSQL instance to connect to in the format <PROJECT_ID>:\<REGION>:<INSTANCE_NAME>.
+Can be found in the instance overview page.
+
+**Database:** MySQL database name.
+
+**Username:** User identity for connecting to the specified database. Required for databases that need
+authentication. Optional for databases that do not require authentication.
+
+**Password:** Password to use to connect to the specified database.
+
+**Connection Arguments:** A list of arbitrary string tag/value pairs as connection arguments. These arguments
+will be passed to the JDBC driver, as connection arguments, for JDBC drivers that may need additional configurations.
+This is a semicolon-separated list of key-value pairs, where each pair is separated by a equals '=' and specifies
+the key and value for the argument. For example, 'key1=value1;key2=value' specifies that the connection will be
+given arguments 'key1' mapped to 'value1' and the argument 'key2' mapped to 'value2'.
+
+Path of the connection
+----------------------
+To browse, get a sample from, or get the specification for this connection through
+[Pipeline Microservices](https://cdap.atlassian.net/wiki/spaces/DOCS/pages/975929350/Pipeline+Microservices), the `path`
+property is required in the request body. It can be in the following form :
+
+1. `/{database}/{table}`
+   This path indicates a table. A table is the only one that can be sampled. Browse on this path to return the specified table.
+
+2. `/{database}`
+   This path indicates a database. A database cannot be sampled. Browse on this path to get all the tables under this database.
+
+3. `/`
+   This path indicates the root. A root cannot be sampled. Browse on this path to get all the databases visible through this connection.
+
+Examples
+--------
+**Connecting to a public CloudSQL MySQL instance**
+
+Suppose you want to read data from CloudSQL MySQL database named "prod", as "root" user with "root" password (Get the
+latest version of the CloudSQL socket factory jar with driver and dependencies
+[here](https://github.com/GoogleCloudPlatform/cloud-sql-jdbc-socket-factory/releases)), then configure plugin with:
+
+
+```
+Reference Name: "connection1"
+Driver Name: "cloudsql-mysql"
+Database: "prod"
+CloudSQL Instance Type: "Public" 
+Connection Name: [PROJECT_ID]:[REGION]:[INSTANCE_NAME]
+Username: "root"
+Password: "root"
+```
+
+**Connecting to a private CloudSQL MySQL instance**
+
+If you want to connect to a private CloudSQL MySQL instance, create a Compute Engine VM that runs the CloudSQL Proxy
+docker image using the following command
+
+```
+# Set the environment variables
+export PROJECT=[project_id]
+export REGION=[vm-region]
+export ZONE=`gcloud compute zones list --filter="name=${REGION}" --limit
+1 --uri --project=${PROJECT}| sed 's/.*\///'`
+export SUBNET=[vpc-subnet-name]
+export NAME=[gce-vm-name]
+export MYSQL_CONN=[mysql-instance-connection-name]
+
+# Create a Compute Engine VM
+gcloud beta compute --project=${PROJECT_ID} instances create ${INSTANCE_NAME}
+--zone=${ZONE} --machine-type=g1-small --subnet=${SUBNE} --no-address
+--metadata=startup-script="docker run -d -p 0.0.0.0:3306:3306
+gcr.io/cloudsql-docker/gce-proxy:1.16 /cloud_sql_proxy
+-instances=${MYSQL_CONNECTION_NAME}=tcp:0.0.0.0:3306" --maintenance-policy=MIGRATE
+--scopes=https://www.googleapis.com/auth/cloud-platform
+--image=cos-69-10895-385-0 --image-project=cos-cloud
+```
+
+Optionally, you can promote the internal IP address of the VM running the Proxy image to a static IP using
+
+```
+# Get the VM internal IP
+export IP=`gcloud compute instances describe ${NAME} --zone ${ZONE} |
+grep "networkIP" | awk '{print $2}'`
+
+# Promote the VM internal IP to static IP
+gcloud compute addresses create mysql-proxy --addresses ${IP} --region
+${REGION} --subnet ${SUBNET}
+
+
+# Note down the IP to be used in MySQL or PostgreSQL JDBC 
+# connection string
+echo Proxy IP: ${IP}
+
+echo "JDBC Connection strings:"
+echo "jdbc:postgresql://${IP}:5432/{PostgreSQL_DB_NAME}"
+echo "jdbc:mysql://${IP}:3306/{MySQL_DB_NAME}"
+```
+
+Get the latest version of the CloudSQL socket factory jar with driver and dependencies from
+[here](https://github.com/GoogleCloudPlatform/cloud-sql-jdbc-socket-factory/releases), then configure plugin with:
+
+```
+Reference Name: "connection1"
+Driver Name: "cloudsql-mysql"
+Database: "prod"
+CloudSQL Instance Type: "Private"
+Connection Name: <proxy-ip> (obtained from commands above)
+Username: "root"
+Password: "root"
+```

cloudsql-mysql-plugin/pom.xml

@@ -43,9 +43,55 @@
       <groupId>io.cdap.plugin</groupId>
       <artifactId>hydrator-common</artifactId>
     </dependency>
+
+    <!-- test dependencies -->
+    <dependency>
+      <groupId>io.cdap.plugin</groupId>
+      <artifactId>database-commons</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>hydrator-test</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-data-pipeline2_2.11</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-api</artifactId>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.jetbrains</groupId>
+      <artifactId>annotations</artifactId>
+      <version>RELEASE</version>
+      <scope>compile</scope>
+    </dependency>
+    <!-- https://mvnrepository.com/artifact/com.google.cloud.sql/mysql-socket-factory-connector-j-8 -->
+    <dependency>
+      <groupId>com.google.cloud.sql</groupId>
+      <artifactId>mysql-socket-factory-connector-j-8</artifactId>
+      <version>1.0.10</version>
+      <scope>test</scope>
+    </dependency>
+    <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
+    <dependency>
+      <groupId>mysql</groupId>
+      <artifactId>mysql-connector-java</artifactId>
+      <version>8.0.11</version>
       <scope>test</scope>
     </dependency>
   </dependencies>

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright © 2021 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.cloudsql.mysql;
+
+import io.cdap.cdap.api.annotation.Category;
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.annotation.Plugin;
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.etl.api.batch.BatchSink;
+import io.cdap.cdap.etl.api.batch.BatchSource;
+import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.cdap.etl.api.connector.ConnectorSpec;
+import io.cdap.cdap.etl.api.connector.ConnectorSpecRequest;
+import io.cdap.cdap.etl.api.connector.PluginSpec;
+import io.cdap.plugin.common.Constants;
+import io.cdap.plugin.common.ReferenceNames;
+import io.cdap.plugin.common.db.DBConnectorPath;
+import io.cdap.plugin.db.ConnectionConfig;
+import io.cdap.plugin.db.DBRecord;
+import io.cdap.plugin.db.connector.AbstractDBSpecificConnector;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.lib.db.DBWritable;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ *  A CLoudSQL MySQL Server Database Connector that connects to CloudSQL MySQL Server database via JDBC
+ */
+@Plugin(type = Connector.PLUGIN_TYPE)
+@Name(CloudSQLMySQLConnector.NAME)
+@Description("Connection to access data in CloudSQL MySQL Server databases using JDBC.")
+@Category("Database")
+public class CloudSQLMySQLConnector extends AbstractDBSpecificConnector<DBRecord> {
+  
+  public static final String NAME = CloudSQLMySQLConstants.PLUGIN_NAME;
+  private final CloudSQLMySQLConnectorConfig config;
+
+  public CloudSQLMySQLConnector(CloudSQLMySQLConnectorConfig config) {
+    super(config);
+    this.config = config;
+  }
+
+  @Override
+  public boolean supportSchema() {
+    return false;
+  }
+
+  @Override
+  protected Class<? extends DBWritable> getDBRecordType() {
+    return DBRecord.class;
+  }
+
+  @Override
+  public StructuredRecord transform(LongWritable longWritable, DBRecord dbRecord) {
+    return dbRecord.getRecord();
+  }
+
+  @Override
+  protected String getTableQuery(DBConnectorPath path) {
+    return String.format("SELECT * FROM `%s`.`%s`", path.getDatabase(), path.getTable());
+  }
+
+  @Override
+  protected String getTableQuery(DBConnectorPath path, int limit) {
+    return String.format("SELECT * FROM `%s`.`%s` LIMIT %d", path.getDatabase(), path.getTable(), limit);
+  }
+
+  @Override
+  protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath path,
+                                  ConnectorSpec.Builder builder) {
+    Map<String, String> properties = new HashMap<>();
+    setConnectionProperties(properties);
+    builder
+      .addRelatedPlugin(new PluginSpec(CloudSQLMySQLConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, properties))
+      .addRelatedPlugin(new PluginSpec(CloudSQLMySQLConstants.PLUGIN_NAME, BatchSink.PLUGIN_TYPE, properties));
+
+    String table = path.getTable();
+    if (table == null) {
+      return;
+    }
+
+    properties.put(CloudSQLMySQLSource.CloudSQLMySQLSourceConfig.IMPORT_QUERY, getTableQuery(path));
+    properties.put(CloudSQLMySQLSource.CloudSQLMySQLSourceConfig.NUM_SPLITS, "1");
+    properties.put(ConnectionConfig.DATABASE, path.getDatabase());
+    properties.put(Constants.Reference.REFERENCE_NAME, ReferenceNames.cleanseReferenceName(table));
+    properties.put(CloudSQLMySQLSink.CloudSQLMySQLSinkConfig.TABLE_NAME, table);
+  }
+
+  @Override
+  protected void setConnectionProperties(Map<String, String> properties) {
+    Map<String, String> rawProperties = config.getRawProperties().getProperties();
+    properties.put(ConnectionConfig.JDBC_PLUGIN_NAME, rawProperties.get(ConnectionConfig.JDBC_PLUGIN_NAME));
+    properties.put(ConnectionConfig.USER, rawProperties.get(ConnectionConfig.USER));
+    properties.put(ConnectionConfig.PASSWORD, rawProperties.get(ConnectionConfig.PASSWORD));
+    properties.put(CloudSQLMySQLConstants.CONNECTION_NAME, rawProperties.get(CloudSQLMySQLConstants.CONNECTION_NAME));
+    properties.put(CloudSQLMySQLConstants.INSTANCE_TYPE, rawProperties.get(CloudSQLMySQLConstants.INSTANCE_TYPE));
+  }
+}

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorConfig.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright © 2021 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.plugin.cloudsql.mysql;
+
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.plugin.db.ConnectionConfig;
+import io.cdap.plugin.db.connector.AbstractDBConnectorConfig;
+
+import java.util.Properties;
+import javax.annotation.Nullable;
+
+/**
+ * Configuration for CloudSQL MySQL connector
+ */
+public class CloudSQLMySQLConnectorConfig extends AbstractDBConnectorConfig {
+
+  private static final String JDBC_PROPERTY_CONNECT_TIMEOUT_MILLIS = "connectTimeout";
+  private static final String JDBC_PROPERTY_SOCKET_TIMEOUT_MILLIS = "socketTimeout";
+
+  @Name(CloudSQLMySQLConstants.CONNECTION_NAME)
+  @Description(
+    "The CloudSQL instance to connect to. For a public instance, the connection string should be in the format "
+      + "<PROJECT_ID>:<REGION>:<INSTANCE_NAME> which can be found in the instance overview page. For a private "
+      + "instance, enter the internal IP address of the Compute Engine VM cloudsql proxy is running on.")
+  private String connectionName;
+
+  @Name(ConnectionConfig.DATABASE)
+  @Description("Database name to connect to")
+  private String database;
+
+  @Name(CloudSQLMySQLConstants.INSTANCE_TYPE)
+  @Description("Whether the CloudSQL instance to connect to is private or public.")
+  @Nullable
+  private String instanceType;
+
+  public CloudSQLMySQLConnectorConfig(String user, String password, String jdbcPluginName, String connectionArguments,
+                                      String instanceType, String connectionName, String database) {
+    this.user = user;
+    this.password = password;
+    this.jdbcPluginName = jdbcPluginName;
+    this.connectionArguments = connectionArguments;
+    this.instanceType = instanceType;
+    this.connectionName = connectionName;
+    this.database = database;
+  }
+
+  public String getDatabase() {
+    return database;
+  }
+
+  public String getInstanceType() {
+    return instanceType;
+  }
+
+  public String getConnectionName() {
+    return connectionName;
+  }
+
+  @Override
+  public String getConnectionString() {
+    if (CloudSQLMySQLConstants.PRIVATE_INSTANCE.equalsIgnoreCase(instanceType)) {
+      return String.format(
+        CloudSQLMySQLConstants.PRIVATE_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT,
+        connectionName,
+        database);
+    }
+
+    return String.format(
+      CloudSQLMySQLConstants.PUBLIC_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT,
+      database,
+      connectionName);
+  }
+
+  @Override
+  public Properties getConnectionArgumentsProperties() {
+    Properties properties = super.getConnectionArgumentsProperties();
+    properties.put(JDBC_PROPERTY_CONNECT_TIMEOUT_MILLIS, "20000");
+    properties.put(JDBC_PROPERTY_SOCKET_TIMEOUT_MILLIS, "20000");
+    return properties;
+  }
+}