Adding resiliency in RedisClusterClient

(handle tcp keepidle in failover scenario, auth with ssl & handle cluster
details endpoint returning IP in ssl scenario.

Signed-off-by: Andrija Perovic <[email protected]>

Author: andrijaperovic

Diff for: Makefile

@@ -71,10 +71,10 @@ push-serving-docker:
 	docker push $(REGISTRY)/feast-serving:$(VERSION)
 
 build-core-docker:
-	docker build --build-arg VERSION=$(VERSION) -t $(REGISTRY)/feast-core:$(VERSION) -f infra/docker/core/Dockerfile .
+	docker build --no-cache --build-arg VERSION=$(VERSION) -t $(REGISTRY)/feast-core:$(VERSION) -f infra/docker/core/Dockerfile .
 
 build-serving-docker:
-	docker build --build-arg VERSION=$(VERSION) -t $(REGISTRY)/feast-serving:$(VERSION) -f infra/docker/serving/Dockerfile .
+	docker build --no-cache --build-arg VERSION=$(VERSION) -t $(REGISTRY)/feast-serving:$(VERSION) -f infra/docker/serving/Dockerfile .
 
 # Versions
 

Diff for: serving/src/main/java/feast/serving/config/FeastProperties.java

@@ -322,7 +322,9 @@ public RedisClusterStoreConfig getRedisClusterConfig() {
       return new RedisClusterStoreConfig(
           this.config.get("connection_string"),
           ReadFrom.valueOf(this.config.get("read_from")),
-          Duration.parse(this.config.get("timeout")));
+          Duration.parse(this.config.get("timeout")),
+          Boolean.valueOf(this.config.getOrDefault("ssl", "false")),
+          this.config.getOrDefault("password", ""));
     }
 
     public RedisStoreConfig getRedisConfig() {

Diff for: storage/connectors/redis/src/main/java/feast/storage/connectors/redis/retriever/RedisClusterClient.java

@@ -22,8 +22,17 @@
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
 import io.lettuce.core.codec.ByteArrayCodec;
+import io.lettuce.core.resource.ClientResources;
+import io.lettuce.core.resource.DnsResolvers;
+import io.lettuce.core.resource.MappingSocketAddressResolver;
+import io.lettuce.core.resource.NettyCustomizer;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.epoll.EpollChannelOption;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 public class RedisClusterClient implements RedisClientAdapter {
@@ -62,18 +71,81 @@ private RedisClusterClient(Builder builder) {
     this.asyncCommands.setAutoFlushCommands(false);
   }
 
+  public static String getAddressString(String host) {
+    try {
+      return InetAddress.getByName(host).getHostAddress();
+    } catch (UnknownHostException e) {
+      throw new RuntimeException(String.format("getAllByName() failed: %s", e.getMessage()));
+    }
+  }
+
+  public static MappingSocketAddressResolver customSocketAddressResolver(
+      RedisClusterStoreConfig config) {
+
+    List<String> configuredHosts =
+        Arrays.stream(config.getConnectionString().split(","))
+            .map(
+                hostPort -> {
+                  return hostPort.trim().split(":")[0];
+                })
+            .collect(Collectors.toList());
+
+    Map<String, String> mapAddressHost =
+        configuredHosts.stream()
+            .collect(
+                Collectors.toMap(host -> ((String) getAddressString(host)), host -> (String) host));
+
+    return MappingSocketAddressResolver.create(
+        DnsResolvers.UNRESOLVED,
+        hostAndPort -> 
+            mapAddressHost.keySet().stream().anyMatch(i -> i.equals(hostAndPort.getHostText()))
+                ? hostAndPort.of(
+                    mapAddressHost.get(hostAndPort.getHostText()), hostAndPort.getPort())
+                : hostAndPort);
+  }
+
+  public static ClientResources customClientResources(RedisClusterStoreConfig config) {
+    ClientResources clientResources =
+        ClientResources.builder()
+            .nettyCustomizer(
+                new NettyCustomizer() {
+                  @Override
+                  public void afterBootstrapInitialized(Bootstrap bootstrap) {
+                    bootstrap.option(EpollChannelOption.TCP_KEEPIDLE, 15);
+                    bootstrap.option(EpollChannelOption.TCP_KEEPINTVL, 5);
+                    bootstrap.option(EpollChannelOption.TCP_KEEPCNT, 3);
+                    // Socket Timeout (milliseconds)
+                    bootstrap.option(EpollChannelOption.TCP_USER_TIMEOUT, 60000);
+                  }
+                })
+            .socketAddressResolver(customSocketAddressResolver(config))
+            .build();
+    return clientResources;
+  }
+
   public static RedisClientAdapter create(RedisClusterStoreConfig config) {
+
     List<RedisURI> redisURIList =
         Arrays.stream(config.getConnectionString().split(","))
             .map(
                 hostPort -> {
                   String[] hostPortSplit = hostPort.trim().split(":");
-                  return RedisURI.create(hostPortSplit[0], Integer.parseInt(hostPortSplit[1]));
+                  RedisURI redisURI =
+                      RedisURI.create(hostPortSplit[0], Integer.parseInt(hostPortSplit[1]));
+                  if (!config.getPassword().isEmpty()) {
+                    redisURI.setPassword(config.getPassword());
+                  }
+                  if (config.getSsl()) {
+                    redisURI.setSsl(true);
+                  }
+                  return redisURI;
                 })
             .collect(Collectors.toList());
 
     io.lettuce.core.cluster.RedisClusterClient client =
-        io.lettuce.core.cluster.RedisClusterClient.create(redisURIList);
+        io.lettuce.core.cluster.RedisClusterClient.create(
+            customClientResources(config), redisURIList);
+
     client.setOptions(
         ClusterClientOptions.builder()
             .socketOptions(SocketOptions.builder().keepAlive(true).tcpNoDelay(true).build())

Diff for: storage/connectors/redis/src/main/java/feast/storage/connectors/redis/retriever/RedisClusterStoreConfig.java

@@ -23,11 +23,16 @@ public class RedisClusterStoreConfig {
   private final String connectionString;
   private final ReadFrom readFrom;
   private final Duration timeout;
+  private final String password;
+  private final Boolean ssl;
 
-  public RedisClusterStoreConfig(String connectionString, ReadFrom readFrom, Duration timeout) {
+  public RedisClusterStoreConfig(
+      String connectionString, ReadFrom readFrom, Duration timeout, Boolean ssl, String password) {
     this.connectionString = connectionString;
     this.readFrom = readFrom;
     this.timeout = timeout;
+    this.password = password;
+    this.ssl = ssl;
   }
 
   public String getConnectionString() {
@@ -41,4 +46,12 @@ public ReadFrom getReadFrom() {
   public Duration getTimeout() {
     return this.timeout;
   }
+
+  public String getPassword() {
+    return this.password;
+  }
+
+  public Boolean getSsl() {
+    return this.ssl;
+  }
 }