Add several functionalities to the BanyanDBClient (#49)

Author: hanahmily

CHANGES.md

@@ -9,6 +9,9 @@ Release Notes.
 
 * Add mod revision check to write requests
 * Add TTL to property.
+* Support setting multiple server addresses 
+* Support DNS name resolution
+* Support round-robin load balancing
 
 0.4.0
 ------------------

README.md

@@ -14,15 +14,22 @@ The client implement for SkyWalking BanyanDB in Java.
 
 ## Create a client
 
-Create a `BanyanDBClient` with host, port and then use `connect()` to establish a connection.
+Create a `BanyanDBClient` with the server's several addresses and then use `connect()` to establish a connection.
 
 ```java
 // use `default` group
-client = new BanyanDBClient("127.0.0.1", 17912);
+BanyanDBClient client = new BanyanDBClient("banyandb.svc:17912", "10.0.12.9:17912");
 // to send any request, a connection to the server must be estabilished
 client.connect();
 ```
 
+These addresses are either IP addresses or DNS names. If DNS names are used, the client will resolve the DNS name to
+IP addresses and use them to connect to the server. The client will periodically refresh the IP addresses of the DNS
+name. The refresh interval can be configured by `resolveDNSInterval` option.
+
+The client will try to connect to the server in a round-robin manner. The client will periodically refresh the server
+addresses. The refresh interval can be configured by `refreshInterval` option.
+
 Besides, you may pass a customized options while building a `BanyanDBClient`. Supported
 options are listed below,
 
@@ -32,6 +39,7 @@ options are listed below,
 | maxInboundMessageSize      | Max inbound message size                                             | 1024 * 1024 * 50 (~50MB) |
 | deadline                   | Threshold of gRPC blocking query, unit is second                     | 30 (seconds)             |
 | refreshInterval            | Refresh interval for the gRPC channel, unit is second                | 30 (seconds)             |
+| resolveDNSInterval         | DNS resolve interval, unit is second                                 | 30 (minutes)             |
 | forceReconnectionThreshold | Threshold of force gRPC reconnection if network issue is encountered | 1                        |
 | forceTLS                   | Force use TLS for gRPC                                               | false                    |
 | sslTrustCAPath             | SSL: Trusted CA Path                                                 |                          |

src/main/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClient.java

@@ -28,7 +28,6 @@
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
-
 import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
 import org.apache.skywalking.banyandb.measure.v1.BanyandbMeasure;
 import org.apache.skywalking.banyandb.measure.v1.MeasureServiceGrpc;
@@ -59,7 +58,9 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.net.URI;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -83,14 +84,8 @@
  */
 @Slf4j
 public class BanyanDBClient implements Closeable {
-    /**
-     * The hostname of BanyanDB server.
-     */
-    private final String host;
-    /**
-     * The port of BanyanDB server.
-     */
-    private final int port;
+
+    private final String[] targets;
     /**
      * Options for server connection.
      */
@@ -138,23 +133,24 @@ public class BanyanDBClient implements Closeable {
     /**
      * Create a BanyanDB client instance with a default options.
      *
-     * @param host IP or domain name
-     * @param port Server port
+     * @param targets server targets
      */
-    public BanyanDBClient(String host, int port) {
-        this(host, port, new Options());
+    public BanyanDBClient(String... targets) {
+        this(targets, new Options());
     }
 
     /**
      * Create a BanyanDB client instance with a customized options.
      *
-     * @param host    IP or domain name
-     * @param port    Server port
+     * @param targets server targets
      * @param options customized options
      */
-    public BanyanDBClient(String host, int port, Options options) {
-        this.host = host;
-        this.port = port;
+    public BanyanDBClient(String[] targets, Options options) {
+        String[] tt = Preconditions.checkNotNull(targets, "targets");
+        checkState(tt.length > 0, "targets' size must be more than 1");
+        tt = Arrays.stream(tt).filter(t -> !Strings.isNullOrEmpty(t)).toArray(size -> new String[size]);
+        checkState(tt.length > 0, "valid targets' size must be more than 1");
+        this.targets = tt;
         this.options = options;
         this.connectionEstablishLock = new ReentrantLock();
         this.metadataCache = new MetadataCache();
@@ -169,8 +165,12 @@ public void connect() throws IOException {
         connectionEstablishLock.lock();
         try {
             if (!isConnected) {
+                URI[] addresses = new URI[this.targets.length];
+                for (int i = 0; i < this.targets.length; i++) {
+                        addresses[i] = URI.create("//" + this.targets[i]);
+                }
                 this.channel = ChannelManager.create(this.options.buildChannelManagerSettings(),
-                        new DefaultChannelFactory(this.host, this.port, this.options));
+                        new DefaultChannelFactory(addresses, this.options));
                 streamServiceBlockingStub = StreamServiceGrpc.newBlockingStub(this.channel);
                 measureServiceBlockingStub = MeasureServiceGrpc.newBlockingStub(this.channel);
                 streamServiceStub = StreamServiceGrpc.newStub(this.channel);

src/main/java/org/apache/skywalking/banyandb/v1/client/Options.java

@@ -45,6 +45,10 @@ public class Options {
      * Threshold of force gRPC reconnection if network issue is encountered
      */
     private long forceReconnectionThreshold = 1;
+    /**
+     * Threshold of resolving the DNS
+     */
+    private long resolveDNSInterval = 30 * 60;
     /**
      * Force use TLS for gRPC
      * Default is false

src/main/java/org/apache/skywalking/banyandb/v1/client/grpc/channel/ChannelManager.java

@@ -29,9 +29,7 @@
 import io.grpc.ManagedChannel;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
-import io.grpc.NameResolverRegistry;
 import io.grpc.Status;
-import io.grpc.internal.DnsNameResolverProvider;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
@@ -57,22 +55,18 @@ public class ChannelManager extends ManagedChannel {
     private final ScheduledExecutorService executor;
     @VisibleForTesting
     final AtomicReference<Entry> entryRef = new AtomicReference<>();
-    private final String authority;
 
     public static ChannelManager create(ChannelManagerSettings settings, ChannelFactory channelFactory)
             throws IOException {
         return new ChannelManager(settings, channelFactory, Executors.newSingleThreadScheduledExecutor());
     }
 
-    public ChannelManager(ChannelManagerSettings settings, ChannelFactory channelFactory, ScheduledExecutorService executor) throws IOException {
+    ChannelManager(ChannelManagerSettings settings, ChannelFactory channelFactory, ScheduledExecutorService executor) throws IOException {
         this.settings = settings;
         this.channelFactory = channelFactory;
         this.executor = executor;
 
-        NameResolverRegistry.getDefaultRegistry().register(new DnsNameResolverProvider());
-
         entryRef.set(new Entry(channelFactory.create()));
-        authority = entryRef.get().channel.authority();
 
         this.executor.scheduleAtFixedRate(
                 this::refreshSafely,
@@ -92,16 +86,17 @@ private void refreshSafely() {
 
     void refresh() throws IOException {
         Entry entry = entryRef.get();
-        if (entry.needReconnect) {
-            if (entry.isConnected(entry.reconnectCount.incrementAndGet() > this.settings.forceReconnectionThreshold())) {
-                // Reconnect to the same server is automatically done by GRPC
-                // clear the flags
-                entry.reset();
-            } else {
-                Entry replacedEntry = entryRef.getAndSet(new Entry(this.channelFactory.create()));
-                replacedEntry.shutdown();
-            }
+        if (!entry.needReconnect) {
+            return;
+        }
+        if (entry.isConnected(entry.reconnectCount.incrementAndGet() > this.settings.forceReconnectionThreshold())) {
+            // Reconnect to the same server is automatically done by GRPC
+            // clear the flags
+            entry.reset();
+            return;
         }
+        Entry replacedEntry = entryRef.getAndSet(new Entry(this.channelFactory.create()));
+        replacedEntry.shutdown();
     }
 
     @Override
@@ -157,7 +152,7 @@ public <REQ, RESP> ClientCall<REQ, RESP> newCall(MethodDescriptor<REQ, RESP> met
 
     @Override
     public String authority() {
-        return this.authority;
+        return this.entryRef.get().channel.authority();
     }
 
     @RequiredArgsConstructor
@@ -192,7 +187,7 @@ public <REQ, RESP> ClientCall<REQ, RESP> newCall(MethodDescriptor<REQ, RESP> met
 
         @Override
         public String authority() {
-            return authority;
+            return ChannelManager.this.authority();
         }
     }
 

src/main/java/org/apache/skywalking/banyandb/v1/client/grpc/channel/DefaultChannelFactory.java

@@ -24,6 +24,8 @@
 import io.grpc.netty.NegotiationType;
 import io.grpc.netty.NettyChannelBuilder;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.util.internal.PlatformDependent;
+import io.netty.util.internal.SocketUtils;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.banyandb.v1.client.Options;
@@ -33,17 +35,30 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.URI;
+import java.net.UnknownHostException;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.stream.Stream;
 
 @Slf4j
 @RequiredArgsConstructor
 public class DefaultChannelFactory implements ChannelFactory {
-    private final String host;
-    private final int port;
+    private final URI[] targets;
     private final Options options;
+    private SocketAddress[] addresses;
+    private long lastTargetsResolvedTime;
 
     @Override
     public ManagedChannel create() throws IOException {
-        NettyChannelBuilder managedChannelBuilder = NettyChannelBuilder.forAddress(this.host, this.port)
+        if (this.addresses == null ||
+                System.currentTimeMillis() - this.lastTargetsResolvedTime > this.options.getResolveDNSInterval()) {
+            resolveTargets();
+        }
+        NettyChannelBuilder managedChannelBuilder = NettyChannelBuilder.forAddress(resolveAddress())
                 .maxInboundMessageSize(options.getMaxInboundMessageSize())
                 .usePlaintext();
 
@@ -75,4 +90,31 @@ public ManagedChannel create() throws IOException {
         }
         return managedChannelBuilder.build();
     }
+
+    private void resolveTargets() {
+        this.addresses = Arrays.stream(this.targets)
+                .flatMap(target -> {
+                    try {
+                        return Arrays.stream(SocketUtils.allAddressesByName(target.getHost()))
+                                .map(InetAddress::getHostAddress)
+                                .map(ip -> new InetSocketAddress(ip, target.getPort()));
+                    } catch (Throwable t) {
+                        log.error("Failed to resolve the BanyanDB server's address ", t);
+                    }
+                    return Stream.empty();
+                })
+                .sorted(Comparator.comparing(InetSocketAddress::toString))
+                .distinct()
+                .toArray(InetSocketAddress[]::new);
+        this.lastTargetsResolvedTime = System.currentTimeMillis();
+    }
+
+    private SocketAddress resolveAddress() throws UnknownHostException {
+        int numAddresses = this.addresses.length;
+        if (numAddresses < 1) {
+            throw new UnknownHostException();
+        }
+        int offset = numAddresses == 1 ? 0 : PlatformDependent.threadLocalRandom().nextInt(numAddresses);
+        return this.addresses[offset];
+    }
 }

src/test/java/org/apache/skywalking/banyandb/v1/client/AbstractBanyanDBClientTest.java

@@ -354,7 +354,7 @@ protected void setUp(SetupFunction... setUpFunctions) throws IOException {
         this.channel = grpcCleanup.register(
                 InProcessChannelBuilder.forName(serverName).directExecutor().build());
 
-        client = new BanyanDBClient("127.0.0.1", s.getPort());
+        client = new BanyanDBClient(String.format("127.0.0.1:%d", s.getPort()));
         client.connect(this.channel);
     }
 

src/test/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClientTestCI.java

@@ -49,7 +49,7 @@ public class BanyanDBClientTestCI {
 
     protected void setUpConnection() throws IOException {
         log.info("create BanyanDB client and try to connect");
-        client = new BanyanDBClient(banyanDB.getHost(), banyanDB.getMappedPort(GRPC_PORT));
+        client = new BanyanDBClient(String.format("%s:%d", banyanDB.getHost(), banyanDB.getMappedPort(GRPC_PORT)));
         client.connect();
     }