Add Feast Serving gRPC call metrics

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,56 @@
+/*
+ * 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.MethodDescriptor;
+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 serviceName = MethodDescriptor.extractFullServiceName(fullMethodName);
+    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(serviceName, methodName, status.getCode().name())
+                .observe((System.currentTimeMillis() - startCallMillis) / 1000f);
+            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

@@ -60,6 +60,7 @@
 
 public class RedisServingService implements ServingService {
 
+  private static final String SERVICE_NAME = "feast.serving.ServingService";
   private static final Logger log = org.slf4j.LoggerFactory.getLogger(RedisServingService.class);
   private final CachedSpecService specService;
   private final Tracer tracer;
@@ -87,7 +88,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 +120,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();
     }
   }
@@ -274,7 +271,7 @@ private void sendAndProcessMultiGet(
       }
     } finally {
       requestLatency
-          .labels("processResponse")
+          .labels(SERVICE_NAME, "processResponse", "NA")
           .observe((System.currentTimeMillis() - startTime) / 1000);
     }
   }
@@ -317,7 +314,7 @@ private List<byte[]> sendMultiGet(List<RedisKey> keys) {
             .asRuntimeException();
       } finally {
         requestLatency
-            .labels("sendMultiGet")
+            .labels(SERVICE_NAME, "sendMultiGet", "NA")
             .observe((System.currentTimeMillis() - startTime) / 1000d);
       }
     }

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

@@ -27,7 +27,7 @@ public class Metrics {
           .name("request_latency_seconds")
           .subsystem("feast_serving")
           .help("Request latency in seconds")
-          .labelNames("method")
+          .labelNames("service", "method", "status_code")
           .register();
 
   public static final Counter requestCount =