Extend WriteMetricsTransform in Ingestion to write feature value stat…

…s to StatsD (#486)

* Extend WriteMetricsTransform to write feature value stats to StatsD

* Apply mvn spotless

* Catch all exception not just StatsDClientException during init
Since there are other exception like UnknownHostException that can be thrown and we want to know such error. Also change the log level to error because so it's not normal for client to fail to be created"

* Change log level due to invalid feature set ref to error (previously warn)
On 2nd thought, this should constitute an error not a warning

* Apply maven spotless to metric transform codes

(cherry picked from commit 5508c92)

.prow/scripts/test-end-to-end.sh

@@ -66,6 +66,7 @@ sleep 20
 tail -n10 /var/log/kafka.log
 kafkacat -b localhost:9092 -L
 
+if [[ ${SKIP_BUILD_JARS} != "true" ]]; then
 echo "
 ============================================================
 Building jars for Feast
@@ -81,6 +82,9 @@ 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 "
 ============================================================

ingestion/pom.xml

@@ -248,5 +248,12 @@
       <version>2.8.1</version>
     </dependency>
 
+    <!-- For calculation of percentiles in feature values -->
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-math3</artifactId>
+      <version>3.6.1</version>
+    </dependency>
+
   </dependencies>
 </project>

ingestion/src/main/java/feast/ingestion/options/ImportOptions.java

@@ -26,6 +26,7 @@
 
 /** Options passed to Beam to influence the job's execution environment */
 public interface ImportOptions extends PipelineOptions, DataflowPipelineOptions, DirectOptions {
+
   @Required
   @Description(
       "JSON string representation of the FeatureSet that the import job will process, in BZip2 binary format."
@@ -83,4 +84,13 @@ public interface ImportOptions extends PipelineOptions, DataflowPipelineOptions,
   int getStatsdPort();
 
   void setStatsdPort(int StatsdPort);
+
+  @Description(
+      "Fixed window size in seconds (default 30) to apply before aggregation of numerical value of features"
+          + "and writing the aggregated value to StatsD. Refer to feast.ingestion.transform.metrics.WriteFeatureValueMetricsDoFn"
+          + "for details on the metric names and types.")
+  @Default.Integer(30)
+  int getWindowSizeInSecForFeatureValueMetric();
+
+  void setWindowSizeInSecForFeatureValueMetric(int seconds);
 }

ingestion/src/main/java/feast/ingestion/transform/metrics/WriteFeatureValueMetricsDoFn.java

@@ -0,0 +1,311 @@
+/*
+ * 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.ingestion.transform.metrics;
+
+import static feast.ingestion.transform.metrics.WriteRowMetricsDoFn.FEATURE_SET_NAME_TAG_KEY;
+import static feast.ingestion.transform.metrics.WriteRowMetricsDoFn.FEATURE_SET_PROJECT_TAG_KEY;
+import static feast.ingestion.transform.metrics.WriteRowMetricsDoFn.FEATURE_SET_VERSION_TAG_KEY;
+import static feast.ingestion.transform.metrics.WriteRowMetricsDoFn.FEATURE_TAG_KEY;
+import static feast.ingestion.transform.metrics.WriteRowMetricsDoFn.INGESTION_JOB_NAME_KEY;
+import static feast.ingestion.transform.metrics.WriteRowMetricsDoFn.METRIC_PREFIX;
+import static feast.ingestion.transform.metrics.WriteRowMetricsDoFn.STORE_TAG_KEY;
+
+import com.google.auto.value.AutoValue;
+import com.timgroup.statsd.NonBlockingStatsDClient;
+import com.timgroup.statsd.StatsDClient;
+import feast.types.FeatureRowProto.FeatureRow;
+import feast.types.FieldProto.Field;
+import feast.types.ValueProto.Value;
+import java.util.ArrayList;
+import java.util.DoubleSummaryStatistics;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.commons.math3.stat.descriptive.rank.Percentile;
+import org.slf4j.Logger;
+
+/**
+ * WriteFeatureValueMetricsDoFn accepts key value of FeatureSetRef(str) to FeatureRow(List) and
+ * writes a histogram of the numerical values of each feature to StatsD.
+ *
+ * <p>The histogram of the numerical values is represented as the following in StatsD:
+ *
+ * <ul>
+ *   <li>gauge of feature_value_min
+ *   <li>gauge of feature_value_max
+ *   <li>gauge of feature_value_mean
+ *   <li>gauge of feature_value_percentile_50
+ *   <li>gauge of feature_value_percentile_90
+ *   <li>gauge of feature_value_percentile_95
+ * </ul>
+ *
+ * <p>StatsD timing/histogram metric type is not used since it does not support negative values.
+ */
+@AutoValue
+public abstract class WriteFeatureValueMetricsDoFn
+    extends DoFn<KV<String, Iterable<FeatureRow>>, Void> {
+
+  abstract String getStoreName();
+
+  abstract String getStatsdHost();
+
+  abstract int getStatsdPort();
+
+  static Builder newBuilder() {
+    return new AutoValue_WriteFeatureValueMetricsDoFn.Builder();
+  }
+
+  @AutoValue.Builder
+  abstract static class Builder {
+
+    abstract Builder setStoreName(String storeName);
+
+    abstract Builder setStatsdHost(String statsdHost);
+
+    abstract Builder setStatsdPort(int statsdPort);
+
+    abstract WriteFeatureValueMetricsDoFn build();
+  }
+
+  private static final Logger log =
+      org.slf4j.LoggerFactory.getLogger(WriteFeatureValueMetricsDoFn.class);
+  private StatsDClient statsDClient;
+  public static String GAUGE_NAME_FEATURE_VALUE_MIN = "feature_value_min";
+  public static String GAUGE_NAME_FEATURE_VALUE_MAX = "feature_value_max";
+  public static String GAUGE_NAME_FEATURE_VALUE_MEAN = "feature_value_mean";
+  public static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_50 = "feature_value_percentile_50";
+  public static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_90 = "feature_value_percentile_90";
+  public static String GAUGE_NAME_FEATURE_VALUE_PERCENTILE_95 = "feature_value_percentile_95";
+
+  @Setup
+  public void setup() {
+    // Note that exception may be thrown during StatsD client instantiation but no exception
+    // will be thrown when sending metrics (mimicking the UDP protocol behaviour).
+    // https://jar-download.com/artifacts/com.datadoghq/java-dogstatsd-client/2.1.1/documentation
+    // https://github.com/DataDog/java-dogstatsd-client#unix-domain-socket-support
+    try {
+      statsDClient = new NonBlockingStatsDClient(METRIC_PREFIX, getStatsdHost(), getStatsdPort());
+    } catch (Exception e) {
+      log.error("StatsD client cannot be started: " + e.getMessage());
+    }
+  }
+
+  @Teardown
+  public void tearDown() {
+    if (statsDClient != null) {
+      statsDClient.close();
+    }
+  }
+
+  @ProcessElement
+  public void processElement(
+      ProcessContext context,
+      @Element KV<String, Iterable<FeatureRow>> featureSetRefToFeatureRows) {
+    if (statsDClient == null) {
+      return;
+    }
+
+    String featureSetRef = featureSetRefToFeatureRows.getKey();
+    if (featureSetRef == null) {
+      return;
+    }
+    String[] colonSplits = featureSetRef.split(":");
+    if (colonSplits.length != 2) {
+      log.error(
+          "Skip writing feature value metrics because the feature set reference '{}' does not"
+              + "follow the required format <project>/<feature_set_name>:<version>",
+          featureSetRef);
+      return;
+    }
+    String[] slashSplits = colonSplits[0].split("/");
+    if (slashSplits.length != 2) {
+      log.error(
+          "Skip writing feature value metrics because the feature set reference '{}' does not"
+              + "follow the required format <project>/<feature_set_name>:<version>",
+          featureSetRef);
+      return;
+    }
+    String projectName = slashSplits[0];
+    String featureSetName = slashSplits[1];
+    String version = colonSplits[1];
+
+    Map<String, DoubleSummaryStatistics> featureNameToStats = new HashMap<>();
+    Map<String, List<Double>> featureNameToValues = new HashMap<>();
+    for (FeatureRow featureRow : featureSetRefToFeatureRows.getValue()) {
+      for (Field field : featureRow.getFieldsList()) {
+        updateStats(featureNameToStats, featureNameToValues, field);
+      }
+    }
+
+    for (Entry<String, DoubleSummaryStatistics> entry : featureNameToStats.entrySet()) {
+      String featureName = entry.getKey();
+      DoubleSummaryStatistics stats = entry.getValue();
+      String[] tags = {
+        STORE_TAG_KEY + ":" + getStoreName(),
+        FEATURE_SET_PROJECT_TAG_KEY + ":" + projectName,
+        FEATURE_SET_NAME_TAG_KEY + ":" + featureSetName,
+        FEATURE_SET_VERSION_TAG_KEY + ":" + version,
+        FEATURE_TAG_KEY + ":" + featureName,
+        INGESTION_JOB_NAME_KEY + ":" + context.getPipelineOptions().getJobName()
+      };
+
+      // stats can return non finite values when there is no element
+      // or there is an element that is not a number. Metric should only be sent for finite values.
+      if (Double.isFinite(stats.getMin())) {
+        if (stats.getMin() < 0) {
+          // StatsD gauge will asssign a delta instead of the actual value, if there is a sign in
+          // the value. E.g. if the value is negative, a delta will be assigned. For this reason,
+          // the gauge value is set to zero beforehand.
+          // https://github.com/statsd/statsd/blob/master/docs/metric_types.md#gauges
+          statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_MIN, 0, tags);
+        }
+        statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_MIN, stats.getMin(), tags);
+      }
+      if (Double.isFinite(stats.getMax())) {
+        if (stats.getMax() < 0) {
+          statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_MAX, 0, tags);
+        }
+        statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_MAX, stats.getMax(), tags);
+      }
+      if (Double.isFinite(stats.getAverage())) {
+        if (stats.getAverage() < 0) {
+          statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_MEAN, 0, tags);
+        }
+        statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_MEAN, stats.getAverage(), tags);
+      }
+
+      // For percentile calculation, Percentile class from commons-math3 from Apache is used.
+      // Percentile requires double[], hence the conversion below.
+      if (!featureNameToValues.containsKey(featureName)) {
+        continue;
+      }
+      List<Double> valueList = featureNameToValues.get(featureName);
+      if (valueList == null || valueList.size() < 1) {
+        continue;
+      }
+      double[] values = new double[valueList.size()];
+      for (int i = 0; i < values.length; i++) {
+        values[i] = valueList.get(i);
+      }
+
+      double p50 = new Percentile().evaluate(values, 50);
+      if (p50 < 0) {
+        statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_PERCENTILE_50, 0, tags);
+      }
+      statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_PERCENTILE_50, p50, tags);
+
+      double p90 = new Percentile().evaluate(values, 90);
+      if (p90 < 0) {
+        statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_PERCENTILE_90, 0, tags);
+      }
+      statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_PERCENTILE_90, p90, tags);
+
+      double p95 = new Percentile().evaluate(values, 95);
+      if (p95 < 0) {
+        statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_PERCENTILE_95, 0, tags);
+      }
+      statsDClient.gauge(GAUGE_NAME_FEATURE_VALUE_PERCENTILE_95, p95, tags);
+    }
+  }
+
+  // Update stats and values array for the feature represented by the field.
+  // If the field contains non-numerical or non-boolean value, the stats and values array
+  // won't get updated because we are only concerned with numerical value in metrics data.
+  // For boolean value, true and false are treated as numerical value of 1 of 0 respectively.
+  private void updateStats(
+      Map<String, DoubleSummaryStatistics> featureNameToStats,
+      Map<String, List<Double>> featureNameToValues,
+      Field field) {
+    if (featureNameToStats == null || featureNameToValues == null || field == null) {
+      return;
+    }
+
+    String featureName = field.getName();
+    if (!featureNameToStats.containsKey(featureName)) {
+      featureNameToStats.put(featureName, new DoubleSummaryStatistics());
+    }
+    if (!featureNameToValues.containsKey(featureName)) {
+      featureNameToValues.put(featureName, new ArrayList<>());
+    }
+
+    Value value = field.getValue();
+    DoubleSummaryStatistics stats = featureNameToStats.get(featureName);
+    List<Double> values = featureNameToValues.get(featureName);
+
+    switch (value.getValCase()) {
+      case INT32_VAL:
+        stats.accept(value.getInt32Val());
+        values.add(((double) value.getInt32Val()));
+        break;
+      case INT64_VAL:
+        stats.accept(value.getInt64Val());
+        values.add((double) value.getInt64Val());
+        break;
+      case DOUBLE_VAL:
+        stats.accept(value.getDoubleVal());
+        values.add(value.getDoubleVal());
+        break;
+      case FLOAT_VAL:
+        stats.accept(value.getFloatVal());
+        values.add((double) value.getFloatVal());
+        break;
+      case BOOL_VAL:
+        stats.accept(value.getBoolVal() ? 1 : 0);
+        values.add(value.getBoolVal() ? 1d : 0d);
+        break;
+      case INT32_LIST_VAL:
+        for (Integer val : value.getInt32ListVal().getValList()) {
+          stats.accept(val);
+          values.add(((double) val));
+        }
+        break;
+      case INT64_LIST_VAL:
+        for (Long val : value.getInt64ListVal().getValList()) {
+          stats.accept(val);
+          values.add(((double) val));
+        }
+        break;
+      case DOUBLE_LIST_VAL:
+        for (Double val : value.getDoubleListVal().getValList()) {
+          stats.accept(val);
+          values.add(val);
+        }
+        break;
+      case FLOAT_LIST_VAL:
+        for (Float val : value.getFloatListVal().getValList()) {
+          stats.accept(val);
+          values.add(((double) val));
+        }
+        break;
+      case BOOL_LIST_VAL:
+        for (Boolean val : value.getBoolListVal().getValList()) {
+          stats.accept(val ? 1 : 0);
+          values.add(val ? 1d : 0d);
+        }
+        break;
+      case BYTES_VAL:
+      case BYTES_LIST_VAL:
+      case STRING_VAL:
+      case STRING_LIST_VAL:
+      case VAL_NOT_SET:
+      default:
+    }
+  }
+}