blue new connections

add RDS PG support; fixes
fix IAM and routing
green change node name
APG support
switchover rollback
fix unit tests
codestyle
tags
separate host list provider per monitor
review todo
fixes for Aurora
refactoring
IAM fixes; summary
fixes; BG switchover summary; IAM node changed name
resources tier down
fix integration test suite
bgdId
smart waiting
routing verified
bg monitors
IAM reroute to IP and a proper token
useBlueNode flag in BG status
iam, dialect identification
Blue/Green Deployments for MultiAz clusters

Author: sergiyvamz

wrapper/build.gradle.kts

@@ -35,6 +35,7 @@
 import software.amazon.jdbc.plugin.DriverMetaDataConnectionPluginFactory;
 import software.amazon.jdbc.plugin.ExecutionTimeConnectionPluginFactory;
 import software.amazon.jdbc.plugin.LogQueryConnectionPluginFactory;
+import software.amazon.jdbc.plugin.bluegreen.BlueGreenConnectionPluginFactory;
 import software.amazon.jdbc.plugin.customendpoint.CustomEndpointPluginFactory;
 import software.amazon.jdbc.plugin.dev.DeveloperConnectionPluginFactory;
 import software.amazon.jdbc.plugin.efm.HostMonitoringConnectionPluginFactory;
@@ -82,6 +83,7 @@ public class ConnectionPluginChainBuilder {
           put("fastestResponseStrategy", FastestResponseStrategyPluginFactory.class);
           put("initialConnection", AuroraInitialConnectionStrategyPluginFactory.class);
           put("limitless", LimitlessConnectionPluginFactory.class);
+          put("bg", BlueGreenConnectionPluginFactory.class);
         }
       };
 
@@ -99,6 +101,7 @@ public class ConnectionPluginChainBuilder {
           put(AuroraInitialConnectionStrategyPluginFactory.class, 390);
           put(AuroraConnectionTrackerPluginFactory.class, 400);
           put(AuroraStaleDnsPluginFactory.class, 500);
+          put(BlueGreenConnectionPluginFactory.class, 550);
           put(ReadWriteSplittingPluginFactory.class, 600);
           put(FailoverConnectionPluginFactory.class, 700);
           put(software.amazon.jdbc.plugin.failover2.FailoverConnectionPluginFactory.class, 710);

wrapper/src/main/java/software/amazon/jdbc/ConnectionPluginChainBuilder.java

@@ -221,7 +221,7 @@ protected <T, E extends Exception> T executeWithSubscribedPlugins(
       throw new IllegalArgumentException("jdbcMethodFunc");
     }
 
-    // noinspection unchecked
+    @SuppressWarnings({"unchecked", "noinspection"})
     PluginChainJdbcCallable<T, E> pluginChainFunc = this.pluginChainFuncMap.get(methodName);
 
     if (pluginChainFunc == null) {
@@ -641,7 +641,7 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
     return false;
   }
 
-  public ConnectionProvider getDefaultConnProvider() {
+  public @NonNull ConnectionProvider getDefaultConnProvider() {
     return this.defaultConnProvider;
   }
 

wrapper/src/main/java/software/amazon/jdbc/ConnectionPluginManager.java

@@ -80,6 +80,8 @@ EnumSet<NodeChangeOptions> setCurrentConnection(
 
   HostSpec getInitialConnectionHostSpec();
 
+  String getOriginalUrl();
+
   /**
    * Set the collection of hosts that should be allowed and/or blocked for connections.
    *
@@ -249,4 +251,14 @@ Connection forceConnect(
   @NonNull SessionStateService getSessionStateService();
 
   <T> T getPlugin(final Class<T> pluginClazz);
+
+  <T> void setStatus(final Class<T> clazz, final @Nullable T status, final boolean clusterBound);
+
+  <T> void setStatus(final Class<T> clazz, final @Nullable T status, final String key);
+
+  <T> T getStatus(final @NonNull Class<T> clazz, final boolean clusterBound);
+
+  <T> T getStatus(final @NonNull Class<T> clazz, final String key);
+
+  boolean isPluginInUse(final Class<? extends ConnectionPlugin> pluginClazz);
 }

wrapper/src/main/java/software/amazon/jdbc/PluginService.java

@@ -63,6 +63,10 @@ public class PluginServiceImpl implements PluginService, CanReleaseResources,
   protected static final long DEFAULT_HOST_AVAILABILITY_CACHE_EXPIRE_NANO = TimeUnit.MINUTES.toNanos(5);
 
   protected static final CacheMap<String, HostAvailability> hostAvailabilityExpiringCache = new CacheMap<>();
+
+  protected static final CacheMap<String, Object> statusesExpiringCache = new CacheMap<>();
+  protected static final long DEFAULT_STATUS_CACHE_EXPIRE_NANO = TimeUnit.MINUTES.toNanos(60);
+
   protected final ConnectionPluginManager pluginManager;
   private final Properties props;
   private final String originalUrl;
@@ -205,6 +209,11 @@ public HostSpec getInitialConnectionHostSpec() {
     return this.initialConnectionHostSpec;
   }
 
+  @Override
+  public String getOriginalUrl() {
+    return this.originalUrl;
+  }
+
   @Override
   public void setAllowedAndBlockedHosts(AllowedAndBlockedHosts allowedAndBlockedHosts) {
     this.allowedAndBlockedHosts.set(allowedAndBlockedHosts);
@@ -716,6 +725,7 @@ public void updateDialect(final @NonNull Connection connection) throws SQLExcept
 
     final HostListProviderSupplier supplier = this.dialect.getHostListProvider();
     this.setHostListProvider(supplier.getProvider(props, this.originalUrl, this, this));
+    this.refreshHostList(connection);
   }
 
   @Override
@@ -790,4 +800,53 @@ public <T> T getPlugin(final Class<T> pluginClazz) {
   public static void clearCache() {
     hostAvailabilityExpiringCache.clear();
   }
+
+  public <T> void setStatus(final Class<T> clazz, final @Nullable T status, final boolean clusterBound) {
+    String clusterId = null;
+    if (clusterBound) {
+      try {
+        clusterId = this.hostListProvider.getClusterId();
+      } catch (Exception ex) {
+        // do nothing
+      }
+    }
+    this.setStatus(clazz, status, clusterId);
+  }
+
+  public <T> void setStatus(final Class<T> clazz, final @Nullable T status, final String key) {
+    final String cacheKey = this.getStatusCacheKey(clazz, key);
+    if (status == null) {
+      statusesExpiringCache.remove(cacheKey);
+    } else {
+      statusesExpiringCache.put(cacheKey, status, DEFAULT_STATUS_CACHE_EXPIRE_NANO);
+    }
+  }
+
+  public <T> T getStatus(final @NonNull Class<T> clazz, final boolean clusterBound) {
+    String clusterId = null;
+    if (clusterBound) {
+      try {
+        clusterId = this.hostListProvider.getClusterId();
+      } catch (Exception ex) {
+        // do nothing
+      }
+    }
+    return this.getStatus(clazz, clusterId);
+  }
+
+  public <T> T getStatus(final @NonNull Class<T> clazz, String key) {
+    return clazz.cast(statusesExpiringCache.get(this.getStatusCacheKey(clazz, key)));
+  }
+
+  protected <T> String getStatusCacheKey(final Class<T> clazz, final String key) {
+    return String.format("%s::%s", key == null ? "" : key.trim().toLowerCase(), clazz.getName());
+  }
+
+  public boolean isPluginInUse(final Class<? extends ConnectionPlugin> pluginClazz) {
+    try {
+      return this.pluginManager.isWrapperFor(pluginClazz);
+    } catch (SQLException e) {
+      return false;
+    }
+  }
 }

wrapper/src/main/java/software/amazon/jdbc/PluginServiceImpl.java

@@ -26,7 +26,7 @@
 import software.amazon.jdbc.hostlistprovider.monitoring.MonitoringRdsHostListProvider;
 import software.amazon.jdbc.plugin.failover2.FailoverConnectionPlugin;
 
-public class AuroraMysqlDialect extends MysqlDialect {
+public class AuroraMysqlDialect extends MysqlDialect implements SupportBlueGreen {
 
   private static final String TOPOLOGY_QUERY =
       "SELECT SERVER_ID, CASE WHEN SESSION_ID = 'MASTER_SESSION_ID' THEN TRUE ELSE FALSE END, "
@@ -42,6 +42,13 @@ public class AuroraMysqlDialect extends MysqlDialect {
   private static final String NODE_ID_QUERY = "SELECT @@aurora_server_id";
   private static final String IS_READER_QUERY = "SELECT @@innodb_read_only";
 
+  private static final String BG_STATUS_QUERY =
+      "SELECT * FROM mysql.rds_topology";
+
+  private static final String TOPOLOGY_TABLE_EXIST_QUERY =
+      "SELECT 1 AS tmp FROM information_schema.tables WHERE"
+          + " table_schema = 'mysql' AND table_name = 'rds_topology'";
+
   @Override
   public boolean isDialect(final Connection connection) {
     Statement stmt = null;
@@ -105,5 +112,23 @@ public HostListProviderSupplier getHostListProvider() {
           IS_READER_QUERY);
     };
   }
+
+  @Override
+  public String getBlueGreenStatusQuery() {
+    return BG_STATUS_QUERY;
+  }
+
+  @Override
+  public boolean isStatusAvailable(final Connection connection) {
+    try {
+      try (Statement statement = connection.createStatement();
+          ResultSet rs = statement.executeQuery(TOPOLOGY_TABLE_EXIST_QUERY)) {
+        return rs.next();
+      }
+    } catch (SQLException ex) {
+      return false;
+    }
+  }
+
 }
 

wrapper/src/main/java/software/amazon/jdbc/dialect/AuroraMysqlDialect.java

@@ -24,12 +24,13 @@
 import software.amazon.jdbc.hostlistprovider.AuroraHostListProvider;
 import software.amazon.jdbc.hostlistprovider.monitoring.MonitoringRdsHostListProvider;
 import software.amazon.jdbc.plugin.failover2.FailoverConnectionPlugin;
+import software.amazon.jdbc.util.DriverInfo;
 
 /**
  * Suitable for the following AWS PG configurations.
  * - Regional Cluster
  */
-public class AuroraPgDialect extends PgDialect implements AuroraLimitlessDialect {
+public class AuroraPgDialect extends PgDialect implements AuroraLimitlessDialect, SupportBlueGreen {
   private static final Logger LOGGER = Logger.getLogger(AuroraPgDialect.class.getName());
 
   private static final String extensionsSql =
@@ -56,6 +57,12 @@ public class AuroraPgDialect extends PgDialect implements AuroraLimitlessDialect
   protected static final String LIMITLESS_ROUTER_ENDPOINT_QUERY =
       "select router_endpoint, load from aurora_limitless_router_endpoints()";
 
+  private static final String BG_STATUS_QUERY =
+      "SELECT * FROM get_blue_green_fast_switchover_metadata('aws_jdbc_driver-" + DriverInfo.DRIVER_VERSION + "')";
+
+  private static final String TOPOLOGY_TABLE_EXIST_QUERY =
+      "SELECT 'get_blue_green_fast_switchover_metadata'::regproc";
+
   @Override
   public boolean isDialect(final Connection connection) {
     if (!super.isDialect(connection)) {
@@ -156,4 +163,21 @@ public HostListProviderSupplier getHostListProvider() {
   public String getLimitlessRouterEndpointQuery() {
     return LIMITLESS_ROUTER_ENDPOINT_QUERY;
   }
+
+  @Override
+  public String getBlueGreenStatusQuery() {
+    return BG_STATUS_QUERY;
+  }
+
+  @Override
+  public boolean isStatusAvailable(final Connection connection) {
+    try {
+      try (Statement statement = connection.createStatement();
+          ResultSet rs = statement.executeQuery(TOPOLOGY_TABLE_EXIST_QUERY)) {
+        return rs.next();
+      }
+    } catch (SQLException ex) {
+      return false;
+    }
+  }
 }

wrapper/src/main/java/software/amazon/jdbc/dialect/AuroraPgDialect.java

@@ -30,6 +30,8 @@
 import software.amazon.jdbc.plugin.failover.FailoverRestriction;
 import software.amazon.jdbc.plugin.failover2.FailoverConnectionPlugin;
 import software.amazon.jdbc.util.DriverInfo;
+import software.amazon.jdbc.util.RdsUtils;
+import software.amazon.jdbc.util.StringUtils;
 
 public class RdsMultiAzDbClusterMysqlDialect extends MysqlDialect {
 
@@ -50,41 +52,36 @@ public class RdsMultiAzDbClusterMysqlDialect extends MysqlDialect {
   private static final EnumSet<FailoverRestriction> RDS_MULTI_AZ_RESTRICTIONS =
       EnumSet.of(FailoverRestriction.DISABLE_TASK_A, FailoverRestriction.ENABLE_WRITER_IN_TASK_B);
 
+  protected final RdsUtils rdsUtils = new RdsUtils();
+
   @Override
   public boolean isDialect(final Connection connection) {
-    Statement stmt = null;
-    ResultSet rs = null;
     try {
-      stmt = connection.createStatement();
-      rs = stmt.executeQuery(TOPOLOGY_TABLE_EXIST_QUERY);
-
-      if (rs.next()) {
-        rs.close();
-        stmt.close();
-
-        stmt = connection.createStatement();
-        rs = stmt.executeQuery(TOPOLOGY_QUERY);
-
-        return rs.next();
+      try (Statement stmt = connection.createStatement();
+          ResultSet rs = stmt.executeQuery(TOPOLOGY_TABLE_EXIST_QUERY)) {
+        if (!rs.next()) {
+          return false;
+        }
       }
-      return false;
-    } catch (final SQLException ex) {
-      // ignore
-    } finally {
-      if (rs != null) {
-        try {
-          rs.close();
-        } catch (SQLException ex) {
-          // ignore
+
+      try (Statement stmt = connection.createStatement();
+          ResultSet rs = stmt.executeQuery(TOPOLOGY_QUERY)) {
+        if (!rs.next()) {
+          return false;
         }
       }
-      if (stmt != null) {
-        try {
-          stmt.close();
-        } catch (SQLException ex) {
-          // ignore
+
+      try (Statement stmt = connection.createStatement();
+          ResultSet rs = stmt.executeQuery("SHOW VARIABLES LIKE 'report_host'")) {
+        if (!rs.next()) {
+          return false;
         }
+        final String reportHost = rs.getString(2); // get variable value; expected value is IP address
+        return !StringUtils.isNullOrEmpty(reportHost);
       }
+
+    } catch (final SQLException ex) {
+      // ignore
     }
     return false;
   }