Merge pull request #298 from data-integrations/feature/add-fqn

Add FQN support for DB plugins

Author: snehalakshmisha

cloudsql-mysql-plugin/pom.xml

@@ -126,7 +126,7 @@
               io.cdap.plugin.cloudsql.mysql.*;
               io.cdap.plugin.db.batch.source.*;
               io.cdap.plugin.db.batch.sink.*;
-              org.apache.commons.lang;
+              org.apache.commons.lang.*;
               org.apache.commons.logging.*;
               org.codehaus.jackson.*
             </_exportcontents>

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

@@ -26,13 +26,18 @@
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchSink;
+import io.cdap.cdap.etl.api.batch.BatchSinkContext;
 import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.plugin.common.Asset;
 import io.cdap.plugin.common.ConfigUtil;
+import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.db.CommonSchemaReader;
 import io.cdap.plugin.db.SchemaReader;
 import io.cdap.plugin.db.batch.config.AbstractDBSpecificSinkConfig;
 import io.cdap.plugin.db.batch.sink.AbstractDBSink;
 import io.cdap.plugin.util.CloudSQLUtil;
+import io.cdap.plugin.util.DBUtils;
+import org.apache.commons.lang.StringUtils;
 
 import java.util.Map;
 import javax.annotation.Nullable;
@@ -71,6 +76,29 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
   protected SchemaReader getSchemaReader() {
     return new CommonSchemaReader();
   }
+
+  @Override
+  protected LineageRecorder getLineageRecorder(BatchSinkContext context) {
+    String host;
+    String location = "";
+    if (CloudSQLUtil.PRIVATE_INSTANCE.equalsIgnoreCase(cloudsqlMysqlSinkConfig.getConnection().getInstanceType())) {
+      // connection is the private IP address
+      host = cloudsqlMysqlSinkConfig.getConnection().getConnectionName();
+    } else {
+      // connection is of the form <projectId>:<region>:<instanceName>
+      String[] connectionParams = cloudsqlMysqlSinkConfig.getConnection().getConnectionName().split(":");
+      host = connectionParams[2];
+      location = connectionParams[1];
+    }
+    String fqn = DBUtils.constructFQN("mysql", host, 3306,
+                                      cloudsqlMysqlSinkConfig.getConnection().getDatabase(),
+                                      cloudsqlMysqlSinkConfig.getReferenceName());
+    Asset.Builder assetBuilder = Asset.builder(cloudsqlMysqlSinkConfig.getReferenceName()).setFqn(fqn);
+    if (!StringUtils.isEmpty(location)) {
+      assetBuilder.setLocation(location);
+    }
+    return new LineageRecorder(context, assetBuilder.build());
+  }
 
   /** CloudSQL MySQL sink configuration. */
   public static class CloudSQLMySQLSinkConfig extends AbstractDBSpecificSinkConfig {

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

@@ -25,13 +25,18 @@
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchSource;
+import io.cdap.cdap.etl.api.batch.BatchSourceContext;
 import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.plugin.common.Asset;
 import io.cdap.plugin.common.ConfigUtil;
+import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.db.CommonSchemaReader;
 import io.cdap.plugin.db.SchemaReader;
 import io.cdap.plugin.db.batch.config.AbstractDBSpecificSourceConfig;
 import io.cdap.plugin.db.batch.source.AbstractDBSource;
 import io.cdap.plugin.util.CloudSQLUtil;
+import io.cdap.plugin.util.DBUtils;
+import org.apache.commons.lang.StringUtils;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -90,6 +95,29 @@ protected String createConnectionString() {
         cloudsqlMysqlSourceConfig.connection.getConnectionName());
   }
 
+  @Override
+  protected LineageRecorder getLineageRecorder(BatchSourceContext context) {
+    String host;
+    String location = "";
+    if (CloudSQLUtil.PRIVATE_INSTANCE.equalsIgnoreCase(cloudsqlMysqlSourceConfig.getConnection().getInstanceType())) {
+      // connection is the private IP address
+      host = cloudsqlMysqlSourceConfig.getConnection().getConnectionName();
+    } else {
+      // connection is of the form <projectId>:<region>:<instanceName>
+      String[] connectionParams = cloudsqlMysqlSourceConfig.getConnection().getConnectionName().split(":");
+      host = connectionParams[2];
+      location = connectionParams[1];
+    }
+    String fqn = DBUtils.constructFQN("mysql", host, 3306,
+                                      cloudsqlMysqlSourceConfig.getConnection().getDatabase(),
+                                      cloudsqlMysqlSourceConfig.getReferenceName());
+    Asset.Builder assetBuilder = Asset.builder(cloudsqlMysqlSourceConfig.getReferenceName()).setFqn(fqn);
+    if (!StringUtils.isEmpty(location)) {
+      assetBuilder.setLocation(location);
+    }
+    return new LineageRecorder(context, assetBuilder.build());
+  }
+
   /** CloudSQL MySQL source config. */
   public static class CloudSQLMySQLSourceConfig extends AbstractDBSpecificSourceConfig {
 

cloudsql-mysql-plugin/widgets/CloudSQLMySQL-batchsink.json

@@ -98,7 +98,7 @@
           "label": "Reference Name",
           "name": "referenceName",
           "widget-attributes": {
-            "placeholder": "Name used to identify this sink for lineage"
+            "placeholder": "Name used to identify this sink for lineage. Typically, the name of the table/view."
           }
         },
         {

cloudsql-mysql-plugin/widgets/CloudSQLMySQL-batchsource.json

@@ -98,7 +98,7 @@
           "label": "Reference Name",
           "name": "referenceName",
           "widget-attributes": {
-            "placeholder": "Name used to identify this source for lineage"
+            "placeholder": "Name used to identify this source for lineage. Typically, the name of the table/view."
           }
         },
         {

cloudsql-postgresql-plugin/pom.xml

@@ -134,9 +134,10 @@
           <instructions>
             <_exportcontents>
               io.cdap.plugin.cloudsql.postgres.*;
+              io.cdap.plugin.postgres.*;
               io.cdap.plugin.db.batch.source.*;
               io.cdap.plugin.db.batch.sink.*;
-              org.apache.commons.lang;
+              org.apache.commons.lang.*;
               org.apache.commons.logging.*;
               org.codehaus.jackson.*
             </_exportcontents>

cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLSink.java

@@ -28,8 +28,11 @@
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchSink;
+import io.cdap.cdap.etl.api.batch.BatchSinkContext;
 import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.plugin.common.Asset;
 import io.cdap.plugin.common.ConfigUtil;
+import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.db.DBRecord;
 import io.cdap.plugin.db.SchemaReader;
 import io.cdap.plugin.db.batch.config.AbstractDBSpecificSinkConfig;
@@ -39,6 +42,8 @@
 import io.cdap.plugin.postgres.PostgresFieldsValidator;
 import io.cdap.plugin.postgres.PostgresSchemaReader;
 import io.cdap.plugin.util.CloudSQLUtil;
+import io.cdap.plugin.util.DBUtils;
+import org.apache.commons.lang.StringUtils;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -107,6 +112,30 @@ protected FieldsValidator getFieldsValidator() {
     return new PostgresFieldsValidator();
   }
 
+  @Override
+  protected LineageRecorder getLineageRecorder(BatchSinkContext context) {
+    String host;
+    String location = "";
+    if (CloudSQLUtil.PRIVATE_INSTANCE.equalsIgnoreCase(
+      cloudsqlPostgresqlSinkConfig.getConnection().getInstanceType())) {
+      // connection is the private IP address
+      host = cloudsqlPostgresqlSinkConfig.getConnection().getConnectionName();
+    } else {
+      // connection is of the form <projectId>:<region>:<instanceName>
+      String[] connectionParams = cloudsqlPostgresqlSinkConfig.getConnection().getConnectionName().split(":");
+      host = connectionParams[2];
+      location = connectionParams[1];
+    }
+    String fqn = DBUtils.constructFQN("postgres", host, 5432,
+                                      cloudsqlPostgresqlSinkConfig.getConnection().getDatabase(),
+                                      cloudsqlPostgresqlSinkConfig.getReferenceName());
+    Asset.Builder assetBuilder = Asset.builder(cloudsqlPostgresqlSinkConfig.getReferenceName()).setFqn(fqn);
+    if (!StringUtils.isEmpty(location)) {
+      assetBuilder.setLocation(location);
+    }
+    return new LineageRecorder(context, assetBuilder.build());
+  }
+
   /** CloudSQL PostgreSQL sink config. */
   public static class CloudSQLPostgreSQLSinkConfig extends AbstractDBSpecificSinkConfig {
 

cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLSource.java

@@ -25,14 +25,19 @@
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.batch.BatchSource;
+import io.cdap.cdap.etl.api.batch.BatchSourceContext;
 import io.cdap.cdap.etl.api.connector.Connector;
+import io.cdap.plugin.common.Asset;
 import io.cdap.plugin.common.ConfigUtil;
+import io.cdap.plugin.common.LineageRecorder;
 import io.cdap.plugin.db.SchemaReader;
 import io.cdap.plugin.db.batch.config.AbstractDBSpecificSourceConfig;
 import io.cdap.plugin.db.batch.source.AbstractDBSource;
 import io.cdap.plugin.postgres.PostgresDBRecord;
 import io.cdap.plugin.postgres.PostgresSchemaReader;
 import io.cdap.plugin.util.CloudSQLUtil;
+import io.cdap.plugin.util.DBUtils;
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.mapreduce.lib.db.DBWritable;
 
 import java.util.Collections;
@@ -97,6 +102,30 @@ protected String createConnectionString() {
         cloudsqlPostgresqlSourceConfig.connection.getConnectionName());
   }
 
+  @Override
+  protected LineageRecorder getLineageRecorder(BatchSourceContext context) {
+    String host;
+    String location = "";
+    if (CloudSQLUtil.PRIVATE_INSTANCE.equalsIgnoreCase(
+      cloudsqlPostgresqlSourceConfig.getConnection().getInstanceType())) {
+      // connection is the private IP address
+      host = cloudsqlPostgresqlSourceConfig.getConnection().getConnectionName();
+    } else {
+      // connection is of the form <projectId>:<region>:<instanceName>
+      String[] connectionParams = cloudsqlPostgresqlSourceConfig.getConnection().getConnectionName().split(":");
+      host = connectionParams[2];
+      location = connectionParams[1];
+    }
+    String fqn = DBUtils.constructFQN("postgres", host, 5432,
+                                      cloudsqlPostgresqlSourceConfig.getConnection().getDatabase(),
+                                      cloudsqlPostgresqlSourceConfig.getReferenceName());
+    Asset.Builder assetBuilder = Asset.builder(cloudsqlPostgresqlSourceConfig.getReferenceName()).setFqn(fqn);
+    if (!StringUtils.isEmpty(location)) {
+      assetBuilder.setLocation(location);
+    }
+    return new LineageRecorder(context, assetBuilder.build());
+  }
+
   /** CloudSQL PostgreSQL source config. */
   public static class CloudSQLPostgreSQLSourceConfig extends AbstractDBSpecificSourceConfig {
 

cloudsql-postgresql-plugin/widgets/CloudSQLPostgreSQL-batchsink.json

@@ -98,7 +98,7 @@
           "label": "Reference Name",
           "name": "referenceName",
           "widget-attributes": {
-            "placeholder": "Name used to identify this sink for lineage"
+            "placeholder": "Name used to identify this sink for lineage. Typically, the name of the table/view."
           }
         },
         {

cloudsql-postgresql-plugin/widgets/CloudSQLPostgreSQL-batchsource.json

@@ -98,7 +98,7 @@
           "label": "Reference Name",
           "name": "referenceName",
           "widget-attributes": {
-            "placeholder": "Name used to identify this source for lineage"
+            "placeholder": "Name used to identify this source for lineage. Typically, the name of the table/view."
           }
         },
         {