Add Redis Cluster Support, (#502)

Co-authored-by: Khor Shu Heng <khor.heng@gojek.com>
Co-authored-by: Lavkesh Lahngir <lavkesh51@gmail.com>

Co-authored-by: Khor Shu Heng <khor.heng@gojek.com>

.prow/config.yaml

@@ -153,6 +153,19 @@ presubmits:
     skip_branches:
     - ^v0\.(3|4)-branch$
 
+  - name: test-end-to-end-redis-cluster
+    decorate: true
+    spec:
+      containers:
+        - image: maven:3.6-jdk-11
+          command: ["infra/scripts/test-end-to-end-redis-cluster.sh"]
+          resources:
+            requests:
+              cpu: "6"
+              memory: "6144Mi"
+    skip_branches:
+      - ^v0\.(3|4)-branch$
+
   - name: test-end-to-end-java-8
     decorate: true
     always_run: true

core/src/main/java/feast/core/model/Store.java

@@ -21,6 +21,7 @@
 import feast.core.StoreProto.Store.BigQueryConfig;
 import feast.core.StoreProto.Store.Builder;
 import feast.core.StoreProto.Store.CassandraConfig;
+import feast.core.StoreProto.Store.RedisClusterConfig;
 import feast.core.StoreProto.Store.RedisConfig;
 import feast.core.StoreProto.Store.StoreType;
 import feast.core.StoreProto.Store.Subscription;
@@ -82,6 +83,9 @@ public static Store fromProto(StoreProto.Store storeProto) throws IllegalArgumen
       case CASSANDRA:
         config = storeProto.getCassandraConfig().toByteArray();
         break;
+      case REDIS_CLUSTER:
+        config = storeProto.getRedisClusterConfig().toByteArray();
+        break;
       default:
         throw new IllegalArgumentException("Invalid store provided");
     }
@@ -106,6 +110,9 @@ public StoreProto.Store toProto() throws InvalidProtocolBufferException {
       case CASSANDRA:
         CassandraConfig cassConfig = CassandraConfig.parseFrom(config);
         return storeProtoBuilder.setCassandraConfig(cassConfig).build();
+      case REDIS_CLUSTER:
+        RedisClusterConfig redisClusterConfig = RedisClusterConfig.parseFrom(config);
+        return storeProtoBuilder.setRedisClusterConfig(redisClusterConfig).build();
       default:
         throw new InvalidProtocolBufferException("Invalid store set");
     }

infra/scripts/setup-redis-cluster.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+
+apt-get -y install redis-server > /var/log/redis.install.log
+
+mkdir 7000 7001 7002 7003 7004 7005
+for i in {0..5} ; do
+echo "port 700$i
+cluster-enabled yes
+cluster-config-file nodes-$i.conf
+cluster-node-timeout 5000
+appendonly yes" > 700$i/redis.conf
+redis-server 700$i/redis.conf --daemonize yes
+done
+echo yes | redis-cli --cluster create 127.0.0.1:7000 127.0.0.1:7001 \
+127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 \
+--cluster-replicas 1

infra/scripts/test-end-to-end-redis-cluster.sh

@@ -0,0 +1,248 @@
+#!/usr/bin/env bash
+
+set -e
+set -o pipefail
+
+test -z ${GOOGLE_APPLICATION_CREDENTIALS} && GOOGLE_APPLICATION_CREDENTIALS="/etc/service-account/service-account.json"
+test -z ${SKIP_BUILD_JARS} && SKIP_BUILD_JARS="false"
+test -z ${GOOGLE_CLOUD_PROJECT} && GOOGLE_CLOUD_PROJECT="kf-feast"
+test -z ${TEMP_BUCKET} && TEMP_BUCKET="feast-templocation-kf-feast"
+test -z ${JOBS_STAGING_LOCATION} && JOBS_STAGING_LOCATION="gs://${TEMP_BUCKET}/staging-location"
+test -z ${JAR_VERSION_SUFFIX} && JAR_VERSION_SUFFIX="-SNAPSHOT"
+
+echo "
+This script will run end-to-end tests for Feast Core and Online Serving.
+
+1. Install Redis as the store for Feast Online Serving.
+2. Install Postgres for persisting Feast metadata.
+3. Install Kafka and Zookeeper as the Source in Feast.
+4. Install Python 3.7.4, Feast Python SDK and run end-to-end tests from
+   tests/e2e via pytest.
+"
+
+apt-get -qq update
+apt-get -y install wget netcat kafkacat
+
+echo "
+============================================================
+Installing Redis at localhost:6379
+============================================================
+"
+# Allow starting serving in this Maven Docker image. Default set to not allowed.
+echo "exit 0" > /usr/sbin/policy-rc.d
+infra/scripts/setup-redis-cluster.sh
+redis-cli -c -p 7000 ping
+
+echo "
+============================================================
+Installing Postgres at localhost:5432
+============================================================
+"
+apt-get -y install postgresql > /var/log/postgresql.install.log
+service postgresql start
+# Initialize with database: 'postgres', user: 'postgres', password: 'password'
+cat <<EOF > /tmp/update-postgres-role.sh
+psql -c "ALTER USER postgres PASSWORD 'password';"
+EOF
+chmod +x /tmp/update-postgres-role.sh
+su -s /bin/bash -c /tmp/update-postgres-role.sh postgres
+export PGPASSWORD=password
+pg_isready
+
+echo "
+============================================================
+Installing Zookeeper at localhost:2181
+Installing Kafka at localhost:9092
+============================================================
+"
+wget -qO- https://www-eu.apache.org/dist/kafka/2.3.0/kafka_2.12-2.3.0.tgz | tar xz
+mv kafka_2.12-2.3.0/ /tmp/kafka
+nohup /tmp/kafka/bin/zookeeper-server-start.sh /tmp/kafka/config/zookeeper.properties &> /var/log/zookeeper.log 2>&1 &
+sleep 5
+tail -n10 /var/log/zookeeper.log
+nohup /tmp/kafka/bin/kafka-server-start.sh /tmp/kafka/config/server.properties &> /var/log/kafka.log 2>&1 &
+sleep 20
+tail -n10 /var/log/kafka.log
+kafkacat -b localhost:9092 -L
+
+if [[ ${SKIP_BUILD_JARS} != "true" ]]; then
+echo "
+============================================================
+Building jars for Feast
+============================================================
+"
+
+.prow/scripts/download-maven-cache.sh \
+    --archive-uri gs://feast-templocation-kf-feast/.m2.2019-10-24.tar \
+    --output-dir /root/
+
+# Build jars for Feast
+mvn --quiet --batch-mode --define skipTests=true clean package
+
+ls -lh core/target/*jar
+ls -lh serving/target/*jar
+else
+  echo "[DEBUG] Skipping building jars"
+fi
+
+echo "
+============================================================
+Starting Feast Core
+============================================================
+"
+# Start Feast Core in background
+cat <<EOF > /tmp/core.application.yml
+grpc:
+  port: 6565
+  enable-reflection: true
+
+feast:
+  version: 0.3
+  jobs:
+    runner: DirectRunner
+    options: {}
+    updates:
+      timeoutSeconds: 240
+    metrics:
+      enabled: false
+
+  stream:
+    type: kafka
+    options:
+      topic: feast-features
+      bootstrapServers: localhost:9092
+      replicationFactor: 1
+      partitions: 1
+
+spring:
+  jpa:
+    properties.hibernate:
+      format_sql: true
+      event.merge.entity_copy_observer: allow
+    hibernate.naming.physical-strategy=org.hibernate.boot.model.naming: PhysicalNamingStrategyStandardImpl
+    hibernate.ddl-auto: update
+  datasource:
+    url: jdbc:postgresql://localhost:5432/postgres
+    username: postgres
+    password: password
+
+management:
+  metrics:
+    export:
+      simple:
+        enabled: false
+      statsd:
+        enabled: false
+EOF
+
+nohup java -jar core/target/feast-core-*${JAR_VERSION_SUFFIX}.jar \
+  --spring.config.location=file:///tmp/core.application.yml \
+  &> /var/log/feast-core.log &
+sleep 35
+tail -n10 /var/log/feast-core.log
+nc -w2 localhost 6565 < /dev/null
+
+echo "
+============================================================
+Starting Feast Online Serving
+============================================================
+"
+# Start Feast Online Serving in background
+cat <<EOF > /tmp/serving.store.redis.cluster.yml
+name: serving
+type: REDIS_CLUSTER
+redis_cluster_config:
+  nodes:
+  - host: localhost
+    port: 7000
+  - host: localhost
+    port: 7001
+  - host: localhost
+    port: 7002
+  - host: localhost
+    port: 7003
+  - host: localhost
+    port: 7004
+  - host: localhost
+    port: 7005
+subscriptions:
+  - name: "*"
+    version: "*"
+    project: "*"
+EOF
+
+cat <<EOF > /tmp/serving.online.application.yml
+feast:
+  version: 0.3
+  core-host: localhost
+  core-grpc-port: 6565
+  tracing:
+    enabled: false
+  store:
+    config-path: /tmp/serving.store.redis.cluster.yml
+    redis-pool-max-size: 128
+    redis-pool-max-idle: 16
+  jobs:
+    staging-location: ${JOBS_STAGING_LOCATION}
+    store-type:
+    store-options: {}
+
+grpc:
+  port: 6566
+  enable-reflection: true
+
+spring:
+  main:
+    web-environment: false
+
+EOF
+
+nohup java -jar serving/target/feast-serving-*${JAR_VERSION_SUFFIX}.jar \
+  --spring.config.location=file:///tmp/serving.online.application.yml \
+  &> /var/log/feast-serving-online.log &
+sleep 15
+tail -n100 /var/log/feast-serving-online.log
+nc -w2 localhost 6566 < /dev/null
+
+echo "
+============================================================
+Installing Python 3.7 with Miniconda and Feast SDK
+============================================================
+"
+# Install Python 3.7 with Miniconda
+wget -q https://repo.continuum.io/miniconda/Miniconda3-4.7.12-Linux-x86_64.sh \
+   -O /tmp/miniconda.sh
+bash /tmp/miniconda.sh -b -p /root/miniconda -f
+/root/miniconda/bin/conda init
+source ~/.bashrc
+
+# Install Feast Python SDK and test requirements
+pip install -qe sdk/python
+pip install -qr tests/e2e/requirements.txt
+
+echo "
+============================================================
+Running end-to-end tests with pytest at 'tests/e2e'
+============================================================
+"
+# Default artifact location setting in Prow jobs
+LOGS_ARTIFACT_PATH=/logs/artifacts
+
+ORIGINAL_DIR=$(pwd)
+cd tests/e2e
+
+set +e
+pytest basic-ingest-redis-serving.py --junitxml=${LOGS_ARTIFACT_PATH}/python-sdk-test-report.xml
+TEST_EXIT_CODE=$?
+
+if [[ ${TEST_EXIT_CODE} != 0 ]]; then
+  echo "[DEBUG] Printing logs"
+  ls -ltrh /var/log/feast*
+  cat /var/log/feast-serving-online.log /var/log/feast-core.log
+
+  echo "[DEBUG] Printing Python packages list"
+  pip list
+fi
+
+cd ${ORIGINAL_DIR}
+exit ${TEST_EXIT_CODE}

ingestion/pom.xml

@@ -113,6 +113,12 @@
       <version>${project.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>dev.feast</groupId>
+      <artifactId>feast-storage-connector-redis-cluster</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>dev.feast</groupId>
       <artifactId>feast-storage-connector-bigquery</artifactId>

ingestion/src/main/java/feast/ingestion/utils/StoreUtil.java

@@ -25,6 +25,7 @@
 import feast.storage.api.writer.FeatureSink;
 import feast.storage.connectors.bigquery.writer.BigQueryFeatureSink;
 import feast.storage.connectors.redis.writer.RedisFeatureSink;
+import feast.storage.connectors.rediscluster.writer.RedisClusterFeatureSink;
 import feast.types.ValueProto.ValueType.Enum;
 import java.util.HashMap;
 import java.util.Map;
@@ -82,12 +83,14 @@ public static FeatureSink getFeatureSink(
       Store store, Map<String, FeatureSetSpec> featureSetSpecs) {
     StoreType storeType = store.getType();
     switch (storeType) {
+      case REDIS_CLUSTER:
+        return RedisClusterFeatureSink.fromConfig(store.getRedisClusterConfig(), featureSetSpecs);
       case REDIS:
         return RedisFeatureSink.fromConfig(store.getRedisConfig(), featureSetSpecs);
       case BIGQUERY:
         return BigQueryFeatureSink.fromConfig(store.getBigqueryConfig(), featureSetSpecs);
       default:
-        throw new RuntimeException(String.format("Store type '{}' is unsupported", storeType));
+        throw new RuntimeException(String.format("Store type '%s' is unsupported", storeType));
     }
   }
 }

protos/feast/core/Store.proto

@@ -105,6 +105,8 @@ message Store {
 
     // Unsupported in Feast 0.3
     CASSANDRA = 3;
+
+    REDIS_CLUSTER = 4;
   }
 
   message RedisConfig {
@@ -130,6 +132,13 @@ message Store {
     int32 port = 2;
   }
 
+  message RedisClusterConfig {
+    // List of Redis Uri for all the nodes in Redis Cluster, comma separated. Eg. host1:6379, host2:6379
+    string connection_string = 1;
+    int32 initial_backoff_ms = 2;
+    int32 max_retries = 3;
+  }
+
   message Subscription {
     // Name of project that the feature sets belongs to. This can be one of
     // - [project_name]
@@ -172,5 +181,6 @@ message Store {
     RedisConfig redis_config = 11;
     BigQueryConfig bigquery_config = 12;
     CassandraConfig cassandra_config = 13;
+    RedisClusterConfig redis_cluster_config = 14;
   }
 }

serving/pom.xml

@@ -97,6 +97,12 @@
       <version>${project.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>dev.feast</groupId>
+      <artifactId>feast-storage-connector-redis-cluster</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
     <dependency>
       <groupId>dev.feast</groupId>
       <artifactId>feast-storage-connector-bigquery</artifactId>