Add Feast Serving gRPC call metrics (#509)

* add feast serving grpc calls metrics

* edited documentation to reflect correct class

* separated into two metrics

* linting updates

serving/src/main/java/feast/serving/controller/HealthServiceController.java

@@ -18,6 +18,7 @@
 
 import feast.core.StoreProto.Store;
 import feast.serving.ServingAPIProto.GetFeastServingInfoRequest;
+import feast.serving.interceptors.GrpcMonitoringInterceptor;
 import feast.serving.service.ServingService;
 import feast.serving.specs.CachedSpecService;
 import io.grpc.health.v1.HealthGrpc.HealthImplBase;
@@ -30,7 +31,7 @@
 
 // Reference: https://github.com/grpc/grpc/blob/master/doc/health-checking.md
 
-@GRpcService
+@GRpcService(interceptors = {GrpcMonitoringInterceptor.class})
 public class HealthServiceController extends HealthImplBase {
   private CachedSpecService specService;
   private ServingService servingService;

serving/src/main/java/feast/serving/controller/ServingServiceGRpcController.java

@@ -26,6 +26,7 @@
 import feast.serving.ServingAPIProto.GetOnlineFeaturesRequest;
 import feast.serving.ServingAPIProto.GetOnlineFeaturesResponse;
 import feast.serving.ServingServiceGrpc.ServingServiceImplBase;
+import feast.serving.interceptors.GrpcMonitoringInterceptor;
 import feast.serving.service.ServingService;
 import feast.serving.util.RequestHelper;
 import io.grpc.stub.StreamObserver;
@@ -36,7 +37,7 @@
 import org.slf4j.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
 
-@GRpcService
+@GRpcService(interceptors = {GrpcMonitoringInterceptor.class})
 public class ServingServiceGRpcController extends ServingServiceImplBase {
 
   private static final Logger log =

serving/src/main/java/feast/serving/interceptors/GrpcMonitoringInterceptor.java

@@ -0,0 +1,55 @@
+/*
+ * 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.serving.interceptors;
+
+import feast.serving.util.Metrics;
+import io.grpc.ForwardingServerCall.SimpleForwardingServerCall;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCall.Listener;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.Status;
+
+/**
+ * GrpcMonitoringInterceptor intercepts GRPC calls to provide request latency histogram metrics in
+ * the Prometheus client.
+ */
+public class GrpcMonitoringInterceptor implements ServerInterceptor {
+
+  @Override
+  public <ReqT, RespT> Listener<ReqT> interceptCall(
+      ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
+
+    long startCallMillis = System.currentTimeMillis();
+    String fullMethodName = call.getMethodDescriptor().getFullMethodName();
+    String methodName = fullMethodName.substring(fullMethodName.indexOf("/") + 1);
+
+    return next.startCall(
+        new SimpleForwardingServerCall<ReqT, RespT>(call) {
+          @Override
+          public void close(Status status, Metadata trailers) {
+            Metrics.requestLatency
+                .labels(methodName)
+                .observe((System.currentTimeMillis() - startCallMillis) / 1000f);
+            Metrics.grpcRequestCount.labels(methodName, status.getCode().name()).inc();
+            super.close(status, trailers);
+          }
+        },
+        headers);
+  }
+}

serving/src/main/java/feast/serving/service/BigQueryServingService.java

@@ -18,7 +18,6 @@
 
 import static feast.serving.store.bigquery.QueryTemplater.createEntityTableUUIDQuery;
 import static feast.serving.store.bigquery.QueryTemplater.generateFullTableName;
-import static feast.serving.util.Metrics.requestLatency;
 
 import com.google.cloud.RetryOption;
 import com.google.cloud.bigquery.BigQuery;
@@ -116,7 +115,6 @@ public GetOnlineFeaturesResponse getOnlineFeatures(GetOnlineFeaturesRequest getF
   /** {@inheritDoc} */
   @Override
   public GetBatchFeaturesResponse getBatchFeatures(GetBatchFeaturesRequest getFeaturesRequest) {
-    long startTime = System.currentTimeMillis();
     List<FeatureSetRequest> featureSetRequests =
         specService.getFeatureSets(getFeaturesRequest.getFeaturesList());
 
@@ -168,9 +166,6 @@ public GetBatchFeaturesResponse getBatchFeatures(GetBatchFeaturesRequest getFeat
                 .build())
         .start();
 
-    requestLatency
-        .labels("getBatchFeatures")
-        .observe((System.currentTimeMillis() - startTime) / 1000);
     return GetBatchFeaturesResponse.newBuilder().setJob(feastJob).build();
   }
 

serving/src/main/java/feast/serving/service/RedisServingService.java

@@ -87,7 +87,6 @@ public GetFeastServingInfoResponse getFeastServingInfo(
   @Override
   public GetOnlineFeaturesResponse getOnlineFeatures(GetOnlineFeaturesRequest request) {
     try (Scope scope = tracer.buildSpan("Redis-getOnlineFeatures").startActive(true)) {
-      long startTime = System.currentTimeMillis();
       GetOnlineFeaturesResponse.Builder getOnlineFeaturesResponseBuilder =
           GetOnlineFeaturesResponse.newBuilder();
 
@@ -120,9 +119,6 @@ public GetOnlineFeaturesResponse getOnlineFeatures(GetOnlineFeaturesRequest requ
           featureValuesMap.values().stream()
               .map(valueMap -> FieldValues.newBuilder().putAllFields(valueMap).build())
               .collect(Collectors.toList());
-      requestLatency
-          .labels("getOnlineFeatures")
-          .observe((System.currentTimeMillis() - startTime) / 1000);
       return getOnlineFeaturesResponseBuilder.addAllFieldValues(fieldValues).build();
     }
   }

serving/src/main/java/feast/serving/util/Metrics.java

@@ -53,4 +53,12 @@ public class Metrics {
           .help("number requested feature rows that were stale")
           .labelNames("project", "feature_name")
           .register();
+
+  public static final Counter grpcRequestCount =
+      Counter.build()
+          .name("grpc_request_count")
+          .subsystem("feast_serving")
+          .help("number of grpc requests served")
+          .labelNames("method", "status_code")
+          .register();
 }