CDAP-16554 check replication permissions during assessment for MySQL. (#69)

shifuxu0301 · web-flow · commit 581496dbde3f · 2020-04-15T12:18:51.000-07:00
* check replication permissions during assessmetn for MySQL.

* address comments.

* address comments.

* add comments for null check.
diff --git a/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlConfig.java b/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlConfig.java
@@ -20,6 +20,7 @@
 import io.cdap.cdap.api.annotation.Macro;
 import io.cdap.cdap.api.plugin.PluginConfig;
 
+import java.util.Properties;
 import javax.annotation.Nullable;
 
 /**
@@ -98,4 +99,16 @@ public String getJdbcPluginName() {
   public String getJDBCPluginId() {
     return String.format("%s.%s.%s", "mysqlsource", "jbdc", jdbcPluginName);
   }
+
+  public String getJdbcURL() {
+    return String.format("jdbc:mysql://%s:%d/%s", host, port, database);
+  }
+
+  public Properties getConnectionProperties() {
+    Properties properties = new Properties();
+    properties.put("user", user);
+    properties.put("password", password);
+    properties.put("serverTimezone", getServerTimezone());
+    return properties;
+  }
 }
diff --git a/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlDeltaSource.java b/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlDeltaSource.java
@@ -32,6 +32,7 @@
 import io.cdap.delta.plugin.common.DriverCleanup;
 
 import java.sql.Driver;
+import java.util.UUID;
 
 /**
  * Mysql origin.
@@ -61,24 +62,24 @@ public EventReader createReader(EventReaderDefinition definition, DeltaSourceCon
   }
 
   @Override
-  public TableRegistry createTableRegistry(Configurer configurer) {
+  public TableRegistry createTableRegistry(Configurer configurer) throws Exception {
+    return new MySqlTableRegistry(conf, getDriverCleanup(configurer));
+  }
+
+  @Override
+  public TableAssessor<TableDetail> createTableAssessor(Configurer configurer) throws Exception {
+    return new MySqlTableAssessor(conf, getDriverCleanup(configurer));
+  }
+
+  private DriverCleanup getDriverCleanup(Configurer configurer) throws Exception {
     Class<? extends Driver> jdbcDriverClass = configurer.usePluginClass("jdbc", conf.getJdbcPluginName(),
-                                                                        conf.getJDBCPluginId(),
+                                                                        conf.getJDBCPluginId() + "." +
+                                                                          UUID.randomUUID().toString(),
                                                                         PluginProperties.builder().build());
     if (jdbcDriverClass == null) {
       throw new IllegalArgumentException("JDBC plugin " + conf.getJdbcPluginName() + " not found.");
     }
-    try {
-      DriverCleanup cleanup = DriverCleanup.ensureJDBCDriverIsAvailable(
-        jdbcDriverClass, String.format("jdbc:mysql://%s:%d/%s", conf.getHost(), conf.getPort(), conf.getDatabase()));
-      return new MySqlTableRegistry(conf, cleanup);
-    } catch (Exception e) {
-      throw new RuntimeException("Unable to instantiate JDBC driver", e);
-    }
-  }
 
-  @Override
-  public TableAssessor<TableDetail> createTableAssessor(Configurer configurer) throws Exception {
-    return new MySqlTableAssessor();
+    return DriverCleanup.ensureJDBCDriverIsAvailable(jdbcDriverClass, conf.getJdbcURL());
   }
 }
diff --git a/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlTableAssessor.java b/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlTableAssessor.java
@@ -17,15 +17,23 @@
 package io.cdap.delta.mysql;
 
 import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.delta.api.assessment.Assessment;
 import io.cdap.delta.api.assessment.ColumnAssessment;
 import io.cdap.delta.api.assessment.ColumnDetail;
 import io.cdap.delta.api.assessment.ColumnSuggestion;
 import io.cdap.delta.api.assessment.ColumnSupport;
+import io.cdap.delta.api.assessment.Problem;
 import io.cdap.delta.api.assessment.TableAssessment;
 import io.cdap.delta.api.assessment.TableAssessor;
 import io.cdap.delta.api.assessment.TableDetail;
 import io.cdap.delta.plugin.common.ColumnEvaluation;
+import io.cdap.delta.plugin.common.DriverCleanup;
 
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
 import java.sql.Types;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -39,6 +47,29 @@ public class MySqlTableAssessor implements TableAssessor<TableDetail> {
   static final String COLUMN_LENGTH = "COLUMN_LENGTH";
   static final String SCALE = "SCALE";
 
+  private final MySqlConfig conf;
+  private final DriverCleanup driverCleanup;
+
+  MySqlTableAssessor(MySqlConfig conf, DriverCleanup driverCleanup) {
+    this.conf = conf;
+    this.driverCleanup = driverCleanup;
+  }
+
+  @Override
+  public Assessment assess() {
+    List<Problem> featureProblems = new ArrayList<>();
+    List<String> permissions = new ArrayList<>();
+    permissions.add("REPLICATION SLAVE");
+    permissions.add("REPLICATION CLIENT");
+    checkReplicationPermissions(featureProblems, permissions);
+    return new Assessment(featureProblems, Collections.emptyList());
+  }
+
+  @Override
+  public void close() throws IOException {
+    driverCleanup.close();
+  }
+
   @Override
   public TableAssessment assess(TableDetail tableDetail) {
     List<ColumnAssessment> columnAssessments = new ArrayList<>();
@@ -126,4 +157,46 @@ static ColumnEvaluation evaluateColumn(ColumnDetail detail) throws IllegalArgume
       .build();
     return new ColumnEvaluation(field, assessment);
   }
+
+  private void checkReplicationPermissions(List<Problem> featureProblems, List<String> permissions) {
+    String query = String.format("SHOW GRANTS FOR '%s'@'%s'", conf.getUser(), conf.getHost());
+    Problem permissionUnknownProblem =
+      new Problem("Table Replication Permission Unknown",
+                  String.format("Unable to check if '%s' permissions were granted for user '%s'@'%s' or not",
+                                String.join(",", permissions), conf.getUser(), conf.getHost()),
+                  "Check database connectivity and user permissions",
+                  "Change events might fail to be read after the snapshot phase");
+    try (Connection connection = DriverManager.getConnection(conf.getJdbcURL(), conf.getConnectionProperties());
+         Statement statement = connection.createStatement();
+         ResultSet rs = statement.executeQuery(query)) {
+      if (rs.next()) {
+        String grants = rs.getString(1);
+
+        // Avoid potential NPE in case the connection was accidentally closed by database before the value is read,
+        // though it will happen very rarely
+        if (grants == null) {
+          featureProblems.add(permissionUnknownProblem);
+          return;
+        }
+
+        if (grants.contains("ALL")) {
+          return;
+        }
+
+        for (String permission : permissions) {
+          if (!grants.contains(permission)) {
+            featureProblems.add(
+              new Problem("Table Replication Permission Not Granted",
+                          String.format("The '%s' permission is not granted for user '%s'@'%s'", permission,
+                                        conf.getUser(), conf.getHost()),
+                          String.format("Grant '%s' permission to user '%s'@'%s'", permission,
+                                        conf.getUser(), conf.getHost()),
+                          "Will not be able to read replication events after the initial snapshot completes."));
+          }
+        }
+      }
+    } catch (Exception e) {
+      featureProblems.add(permissionUnknownProblem);
+    }
+  }
 }
diff --git a/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlTableRegistry.java b/mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlTableRegistry.java
@@ -40,29 +40,24 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Properties;
 
 /**
  * Lists and describes tables.
  */
 public class MySqlTableRegistry implements TableRegistry {
   private final MySqlConfig conf;
   private final DriverCleanup driverCleanup;
-  private final Properties properties;
 
   public MySqlTableRegistry(MySqlConfig conf, DriverCleanup driverCleanup) {
     this.conf = conf;
     this.driverCleanup = driverCleanup;
-    this.properties = new Properties();
-    properties.put("user", conf.getUser());
-    properties.put("password", conf.getPassword());
-    properties.put("serverTimezone", conf.getServerTimezone());
   }
 
   @Override
   public TableList listTables() throws IOException {
     List<TableSummary> tables = new ArrayList<>();
-    try (Connection connection = DriverManager.getConnection(getConnectionString(conf.getDatabase()), properties)) {
+    try (Connection connection = DriverManager.getConnection(getConnectionString(conf.getDatabase()),
+                                                             conf.getConnectionProperties())) {
       DatabaseMetaData dbMeta = connection.getMetaData();
       try (ResultSet tableResults = dbMeta.getTables(null, null, null, null)) {
         while (tableResults.next()) {
@@ -85,7 +80,7 @@ public TableList listTables() throws IOException {
 
   @Override
   public TableDetail describeTable(String db, String table) throws TableNotFoundException, IOException {
-    try (Connection connection = DriverManager.getConnection(getConnectionString(db), properties)) {
+    try (Connection connection = DriverManager.getConnection(getConnectionString(db), conf.getConnectionProperties())) {
       DatabaseMetaData dbMeta = connection.getMetaData();
       TableDetail.Builder builder = getTableDetailBuilder(dbMeta, db, table)
         .orElseThrow(() -> new TableNotFoundException(db, table, ""));
diff --git a/sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerDeltaSource.java b/sqlserver-delta-plugins/src/main/java/io/cdap/delta/sqlserver/SqlServerDeltaSource.java
@@ -60,20 +60,17 @@ public EventReader createReader(EventReaderDefinition definition, DeltaSourceCon
   }
 
   @Override
-  public TableRegistry createTableRegistry(Configurer configurer) {
+  public TableRegistry createTableRegistry(Configurer configurer) throws Exception {
     Class<? extends Driver> jdbcDriverClass = configurer.usePluginClass("jdbc", config.getJdbcPluginName(),
                                                                         config.getJDBCPluginId(),
                                                                         PluginProperties.builder().build());
     if (jdbcDriverClass == null) {
       throw new IllegalArgumentException("JDBC plugin " + config.getJdbcPluginName() + " not found.");
     }
-    try {
-      DriverCleanup cleanup = DriverCleanup.ensureJDBCDriverIsAvailable(
-        jdbcDriverClass, String.format("jdbc:sqlserver://%s:%d", config.getHost(), config.getPort()));
-      return new SqlServerTableRegistry(config, cleanup);
-    } catch (Exception e) {
-      throw new RuntimeException("Unable to instantiate JDBC driver", e);
-    }
+
+    DriverCleanup cleanup = DriverCleanup.ensureJDBCDriverIsAvailable(
+      jdbcDriverClass, String.format("jdbc:sqlserver://%s:%d", config.getHost(), config.getPort()));
+    return new SqlServerTableRegistry(config, cleanup);
   }
 
   @Override
