JCBC-337: Implement a Combined Configuration Provider (Binary + HTTP)

This changeset provides a combined http and binary configuration provider. It is
designed to handle bootstrap and monitoring completely autonomously, further
decoupling it from the rest of the codebase.

Change-Id: I13b5029b98c9327cac6927ea35efe529edbb2f84
Reviewed-on: http://review.couchbase.org/32589
Reviewed-by: Michael Nitschinger <[email protected]>
Tested-by: Michael Nitschinger <[email protected]>

Author: daschl

src/main/java/com/couchbase/client/CouchbaseClient.java

@@ -263,24 +263,13 @@ public CouchbaseClient(CouchbaseConnectionFactory cf)
     executorService = cbConnFactory.getListenerExecutorService();
 
     getLogger().info(MODE_ERROR);
-    cf.getConfigurationProvider().subscribe(cf.getBucketName(), this);
+    cf.getConfigurationProvider().subscribe(this);
   }
 
   @Override
   public void reconfigure(Bucket bucket) {
     reconfiguring = true;
-    if (bucket.isNotUpdating()) {
-      getLogger().info("Bucket configuration is disconnected from cluster "
-        + "configuration updates, attempting to reconnect.");
-      CouchbaseConnectionFactory cbcf = (CouchbaseConnectionFactory)connFactory;
-      cbcf.requestConfigReconnect(cbcf.getBucketName(), this);
-      cbcf.checkConfigUpdate();
-    }
     try {
-      cbConnFactory.getConfigurationProvider().updateBucket(
-        cbConnFactory.getBucketName(), bucket);
-      cbConnFactory.updateStoredBaseList(bucket.getConfig());
-
       if(vconn != null) {
         vconn.reconfigure(bucket);
       }

src/main/java/com/couchbase/client/CouchbaseConnection.java

@@ -163,7 +163,7 @@ public void reconfigure(Bucket bucket) {
       // schedule shutdown for the oddNodes
       for(MemcachedNode shutDownNode : oddNodes) {
         getLogger().info("Scheduling Node "
-          + shutDownNode.getSocketAddress() + "for shutdown.");
+          + shutDownNode.getSocketAddress() + " for shutdown.");
       }
       nodesToShutdown.addAll(oddNodes);
     } catch (IOException e) {
@@ -313,7 +313,7 @@ public void run() {
   protected void handleRetryInformation(byte[] retryMessage) {
     String message = new String(retryMessage).trim();
     if (message.startsWith("{")) {
-      cf.getConfigurationProvider().updateBucket(
+      cf.getConfigurationProvider().setConfig(
         replaceConfigWildcards(message)
       );
     }
@@ -328,7 +328,7 @@ protected void handleRetryInformation(byte[] retryMessage) {
    * @param original the raw new config string.
    * @return the potentially changed config string.
    */
-  private String replaceConfigWildcards(String original) {
+  public String replaceConfigWildcards(String original) {
     if (original.contains("$HOST")) {
       ArrayList<MemcachedNode> nodes =
         new ArrayList<MemcachedNode>(getLocator().getAll());

src/main/java/com/couchbase/client/CouchbaseConnectionFactory.java

@@ -23,7 +23,8 @@
 package com.couchbase.client;
 
 import com.couchbase.client.vbucket.ConfigurationException;
-import com.couchbase.client.vbucket.ConfigurationProvider;
+import com.couchbase.client.vbucket.provider.BucketConfigurationProvider;
+import com.couchbase.client.vbucket.provider.ConfigurationProvider;
 import com.couchbase.client.vbucket.ConfigurationProviderHTTP;
 import com.couchbase.client.vbucket.CouchbaseNodeOrder;
 import com.couchbase.client.vbucket.Reconfigurable;
@@ -204,8 +205,6 @@ public CouchbaseConnectionFactory(final List<URI> baseList,
   }
 
   private void initialize(List<URI> baseList, String bucket, String password) {
-    potentiallyRandomizeNodeList(baseList);
-
     storedBaseList = new ArrayList<URI>();
     for (URI bu : baseList) {
       if (!bu.isAbsolute()) {
@@ -226,7 +225,7 @@ private void initialize(List<URI> baseList, String bucket, String password) {
     this.bucket = bucket;
     pass = password;
     configurationProvider =
-      new ConfigurationProviderHTTP(baseList, bucket, password);
+      new BucketConfigurationProvider(baseList, bucket, password, this);
   }
 
   @Override
@@ -310,25 +309,15 @@ public CouchbaseNodeOrder getStreamingNodeOrder() {
   }
 
   public Config getVBucketConfig() {
-    Bucket config = configurationProvider.getBucketConfiguration(bucket);
-    if(config == null) {
-      throw new ConfigurationException("Could not fetch valid configuration "
-        + "from provided nodes. Stopping.");
-    } else if (config.isNotUpdating()) {
-      LOGGER.warning("Noticed bucket configuration to be disconnected, "
-        + "will attempt to reconnect");
-      setConfigurationProvider(new ConfigurationProviderHTTP(storedBaseList,
-        bucket, pass));
-    }
-    return configurationProvider.getBucketConfiguration(bucket).getConfig();
+    return configurationProvider.getConfig().getConfig();
   }
 
   public synchronized ConfigurationProvider getConfigurationProvider() {
     return configurationProvider;
   }
 
   protected void requestConfigReconnect(String bucketName, Reconfigurable rec) {
-    configurationProvider.markForResubscribe(bucketName, rec);
+    configurationProvider.signalOutdated();
     needsReconnect = true;
   }
 
@@ -367,7 +356,7 @@ void checkConfigUpdate() {
       }
 
       if (doingResubscribe.compareAndSet(false, true)) {
-        resubConfigUpdate();
+        getConfigurationProvider().signalOutdated();
       } else {
         LOGGER.log(Level.CONFIG, "Duplicate resubscribe for config updates"
           + " suppressed.");
@@ -379,15 +368,6 @@ void checkConfigUpdate() {
     }
   }
 
-  /**
-   * Resubscribe for configuration updates.
-   */
-  private synchronized void resubConfigUpdate() {
-    LOGGER.log(Level.INFO, "Attempting to resubscribe for cluster config"
-      + " updates.");
-    resubExec.execute(new Resubscriber());
-  }
-
   /**
    * Checks if there have been more requests than allowed through
    * maxConfigCheck in a 10 second period.
@@ -466,51 +446,6 @@ int getMaxConfigCheck() {
     return maxConfigCheck;
   }
 
-  private class Resubscriber implements Runnable {
-
-    public void run() {
-      String threadNameBase = "Couchbase/Resubscriber (Status: ";
-      Thread.currentThread().setName(threadNameBase + "running)");
-      LOGGER.log(Level.CONFIG, "Resubscribing for {0} using base list {1}",
-        new Object[]{bucket, storedBaseList});
-
-      long reconnectAttempt = 0;
-      long backoffTime = 1000;
-      long maxWaitTime = 10000;
-      do {
-        try {
-          long waitTime = (reconnectAttempt++)*backoffTime;
-          if(reconnectAttempt >= 10) {
-            waitTime = maxWaitTime;
-          }
-          LOGGER.log(Level.INFO, "Reconnect attempt {0}, waiting {1}ms",
-            new Object[]{reconnectAttempt, waitTime});
-          Thread.sleep(waitTime);
-
-          ConfigurationProvider oldConfigProvider = getConfigurationProvider();
-          Reconfigurable oldRec = oldConfigProvider.getReconfigurable();
-
-          ConfigurationProvider newConfigProvider =
-            new ConfigurationProviderHTTP(storedBaseList, bucket, pass);
-          newConfigProvider.subscribe(bucket, oldRec);
-
-          setConfigurationProvider(newConfigProvider);
-          oldConfigProvider.shutdown();
-
-          if (!doingResubscribe.compareAndSet(true, false)) {
-            LOGGER.log(Level.WARNING,
-              "Could not reset from doing a resubscribe.");
-          }
-        } catch (Exception ex) {
-          LOGGER.log(Level.WARNING,
-            "Resubscribe attempt failed: ", ex);
-        }
-      } while(doingResubscribe.get());
-
-      Thread.currentThread().setName(threadNameBase + "complete)");
-    }
-  }
-
   /**
    * Returns a ClusterManager and initializes one if it does not exist.
    * @return Returns an instance of a ClusterManager.
@@ -522,35 +457,6 @@ public ClusterManager getClusterManager() {
     return clusterManager;
   }
 
-  /**
-   * Updates the stored base list with a new one based on the config.
-   *
-   * @param config
-   */
-  public void updateStoredBaseList(Config config) {
-    List<String> bucketServers = config.getRestEndpoints();
-    if (bucketServers.size() > 0) {
-      List<URI> newList = new ArrayList<URI>();
-      for (String bucketServer : bucketServers) {
-        try {
-          newList.add(new URI(bucketServer));
-        } catch(URISyntaxException ex) {
-          getLogger().warn("Could not add node to updated bucket list because "
-            + "of a parsing exception.");
-          getLogger().debug("Could not parse list because: " + ex);
-        }
-      }
-
-      if (nodeListsAreDifferent(storedBaseList, newList)) {
-        getLogger().info("Replacing current streaming node list "
-          + storedBaseList + " with " + newList);
-        potentiallyRandomizeNodeList(newList);
-        storedBaseList = newList;
-        getConfigurationProvider().updateBaseListFromConfig(newList);
-      }
-    }
-  }
-
   /**
    * Returns the current base list.
    *
@@ -560,39 +466,6 @@ List<URI> getStoredBaseList() {
     return storedBaseList;
   }
 
-  /**
-   * Randomizes the entries of the node list if needed.
-   *
-   * @param list the list to potentially randomize.
-   */
-  private void potentiallyRandomizeNodeList(List<URI> list) {
-    if (getStreamingNodeOrder().equals(CouchbaseNodeOrder.ORDERED)) {
-      return;
-    }
-
-    Collections.shuffle(list);
-  }
-
-  /**
-   * Check if two given node lists are different.
-   *
-   * @param left one node list
-   * @param right the other node list
-   * @return true if they are different, false otherwise.
-   */
-  private boolean nodeListsAreDifferent(List<URI> left, List<URI> right) {
-    if (left.size() != right.size()) {
-      return true;
-    }
-
-    for (URI uri : left) {
-      if (!right.contains(uri)) {
-        return true;
-      }
-    }
-    return false;
-  }
-
   @Override
   public String toString() {
     final StringBuilder sb = new StringBuilder("CouchbaseConnectionFactory{");

src/main/java/com/couchbase/client/TapConnectionProvider.java

@@ -22,7 +22,7 @@
 
 package com.couchbase.client;
 
-import com.couchbase.client.vbucket.ConfigurationProvider;
+import com.couchbase.client.vbucket.provider.ConfigurationProvider;
 import com.couchbase.client.vbucket.Reconfigurable;
 import com.couchbase.client.vbucket.config.Bucket;
 
@@ -71,7 +71,7 @@ public TapConnectionProvider(CouchbaseConnectionFactory cf) throws IOException {
     super(cf, AddrUtil.getAddresses(cf.getVBucketConfig().getServers()));
     this.cf=cf;
     ConfigurationProvider cp = cf.getConfigurationProvider();
-    cp.subscribe(cf.getBucketName(), this);
+    cp.subscribe(this);
   }
 
   /**

src/main/java/com/couchbase/client/vbucket/config/Bucket.java

@@ -117,4 +117,15 @@ public final void setIsNotUpdating() {
     LOGGER.finest("Marking bucket as not updating,"
       + " disconnected from config stream");
   }
+
+  @Override
+  public String toString() {
+    return "Bucket{" +
+      "name='" + name + '\'' +
+      ", configuration=" + configuration +
+      ", streamingURI=" + streamingURI +
+      ", isNotUpdating=" + isNotUpdating +
+      ", nodes=" + nodes +
+      '}';
+  }
 }

src/main/java/com/couchbase/client/vbucket/config/CacheConfig.java

@@ -156,4 +156,15 @@ public ConfigType getConfigType() {
   public List<URL> getCouchServers() {
     throw new UnsupportedOperationException("No couch port for cache buckets");
   }
+
+  @Override
+  public String toString() {
+    return "CacheConfig{" +
+      "vbucketsCount=" + vbucketsCount +
+      ", serversCount=" + serversCount +
+      ", servers=" + servers +
+      ", vbuckets=" + vbuckets +
+      ", restEndpoints=" + restEndpoints +
+      '}';
+  }
 }

src/main/java/com/couchbase/client/vbucket/config/DefaultConfig.java

@@ -238,4 +238,20 @@ public boolean nodeHasActiveVBuckets(InetSocketAddress node) {
   public ConfigType getConfigType() {
     return ConfigType.COUCHBASE;
   }
+
+  @Override
+  public String toString() {
+    return "DefaultConfig{" +
+      "hashAlgorithm=" + hashAlgorithm +
+      ", vbucketsCount=" + vbucketsCount +
+      ", mask=" + mask +
+      ", serversCount=" + serversCount +
+      ", replicasCount=" + replicasCount +
+      ", servers=" + servers +
+      ", vbuckets=" + vbuckets +
+      ", couchServers=" + couchServers +
+      ", serversWithVBuckets=" + serversWithVBuckets +
+      ", restEndpoints=" + restEndpoints +
+      '}';
+  }
 }

src/main/java/com/couchbase/client/vbucket/config/Node.java

@@ -82,4 +82,13 @@ public int hashCode() {
     result = 31 * result + ports.hashCode();
     return result;
   }
+
+  @Override
+  public String toString() {
+    return "Node{" +
+      "status=" + status +
+      ", hostname='" + hostname + '\'' +
+      ", ports=" + ports +
+      '}';
+  }
 }