feast-dev · feast-ci-bot · Mar 21, 2019 · Mar 6, 2019 · Mar 6, 2019 · Mar 6, 2019
diff --git a/charts/README.md b/charts/README.md
@@ -34,8 +34,8 @@ The components included in this chart are:
 
 Components that Feast supports, but this installation will not include are:
 
-- Storage
-  - `Redis` or `Postgres` dbs for feature storage
+- Serving Storage
+  - `Redis` or `Bigtable` dbs for feature serving storage
 - [TICK stack](https://www.influxdata.com/time-series-platform/) for metrics monitoring
   - Set `statsd.host` and `statsd.port` to direct job metrics to your metrics store.
   - Note that if you do not provision a metrics store, feast will only retain the latest metrics from your jobs.
@@ -62,13 +62,12 @@ The following table lists the configurable parameters of the Feast chart and the
 | `core.image.registry` | core docker image registry | feast | 
 | `core.image.repository` | core docker image repository | feast-core | 
 | `core.image.tag` | core docker image version | 0.1.0 | 
-| `core.jobs.errorStoreType` | type of errors store to write errors to. One of `stdout`, `stderr`, `file.json` | stdout | 
-| `core.jobs.errorStoreOptions` | additional options for the error store in json string format | `{}` | 
-| `core.jobs.monitoring.initialDelay` | delay before a job starts to be monitored in ms | 60000 | 
+| `core.jobs.monitoring.initialDelay` | delay before a job starts to be monitored in ms | 60000 |
 | `core.jobs.monitoring.period` | polling interval for jobs monitoring in ms | 5000 | 
 | `core.jobs.options` | additional options to be provided to the beam job. Should be a char escaped json k-v object | {} | 
 | `core.jobs.runner` | beam job runner - one of `DirectRunner`, `FlinkRunner` or `DataflowRunner` | DirectRunner | 
-| `core.replicaCount` | core deployment replica count | 3 | 
+| `core.jobs.workspace` | workspace path for ingestion jobs, used for separate job workspaces to share importJobSpecs.yaml with ingestion and for writing errors to if no default errors store is configured | nil |
+| `core.replicaCount` | core deployment replica count | 3 |
 | `core.resources.limits.cpu` | core cpu limits | 1 |
 | `core.resources.limits.memory` | core memory limits | 2G |
 | `core.resources.requests.cpu` | core cpu requested | 1 |
@@ -103,10 +102,16 @@ The following table lists the configurable parameters of the Feast chart and the
 | `serving.service.extIPAdr` | Internal load balancer IP Address for serving, required so jobs on external runners can connect to the service | nil | 
 | `serving.service.loadBalancerSourceRanges` | IP source ranges that will have access to serving. If not set, will default to 0.0.0.0/0 | nil |
 | `serviceAccount.name` | service account secret name to mount to deployments | nil | 
-| `serviceAccount.key` | service account secret key to mount to deployments | nil | 
+| `serviceAccount.key` | service account secret key to mount to deployments | nil |
 | `statsd.host` | host of statsd daemon for job metrics to be sent to | nil | 
-| `statsd.port` | port of statsd daemon for job metrics to be sent to | nil | 
-| `postgresql.provision` | Provision PostgreSQL | true | 
+| `statsd.port` | port of statsd daemon for job metrics to be sent to | nil |
+| `store.errors.type` | type of default errors store to write errors to. One of `stdout`, `stderr`, `file.json` | nil |
+| `store.errors.options` | additional options for the default error store in json string format | `{}` |
+| `store.serving.type` | type of default serving store to write errors to. One of `redis`, `bigtable` | nil |
+| `store.serving.options` | additional options for the default serving store in json string format | `{}` |
+| `store.warehouse.type` | type of default warehouse store to write errors to. One of `bigquery`, `file.json` | nil |
+| `store.warehouse.options` | additional options for the default warehouse store in json string format | `{}` |
+| `postgresql.provision` | Provision PostgreSQL | true |
 | `postgresql.postgresPassword` | specify password if you want the postgres password secret to be generated | nil | 
 | `postgresql.resources.requests.cpu` | postgres requested cpu | 100m | 
 | `postgresql.resources.requests.memory` | postgres requested memory | 256Mi | 

diff --git a/charts/dist/feast-0.1.0.tgz b/charts/dist/feast-0.1.0.tgz
diff --git a/charts/dist/index.yaml b/charts/dist/index.yaml
@@ -3,11 +3,11 @@ entries:
   feast:
   - apiVersion: v1
     appVersion: 0.1.0
-    created: 2019-02-08T15:00:47.594234304+08:00
+    created: 2019-03-15T19:35:11.848995-07:00
     description: A Helm chart to install Feast on kubernetes
-    digest: 099023ae667dd8129fbf72adf14732cf1f32af60784d694634ef4e1983957a5a
+    digest: cbc97b4be6a84a33055900b620d2f6b6176ee81d86eb2e8bfcc9096ea040ff47
     name: feast
     urls:
     - feast-0.1.0.tgz
     version: 0.1.0
-generated: 2019-02-08T15:00:47.592388768+08:00
+generated: 2019-03-15T19:35:11.844526-07:00
diff --git a/charts/feast/templates/core-deploy.yaml b/charts/feast/templates/core-deploy.yaml
@@ -81,20 +81,34 @@ spec:
             value: "{{ .Values.core.projectId}}"
           - name: TRAINING_DATASET_PREFIX
             value: "{{ .Values.core.trainingDatasetPrefix }}"
-          - name: CORE_API_URI
-            value: {{ .Values.core.service.extIPAdr }}:{{ .Values.core.service.grpc.targetPort }}
           - name: JOB_RUNNER
             value: {{ .Values.core.jobs.runner }}
+          - name: JOB_WORKSPACE
+            value: {{ .Values.core.jobs.workspace }}
           - name: JOB_OPTIONS
             value: {{ .Values.core.jobs.options | toJson}}
           - name: JOB_MONITOR_PERIOD_MS
             value: "{{ .Values.core.jobs.monitoring.period }}"
           - name: JOB_MONITOR_INITIAL_DELAY_MS
             value: "{{ .Values.core.jobs.monitoring.initialDelay }}"
-          - name: JOB_ERRORS_STORE_TYPE
-            value: {{ .Values.core.jobs.errorStoreType }}
-          - name: JOB_ERRORS_STORE_OPTIONS
-            value: {{ .Values.core.jobs.errorStoreOptions | toJson}}
+          {{- if .Values.store.serving }}
+          - name: STORE_SERVING_TYPE
+            value: {{ .Values.store.serving.type }}
+          - name: STORE_SERVING_OPTIONS
+            value: {{ .Values.store.serving.options | toJson}}
+          {{- end }}
+          {{- if .Values.store.warehouse }}
+          - name: STORE_WAREHOUSE_TYPE
+            value: {{ .Values.store.warehouse.type }}
+          - name: STORE_WAREHOUSE_OPTIONS
+            value: {{ .Values.store.warehouse.options | toJson}}
+          {{- end }}
+          {{- if .Values.store.errors }}
+          - name: STORE_ERRORS_TYPE
+            value: {{ .Values.store.errors.type }}
+          - name: STORE_ERRORS_OPTIONS
+            value: {{ .Values.store.errors.options | toJson}}
+          {{- end }}
           - name: STATSD_HOST
             value: {{ .Values.statsd.host }}
           - name: STATSD_PORT

diff --git a/charts/feast/values.yaml b/charts/feast/values.yaml
@@ -24,7 +24,8 @@ core:
     http: 
       port: 80
       targetPort: 8080
-  jobs: 
+  jobs:
+    workspace: "/tmp"
     runner: DirectRunner
     options: "{}"
     errorStoreType: "stdout"
@@ -35,6 +36,16 @@ core:
   trainingDatasetPrefix: "fs"
   # logType: JSON
 
+#store:
+#  errors:
+#    type: "stdout"
+#  warehouse:
+#    type: "bigquery"
+#    options: '{"project": "gcp-project-id", "dataset": "feast"}'
+#  serving:
+#    type: "redis"
+#    options: '{"host": "...", "port": "6379"}'
+
 postgresql: 
   provision: true
   imageTag: 9.6.11

diff --git a/core/pom.xml b/core/pom.xml
@@ -221,6 +221,11 @@
             <artifactId>google-cloud-bigquery</artifactId>
             <version>1.48.0</version>
         </dependency>
+        <dependency>
+            <groupId>com.google.cloud</groupId>
+            <artifactId>google-cloud-nio</artifactId>
+            <version>0.83.0-alpha</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-clients_2.11</artifactId>
@@ -243,7 +248,11 @@
             <artifactId>jackson-annotations</artifactId>
             <version>2.9.8</version>
         </dependency>
-
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <version>2.9.8</version>
+        </dependency>
 
         <!--compile 'com.github.spullara.mustache.java:compiler:0.9.5'-->
         <dependency>
@@ -330,5 +339,31 @@
             <version>3.11.0</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-test</artifactId>
+            <version>5.0.8.RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-test</artifactId>
+            <version>${springBootVersion}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-test-autoconfigure</artifactId>
+            <version>${springBootVersion}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.h2database</groupId>
+            <artifactId>h2</artifactId>
+            <version>1.4.198</version>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 </project>
diff --git a/core/src/main/java/feast/core/OnContextRefresh.java b/core/src/main/java/feast/core/OnContextRefresh.java
@@ -0,0 +1,29 @@
+package feast.core;
+
+import feast.core.config.StorageConfig.StorageSpecs;
+import feast.core.service.SpecService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class OnContextRefresh {
+
+  @Autowired
+  private SpecService specService;
+  @Autowired
+  private StorageSpecs storageSpecs;
+
+  @EventListener
+  public void onApplicationEvent(ContextRefreshedEvent event) {
+    if (storageSpecs.getServingStorageSpec() != null) {
+      specService.registerStorage(storageSpecs.getServingStorageSpec());
+    }
+    if (storageSpecs.getWarehouseStorageSpec() != null) {
+      specService.registerStorage(storageSpecs.getWarehouseStorageSpec());
+    }
+  }
+}
diff --git a/core/src/main/java/feast/core/config/AppConfig.java b/core/src/main/java/feast/core/config/AppConfig.java
@@ -28,13 +28,15 @@
 public class AppConfig {
   @Bean
   public ImportJobDefaults getImportJobDefaults(
-      @Value("${feast.jobs.coreUri}") String coreApiUri,
       @Value("${feast.jobs.runner}") String runner,
       @Value("${feast.jobs.options}") String options,
       @Value("${feast.jobs.executable}") String executable,
-      @Value("${feast.jobs.errorsStoreType}") String errorsStoreType,
-      @Value("${feast.jobs.errorsStoreOptions}") String errorsStoreOptions) {
-    return new ImportJobDefaults(
-        coreApiUri, runner, options, executable, errorsStoreType, errorsStoreOptions);
+      @Value("${feast.jobs.workspace}") String workspace) {
+    return ImportJobDefaults.builder()
+        .importJobOptions(options)
+        .runner(runner)
+        .executable(executable)
+        .workspace(workspace)
+        .build();
   }
 }
diff --git a/core/src/main/java/feast/core/config/ImportJobDefaults.java b/core/src/main/java/feast/core/config/ImportJobDefaults.java
@@ -17,7 +17,7 @@
 
 package feast.core.config;
 
-import lombok.AllArgsConstructor;
+import lombok.Builder;
 import lombok.Getter;
 import lombok.Setter;
 
@@ -26,13 +26,12 @@
  */
 @Getter
 @Setter
-@AllArgsConstructor
+@Builder
 public class ImportJobDefaults {
-  private String coreApiUri;
-  private String runner;
+
   private String importJobOptions;
+  private String runner;
+  private String workspace;
   private String executable;
-  private String errorsStoreType;
-  private String errorsStoreOptions;
 }
 
diff --git a/core/src/main/java/feast/core/config/StorageConfig.java b/core/src/main/java/feast/core/config/StorageConfig.java
@@ -0,0 +1,83 @@
+package feast.core.config;
+
+import static feast.core.util.TypeConversion.convertJsonStringToMap;
+
+import com.google.common.base.Strings;
+import feast.core.validators.SpecValidator;
+import feast.specs.StorageSpecProto.StorageSpec;
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@AllArgsConstructor
+@NoArgsConstructor
+public class StorageConfig {
+
+  public static final String DEFAULT_SERVING_ID = "SERVING";
+  public static final String DEFAULT_WAREHOUSE_ID = "WAREHOUSE";
+  public static final String DEFAULT_ERRORS_ID = "ERRORS";
+
+  @Autowired
+  private SpecValidator validator;
+
+  private StorageSpec buildStorageSpec(
+      String id,
+      String type,
+      String options) {
+    options = Strings.isNullOrEmpty(options) ? "{}" : options;
+    Map<String, String> optionsMap = convertJsonStringToMap(options);
+    if (Strings.isNullOrEmpty(type)) {
+      return null;
+    }
+    StorageSpec storageSpec = StorageSpec.newBuilder()
+        .setId(id)
+        .setType(type)
+        .putAllOptions(optionsMap)
+        .build();
+    switch (id) {
+      case DEFAULT_SERVING_ID:
+        validator.validateServingStorageSpec(storageSpec);
+        break;
+      case DEFAULT_WAREHOUSE_ID:
+        validator.validateWarehouseStorageSpec(storageSpec);
+        break;
+      case DEFAULT_ERRORS_ID:
+        validator.validateErrorsStorageSpec(storageSpec);
+        break;
+    }
+    return storageSpec;
+  }
+
+  @Bean
+  public StorageSpecs getStorageSpecs(
+      @Value("${feast.store.serving.type}") String servingType,
+      @Value("${feast.store.serving.options}") String servingOptions,
+      @Value("${feast.store.warehouse.type}") String warehouseType,
+      @Value("${feast.store.warehouse.options}") String warehouseOptions,
+      @Value("${feast.store.errors.type}") String errorsType,
+      @Value("${feast.store.errors.options}") String errorsOptions) {
+    StorageSpecs storageSpecs = StorageSpecs.builder()
+        .servingStorageSpec(buildStorageSpec(DEFAULT_SERVING_ID, servingType, servingOptions))
+        .warehouseStorageSpec(
+            buildStorageSpec(DEFAULT_WAREHOUSE_ID, warehouseType, warehouseOptions))
+        .errorsStorageSpec(buildStorageSpec(DEFAULT_ERRORS_ID, errorsType, errorsOptions))
+        .build();
+    return storageSpecs;
+  }
+
+  @Builder
+  @Getter
+  public static class StorageSpecs {
+
+    private StorageSpec servingStorageSpec;
+    private StorageSpec warehouseStorageSpec;
+    private StorageSpec errorsStorageSpec;
+  }
+}