Fix tracing to continue from existing trace created by grpc client (#245

)

* Fix tracing to continue from existing trace created by grpc client

* Update python SDK with updated protos in Feast 0.1.x
Removal of storage specs

* Add __init__ file in feast/types to fix module not found error

* Add support to configure import job metrics in Feast Core

* Fix CoreApplicationWithNoWarehouseTest

* Update charts
Add import job metrics configuration
Update jaeger tracing configuration for serving

* Add lazy annotation in getBigQueryTrainingDatasetTemplater in TrainingConfig to resolve circular dependency

* Use provided values for INGESTION_METRICS_ENABLED

* Add Python mypy definition

* Increase version of google-cloud-storage in python sdk

* Update docs for influx_db_url field in ImportJobSpecs proto

* Write feature metrics in fixed window for better performance

* Update default image in Helm chart

* Write feature metrics only if the job type is streaming

* Set transform name for writing feature metrics to Influx DB

* Update chart, making resources limit optional
Following how stable postgresql chart specifies it
https://github.com/helm/charts/blob/master/stable/postgresql/values.yaml

* Fix the check for unbounded sources
Check against the ImportSpec.Type value (kafka or pubsub) instead of PipelineOptions.Streaming (the streaming status cannot be reliably determined at pipeline creation stage)

* Fix typo

* Update default Feast image tag used in the chart

charts/feast/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1
-appVersion: "0.1.4"
+appVersion: "0.1.7"
 description: A Helm chart to install Feast on kubernetes
 name: feast
-version: 0.1.2
+version: 0.1.4

charts/feast/README.md

@@ -46,7 +46,6 @@ Components that Feast supports, but this installation will not include are:
   - Note that if you do not provision a metrics store, feast will only retain the latest metrics from your jobs.
 - [Jaeger tracing](www.jaegertracing.io) for serving performance. 
   - Set `serving.jaeger.enabled` to `true`, and configure the following parameters:
-    - `serving.jaeger.host`
     - `serving.jaeger.port`
     - `serving.jaeger.options.samplerType`
     - `serving.jaeger.options.samplerParam`
@@ -72,6 +71,10 @@ The following table lists the configurable parameters of the Feast chart and the
 | `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.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.jobs.writeFeatureMetricsToInfluxDb` | specifies whether Feast import job will write feature metrics (such as feature lag and values summaries) to Influx DB for monitoring and alert purpose | false |
+| `core.jobs.influxDbUrl` | Influx DB url e.g. http://localhost:8086 (required if `core.jobs.writeFeatureMetricsToInfluxDb = true`) |  |
+| `core.jobs.influxDbDatabase` | Influx DB database name (required if `core.jobs.writeFeatureMetricsToInfluxDb = true`) | |
+| `core.jobs.influxDbMeasurement` | Influx DB [measurement name](https://docs.influxdata.com/influxdb/v1.7/concepts/key_concepts/#measurement) (required if `core.jobs.writeFeatureMetricsToInfluxDb = true`) | |
 | `core.replicaCount` | core deployment replica count | 3 |
 | `core.resources.limits.cpu` | core cpu limits | 1 |
 | `core.resources.limits.memory` | core memory limits | 2G |

charts/feast/templates/core-deploy.yaml

@@ -58,12 +58,7 @@ spec:
           timeoutSeconds: 3
           failureThreshold: {{ .Values.core.readinessProbe.failureThreshold }}
         resources:
-          requests:
-            cpu: {{ .Values.core.resources.requests.cpu }}
-            memory: {{ .Values.core.resources.requests.memory }}
-          limits:
-            cpu: {{ .Values.core.resources.limits.cpu }}
-            memory: {{ .Values.core.resources.limits.memory }}
+{{ toYaml .Values.core.resources | indent 10 }}
         {{- if .Values.serviceAccount }}
         volumeMounts:
           - name: "{{ .Values.serviceAccount.name }}"
@@ -98,6 +93,18 @@ spec:
             value: "{{ .Values.core.jobs.monitoring.period }}"
           - name: JOB_MONITOR_INITIAL_DELAY_MS
             value: "{{ .Values.core.jobs.monitoring.initialDelay }}"
+
+          {{- if .Values.core.jobs.writeFeatureMetricsToInfluxDb }}
+          - name: INGESTION_METRICS_ENABLED
+            value: "{{ .Values.core.jobs.writeFeatureMetricsToInfluxDb }}"
+          - name: INGESTION_METRICS_INFLUX_URL
+            value: {{ .Values.core.jobs.influxDbUrl }}
+          - name: INGESTION_METRICS_INFLUX_DB_NAME
+            value: {{ .Values.core.jobs.influxDbDatabase }}
+          - name: INGESTION_METRICS_INFLUX_DB_MEASUREMENT
+            value: {{ .Values.core.jobs.influxDbMeasurement }}
+          {{- end }}
+
           {{- if .Values.store }}
           {{- if .Values.store.serving }}
           - name: STORE_SERVING_TYPE

charts/feast/templates/serving-deploy.yaml

@@ -59,12 +59,7 @@ spec:
           timeoutSeconds: 3
           failureThreshold: {{ .Values.serving.readinessProbe.failureThreshold }}
         resources:
-          requests:
-            cpu: "{{ .Values.serving.resources.requests.cpu }}"
-            memory: "{{ .Values.serving.resources.requests.memory }}"
-          limits:
-            cpu: "{{ .Values.serving.resources.limits.cpu }}"
-            memory: "{{ .Values.serving.resources.limits.memory }}"
+{{ toYaml .Values.serving.resources | indent 10 }}
         env:
           - name: FEAST_SERVING_HTTP_PORT
             value: "{{ .Values.serving.service.http.targetPort }}"
@@ -96,7 +91,9 @@ spec:
           - name: JAEGER_ENABLED
             value: "{{ .Values.serving.jaeger.enabled }}"
           - name: JAEGER_AGENT_HOST
-            value: "{{ .Values.serving.jaeger.host }}"
+            valueFrom:
+              fieldRef:
+                fieldPath: status.hostIP
           - name: JAEGER_AGENT_PORT
             value: "{{ .Values.serving.jaeger.port }}"
           - name: JAEGER_SAMPLER_TYPE

charts/feast/values.yaml

@@ -1,16 +1,26 @@
---- 
+# [required]
+# global.postgresql.secretName is an existing Kubernetes secret name containing Postgresql password
+# The secret needs to have this key: postgresql-password
+#
+# Example of how to generate this secret:
+# kubectl create secret generic <feast_helm_release_name>-postgresql \
+#   --from-literal=postgresql-password=<postgresql_password>
+#
+# With the example above the secretName will be <feast_helm_release_name>-postgresql
+# 
+# global:
+#  postgresql:
+#    secretName: <existing_kubernetes_secret_name_containing_postgresql_credentials>
+
 core:
   projectId: "gcp-project-id"
   image: 
     pullPolicy: IfNotPresent
     registry: gcr.io/kf-feast
     repository: feast-core
-    tag: "0.1.4"
+    tag: "0.1.7"
   replicaCount: 1
   resources: 
-    limits: 
-      cpu: 4
-      memory: 6G
     requests: 
       cpu: 1
       memory: 2G
@@ -38,6 +48,14 @@ core:
     monitoring:
       period: 5000
       initialDelay: 60000
+    # writeFeatureMetricsToInfluxDb specifies whether Feast import job will write feature metrics (such as feature lag and values summaries) to Influx DB for monitoring and alert purpose
+    writeFeatureMetricsToInfluxDb: false 
+    # influxDbUrl, influxDbDatabase and influxDbMeasurement sets the Influx DB configuration where Feast import job will write the feature metrics
+    # Uncomment the following 3 fields if writeFeatureMetricsToInfluxDb = true
+    #
+    # influxDbUrl: http://localhost:8086
+    # influxDbDatabase: influx_db_database
+    # influxDbMeasurement: influx_db_measurement
   trainingDatasetPrefix: "fs"
   # logType: JSON
   livenessProbe:
@@ -84,12 +102,9 @@ serving:
     pullPolicy: IfNotPresent
     registry: gcr.io/kf-feast
     repository: feast-serving
-    tag: "0.1.4"
+    tag: "0.1.7"
   replicaCount: 1
   resources: 
-    limits: 
-      cpu: 2
-      memory: 4G
     requests: 
       cpu: 1
       memory: 1G
@@ -109,6 +124,9 @@ serving:
     # loadBalancerSourceRanges: ["10.0.0.0/8"]
   jaeger:
     enabled: false
+    # options:
+    #   samplerType: constants
+    #   samplerParam: 1
   livenessProbe:
     initialDelaySeconds: 120
     failureThreshold: 3

core/src/main/java/feast/core/config/ImportJobMetricsConfig.java

@@ -0,0 +1,13 @@
+package feast.core.config;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public class ImportJobMetricsConfig {
+  private final boolean ingestionMetricsEnabled;
+  private final String influxDbUrl;
+  private final String influxDbName;
+  private final String influxDbMeasurementName;
+}

core/src/main/java/feast/core/config/InstrumentationConfig.java

@@ -26,8 +26,17 @@
 @Configuration
 public class InstrumentationConfig {
   @Bean
-  public StatsDClient getStatsDClient(@Value("${statsd.host}") String host,
-      @Value("${statsd.port}") int port) {
+  public StatsDClient getStatsDClient(
+      @Value("${statsd.host}") String host, @Value("${statsd.port}") int port) {
     return new NonBlockingStatsDClient("feast_core", host, port);
   }
+
+  @Bean
+  public ImportJobMetricsConfig getImportJobMetricsConfig(
+      @Value("${ingestion.metrics.enabled}") boolean enabled,
+      @Value("${ingestion.metrics.influxUrl}") String influxDbUrl,
+      @Value("${ingestion.metrics.dbName}") String influxDbName,
+      @Value("${ingestion.metrics.dbMeasurement}") String influxDbMeasurementName) {
+    return new ImportJobMetricsConfig(enabled, influxDbUrl, influxDbName, influxDbMeasurementName);
+  }
 }

core/src/main/java/feast/core/service/JobManagementService.java

@@ -30,6 +30,7 @@
 import com.google.protobuf.util.JsonFormat;
 import feast.core.JobServiceProto.JobServiceTypes.JobDetail;
 import feast.core.config.ImportJobDefaults;
+import feast.core.config.ImportJobMetricsConfig;
 import feast.core.config.StorageConfig.StorageSpecs;
 import feast.core.dao.JobInfoRepository;
 import feast.core.dao.MetricsRepository;
@@ -73,6 +74,7 @@ public class JobManagementService {
   private static final String JOB_PREFIX_DEFAULT = "feastimport";
   private static final String UNKNOWN_EXT_JOB_ID = "";
   private static final String IMPORT_JOB_SPECS_FILENAME = "importJobSpecs.yaml";
+  private ImportJobMetricsConfig metricsConfig;
 
   private JobInfoRepository jobInfoRepository;
   private MetricsRepository metricsRepository;
@@ -88,13 +90,15 @@ public JobManagementService(
       JobManager jobManager,
       ImportJobDefaults defaults,
       SpecService specService,
-      StorageSpecs storageSpecs) {
+      StorageSpecs storageSpecs,
+      ImportJobMetricsConfig metricsConfig) {
     this.jobInfoRepository = jobInfoRepository;
     this.metricsRepository = metricsRepository;
     this.jobManager = jobManager;
     this.defaults = defaults;
     this.specService = specService;
     this.storageSpecs = storageSpecs;
+    this.metricsConfig = metricsConfig;
   }
 
   public void writeImportJobSpecs(ImportJobSpecs importJobSpecs, Path workspace) {
@@ -132,7 +136,11 @@ private ImportJobSpecs buildImportJobSpecs(ImportSpec importSpec, String jobId)
         .setJobId(jobId)
         .setImportSpec(importSpec)
         .addAllEntitySpecs(entitySpecs)
-        .addAllFeatureSpecs(featureSpecs);
+        .addAllFeatureSpecs(featureSpecs)
+        .setWriteFeatureMetricsToInfluxDb(metricsConfig.isIngestionMetricsEnabled())
+        .setInfluxDbUrl(metricsConfig.getInfluxDbUrl())
+        .setInfluxDbDatabase(metricsConfig.getInfluxDbName())
+        .setInfluxDbMeasurement(metricsConfig.getInfluxDbMeasurementName());
     if (storageSpecs.getServingStorageSpec() != null) {
       importJobSpecsBuilder.setServingStorageSpec(storageSpecs.getServingStorageSpec());
     }

core/src/main/resources/application.properties

@@ -55,3 +55,8 @@ management.metrics.export.simple.enabled=false
 management.metrics.export.statsd.enabled=true
 management.metrics.export.statsd.host=${STATSD_HOST:localhost}
 management.metrics.export.statsd.port=${STATSD_PORT:8125}
+
+ingestion.metrics.enabled=${INGESTION_METRICS_ENABLED:false}
+ingestion.metrics.influxUrl=${INGESTION_METRICS_INFLUX_URL:}
+ingestion.metrics.dbName=${INGESTION_METRICS_INFLUX_DB_NAME:}
+ingestion.metrics.dbMeasurement=${INGESTION_METRICS_INFLUX_DB_MEASUREMENT:}

core/src/test/java/feast/core/CoreApplicationTest.java

@@ -52,7 +52,11 @@
     "feast.store.warehouse.options={\"path\":\"/tmp/foobar\"}",
     "feast.store.serving.type=redis",
     "feast.store.serving.options={\"host\":\"localhost\",\"port\":1234}",
-    "feast.store.errors.type=stderr"
+    "feast.store.errors.type=stderr",
+    "ingestion.metrics.enabled=true",
+    "ingestion.metrics.influxUrl=localhost",
+    "ingestion.metrics.dbName=db",
+    "ingestion.metrics.dbMeasurement=measurement"
 })
 @DirtiesContext
 public class CoreApplicationTest {
@@ -138,6 +142,10 @@ public void test_withProperties_systemServingAndWarehouseStoresRegistered() thro
             .setId(DEFAULT_WAREHOUSE_ID)
             .setType("file.json")
             .putOptions("path", "/tmp/foobar"))
+        .setWriteFeatureMetricsToInfluxDb(true)
+        .setInfluxDbUrl("localhost")
+        .setInfluxDbDatabase("db")
+        .setInfluxDbMeasurement("measurement")
         .build(), args.get(0));
   }
 

core/src/test/java/feast/core/CoreApplicationWithNoServingTest.java

@@ -42,7 +42,11 @@
     "spring.datasource.url=jdbc:h2:mem:testdb",
     "feast.store.warehouse.type=file.json",
     "feast.store.warehouse.options={\"path\":\"/tmp/foobar\"}",
-    "feast.store.errors.type=stderr"
+    "feast.store.errors.type=stderr",
+    "ingestion.metrics.enabled=true",
+    "ingestion.metrics.influxUrl=localhost",
+    "ingestion.metrics.dbName=db",
+    "ingestion.metrics.dbMeasurement=measurement"
 })
 @DirtiesContext
 public class CoreApplicationWithNoServingTest {

core/src/test/java/feast/core/CoreApplicationWithNoWarehouseTest.java

@@ -51,7 +51,11 @@
     "spring.datasource.url=jdbc:h2:mem:testdb",
     "feast.store.serving.type=redis",
     "feast.store.serving.options={\"host\":\"localhost\",\"port\":1234}",
-    "feast.store.errors.type=stderr"
+    "feast.store.errors.type=stderr",
+    "ingestion.metrics.enabled=true",
+    "ingestion.metrics.influxUrl=localhost",
+    "ingestion.metrics.dbName=db",
+    "ingestion.metrics.dbMeasurement=measurement"
 })
 @DirtiesContext
 public class CoreApplicationWithNoWarehouseTest {
@@ -132,6 +136,10 @@ public void test_withProperties_systemServingAndWarehouseStoresRegistered() thro
             .setId(DEFAULT_SERVING_ID)
             .setType("redis")
             .putOptions("host", "localhost").putOptions("port", "1234"))
+        .setWriteFeatureMetricsToInfluxDb(true)
+        .setInfluxDbUrl("localhost")
+        .setInfluxDbDatabase("db")
+        .setInfluxDbMeasurement("measurement")
         .build(), args.get(0));
   }