feat: Allowing password based authentication and SSL for Redis in Jav…

…a feature server

* Enabling password authentication and SSL in Redis clusters for the Java feature server

Signed-off-by: Danny Chiao <danny@tecton.ai>

* lint

Signed-off-by: Danny Chiao <danny@tecton.ai>

* add test

Signed-off-by: Danny Chiao <danny@tecton.ai>

* Address comments

Signed-off-by: Danny Chiao <danny@tecton.ai>

* Address comments

Signed-off-by: Danny Chiao <danny@tecton.ai>

* lint

Signed-off-by: Danny Chiao <danny@tecton.ai>

* lint

Signed-off-by: Danny Chiao <danny@tecton.ai>

.gitignore

@@ -191,3 +191,9 @@ sdk/go/protos/
 
 #benchmarks
 .benchmarks
+
+# Examples registry
+**/registry.db
+**/*.aof
+**/*.rdb
+**/nodes.conf

java/serving/src/main/java/feast/serving/config/ApplicationProperties.java

@@ -33,9 +33,12 @@
 import javax.validation.*;
 import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
+import org.slf4j.Logger;
 
 /** Feast Serving properties. */
 public class ApplicationProperties {
+  private static final Logger log = org.slf4j.LoggerFactory.getLogger(ApplicationProperties.class);
+
   public static class FeastProperties {
     /* Feast Serving build version */
     @NotBlank private String version = "unknown";
@@ -246,10 +249,33 @@ public void setType(String type) {
      * @return Returns the store specific configuration
      */
     public RedisClusterStoreConfig getRedisClusterConfig() {
+      String read_from;
+      if (!this.config.containsKey("read_from") || this.config.get("read_from") == null) {
+        log.info("'read_from' not defined in Redis cluster config, so setting to UPSTREAM");
+        read_from = ReadFrom.UPSTREAM.toString();
+      } else {
+        read_from = this.config.get("read_from");
+      }
+
+      if (!this.config.containsKey("timeout") || this.config.get("timeout") == null) {
+        throw new IllegalArgumentException(
+            "Redis cluster config does not have 'timeout' specified");
+      }
+
+      Boolean ssl = null;
+      if (!this.config.containsKey("ssl") || this.config.get("ssl") == null) {
+        log.info("'ssl' not defined in Redis cluster config, so setting to false");
+        ssl = false;
+      } else {
+        ssl = Boolean.parseBoolean(this.config.get("ssl"));
+      }
+      Duration timeout = Duration.parse(this.config.get("timeout"));
       return new RedisClusterStoreConfig(
           this.config.get("connection_string"),
-          ReadFrom.valueOf(this.config.get("read_from")),
-          Duration.parse(this.config.get("timeout")));
+          ReadFrom.valueOf(read_from),
+          timeout,
+          ssl,
+          this.config.getOrDefault("password", ""));
     }
 
     public RedisStoreConfig getRedisConfig() {

java/serving/src/test/java/feast/serving/it/TestUtils.java

@@ -83,7 +83,8 @@ public static ApplicationProperties.FeastProperties createBasicFeastProperties(
             new ApplicationProperties.Store(
                 "online",
                 "REDIS",
-                ImmutableMap.of("host", redisHost, "port", redisPort.toString()))));
+                ImmutableMap.of(
+                    "host", redisHost, "port", redisPort.toString(), "password", "testpw"))));
 
     return feastProperties;
   }

java/serving/src/test/resources/docker-compose/docker-compose-redis-it.yml

@@ -3,6 +3,7 @@ version: '3'
 services:
   redis:
     image: redis:6.2
+    command: redis-server --requirepass testpw
     ports:
       - "6379:6379"
   feast:

java/serving/src/test/resources/docker-compose/feast10/feature_store.yaml

@@ -3,7 +3,7 @@ registry: registry.db
 provider: local
 online_store:
   type: redis
-  connection_string: "redis:6379"
+  connection_string: "redis:6379,password=testpw"
 offline_store: {}
 flags:
   alpha_features: true

java/storage/connectors/redis/src/main/java/feast/storage/connectors/redis/retriever/RedisClusterClient.java

@@ -71,7 +71,16 @@ public static RedisClientAdapter create(RedisClusterStoreConfig config) {
             .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);
+                  }
+                  redisURI.setTimeout(config.getTimeout());
+                  return redisURI;
                 })
             .collect(Collectors.toList());
 

java/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 Boolean ssl;
+  private final String password;
 
-  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.ssl = ssl;
+    this.password = password;
   }
 
   public String getConnectionString() {
@@ -41,4 +46,12 @@ public ReadFrom getReadFrom() {
   public Duration getTimeout() {
     return this.timeout;
   }
+
+  public Boolean getSsl() {
+    return ssl;
+  }
+
+  public String getPassword() {
+    return password;
+  }
 }