docker compose & health impl

infra/docker-compose/.env.sample

@@ -2,6 +2,7 @@ COMPOSE_PROJECT_NAME=feast
 FEAST_VERSION=0.6.2
 GCP_SERVICE_ACCOUNT=./gcp-service-accounts/key.json
 FEAST_CORE_CONFIG=./core/core.yml
+FEAST_JC_CONFIG=./jc/jc.yml
 FEAST_HISTORICAL_SERVING_CONFIG=./serving/historical-serving.yml
 FEAST_HISTORICAL_SERVING_ENABLED=false
 FEAST_ONLINE_SERVING_CONFIG=./serving/online-serving.yml

infra/docker-compose/core/core.yml

@@ -1,13 +1,4 @@
 feast:
-  jobs:
-    polling_interval_milliseconds: 20000
-    job_update_timeout_seconds: 240
-    active_runner: direct
-    runners:
-      - name: direct
-        type: DirectRunner
-        options:
-          tempLocation: gs://bucket/tempLocation
   stream:
     type: kafka
     options:

infra/docker-compose/docker-compose.yml

@@ -21,6 +21,23 @@ services:
       - /opt/feast/feast-core.jar
       - --spring.config.location=classpath:/application.yml,file:/etc/feast/application.yml
 
+  jc:
+    image: gcr.io/kf-feast/feast-jc:${FEAST_VERSION}
+    volumes:
+      - ${FEAST_JC_CONFIG}:/etc/feast/application.yml
+      - ${GCP_SERVICE_ACCOUNT}:/etc/gcloud/service-accounts/key.json
+    environment:
+      GOOGLE_APPLICATION_CREDENTIALS: /etc/gcloud/service-accounts/key.json
+    depends_on:
+      - kafka
+    ports:
+      - 6570:6570
+    command:
+      - java
+      - -jar
+      - /opt/feast/feast-job-coordinator.jar
+      - --spring.config.location=classpath:/application.yml,file:/etc/feast/application.yml
+
   jupyter:
     image: gcr.io/kf-feast/feast-jupyter:${FEAST_VERSION}
     volumes:

infra/docker-compose/jc/jc.yml

@@ -0,0 +1,15 @@
+feast:
+  jobs:
+    polling_interval_milliseconds: 20000
+    job_update_timeout_seconds: 240
+    active_runner: direct
+    runners:
+      - name: direct
+        type: DirectRunner
+        options:
+          tempLocation: gs://bucket/tempLocation
+  stream:
+    type: kafka
+    options:
+      topic: feast-features
+      bootstrapServers: "kafka:9092,localhost:9094"

infra/scripts/test-load.sh

@@ -36,6 +36,7 @@ else
 fi
 
 wait_for_docker_image gcr.io/kf-feast/feast-core:"${FEAST_VERSION}"
+wait_for_docker_image gcr.io/kf-feast/feast-jc:"${FEAST_VERSION}"
 wait_for_docker_image gcr.io/kf-feast/feast-serving:"${FEAST_VERSION}"
 wait_for_docker_image gcr.io/kf-feast/feast-jupyter:"${FEAST_VERSION}"
 
@@ -65,6 +66,12 @@ export FEAST_CORE_CONTAINER_IP_ADDRESS=$(docker inspect -f '{{range .NetworkSett
 # Wait for Feast Core to be ready
 "${PROJECT_ROOT_DIR}"/infra/scripts/wait-for-it.sh ${FEAST_CORE_CONTAINER_IP_ADDRESS}:6565 --timeout=120
 
+# Get Feast Job Coordinator container IP address
+export FEAST_JC_CONTAINER_IP_ADDRESS=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' feast_jc_1)
+
+# Wait for Feast Job Coordinator to be ready
+"${PROJECT_ROOT_DIR}"/infra/scripts/wait-for-it.sh ${FEAST_JC_CONTAINER_IP_ADDRESS}:6570 --timeout=120
+
 # Get Feast Online Serving container IP address
 export FEAST_ONLINE_SERVING_CONTAINER_IP_ADDRESS=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' feast_online_serving_1)
 

job-coordinator/src/main/java/feast/jc/dao/InMemoryJobRepository.java

@@ -49,11 +49,9 @@ public class InMemoryJobRepository implements JobRepository {
   @Autowired
   public InMemoryJobRepository(JobManager jobManager) {
     this.jobManager = jobManager;
-    this.storage = new HashMap<>();
 
-    //    this.storage =
-    //        this.jobManager.listRunningJobs().stream().collect(Collectors.toMap(Job::getId, j ->
-    // j));
+    this.storage =
+        this.jobManager.listRunningJobs().stream().collect(Collectors.toMap(Job::getId, j -> j));
   }
 
   /**

job-coordinator/src/main/java/feast/jc/grpc/HealthServiceImpl.java

@@ -0,0 +1,54 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package feast.jc.grpc;
+
+import feast.jc.dao.JobRepository;
+import io.grpc.Status;
+import io.grpc.health.v1.HealthGrpc.HealthImplBase;
+import io.grpc.health.v1.HealthProto.HealthCheckRequest;
+import io.grpc.health.v1.HealthProto.HealthCheckResponse;
+import io.grpc.health.v1.HealthProto.HealthCheckResponse.ServingStatus;
+import io.grpc.stub.StreamObserver;
+import lombok.extern.slf4j.Slf4j;
+import net.devh.boot.grpc.server.service.GrpcService;
+import org.springframework.beans.factory.annotation.Autowired;
+
+@Slf4j
+@GrpcService
+public class HealthServiceImpl extends HealthImplBase {
+  private final JobRepository jobRepository;
+
+  @Autowired
+  public HealthServiceImpl(JobRepository jobRepository) {
+    this.jobRepository = jobRepository;
+  }
+
+  @Override
+  public void check(
+      HealthCheckRequest request, StreamObserver<HealthCheckResponse> responseObserver) {
+    try {
+      jobRepository.findAll();
+      responseObserver.onNext(
+          HealthCheckResponse.newBuilder().setStatus(ServingStatus.SERVING).build());
+      responseObserver.onCompleted();
+    } catch (Exception e) {
+      log.error("Health Check: unable to retrieve projects.\nError: %s", e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    }
+  }
+}