Merge pull request #6097 from halibobo1205/477/metric_http_grpc

feat(metric): add interceptors to monitor gRPC using Prometheus
tronprotocol · Nov 28, 2024 · 4a28e3d · 4a28e3d
2 parents ef08c67 + 39d99ee
commit 4a28e3d
Show file tree

Hide file tree

Showing 6 changed files with 75 additions and 8 deletions.
diff --git a/framework/src/main/java/org/tron/core/services/RpcApiService.java b/framework/src/main/java/org/tron/core/services/RpcApiService.java
@@ -98,6 +98,7 @@
 import org.tron.core.exception.ZksnarkException;
 import org.tron.core.metrics.MetricsApiService;
 import org.tron.core.services.filter.LiteFnQueryGrpcInterceptor;
+import org.tron.core.services.ratelimiter.PrometheusInterceptor;
 import org.tron.core.services.ratelimiter.RateLimiterInterceptor;
 import org.tron.core.services.ratelimiter.RpcApiAccessInterceptor;
 import org.tron.core.utils.TransactionUtil;
@@ -189,6 +190,8 @@ public class RpcApiService extends RpcService {
   private RpcApiAccessInterceptor apiAccessInterceptor;
   @Autowired
   private MetricsApiService metricsApiService;
+  @Autowired
+  private PrometheusInterceptor prometheusInterceptor;
   @Getter
   private DatabaseApi databaseApi = new DatabaseApi();
   private WalletApi walletApi = new WalletApi();
@@ -252,6 +255,11 @@ public void start() {
       // add lite fullnode query interceptor
       serverBuilder.intercept(liteFnQueryGrpcInterceptor);
 
+      // add prometheus interceptor
+      if (parameter.isMetricsPrometheusEnable()) {
+        serverBuilder.intercept(prometheusInterceptor);
+      }
+
       if (parameter.isRpcReflectionServiceEnable()) {
         serverBuilder.addService(ProtoReflectionService.newInstance());
       }

diff --git a/framework/src/main/java/org/tron/core/services/interfaceOnPBFT/RpcApiServiceOnPBFT.java b/framework/src/main/java/org/tron/core/services/interfaceOnPBFT/RpcApiServiceOnPBFT.java
@@ -41,6 +41,7 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.services.RpcApiService;
 import org.tron.core.services.filter.LiteFnQueryGrpcInterceptor;
+import org.tron.core.services.ratelimiter.PrometheusInterceptor;
 import org.tron.core.services.ratelimiter.RateLimiterInterceptor;
 import org.tron.core.services.ratelimiter.RpcApiAccessInterceptor;
 import org.tron.protos.Protocol.Account;
@@ -79,6 +80,9 @@ public class RpcApiServiceOnPBFT extends RpcService {
   @Autowired
   private RpcApiAccessInterceptor apiAccessInterceptor;
 
+  @Autowired
+  private PrometheusInterceptor prometheusInterceptor;
+
   private final String executorName = "rpc-pbft-executor";
 
   @Override
@@ -124,6 +128,11 @@ public void start() {
       // add lite fullnode query interceptor
       serverBuilder.intercept(liteFnQueryGrpcInterceptor);
 
+      // add prometheus interceptor
+      if (args.isMetricsPrometheusEnable()) {
+        serverBuilder.intercept(prometheusInterceptor);
+      }
+
       if (args.isRpcReflectionServiceEnable()) {
         serverBuilder.addService(ProtoReflectionService.newInstance());
       }

diff --git a/...ork/src/main/java/org/tron/core/services/interfaceOnSolidity/RpcApiServiceOnSolidity.java b/...ork/src/main/java/org/tron/core/services/interfaceOnSolidity/RpcApiServiceOnSolidity.java
@@ -42,6 +42,7 @@
 import org.tron.core.config.args.Args;
 import org.tron.core.services.RpcApiService;
 import org.tron.core.services.filter.LiteFnQueryGrpcInterceptor;
+import org.tron.core.services.ratelimiter.PrometheusInterceptor;
 import org.tron.core.services.ratelimiter.RateLimiterInterceptor;
 import org.tron.core.services.ratelimiter.RpcApiAccessInterceptor;
 import org.tron.protos.Protocol.Account;
@@ -81,6 +82,9 @@ public class RpcApiServiceOnSolidity extends RpcService {
   @Autowired
   private RpcApiAccessInterceptor apiAccessInterceptor;
 
+  @Autowired
+  private PrometheusInterceptor prometheusInterceptor;
+
   private final String executorName = "rpc-solidity-executor";
 
   @Override
@@ -125,6 +129,11 @@ public void start() {
       // add lite fullnode query interceptor
       serverBuilder.intercept(liteFnQueryGrpcInterceptor);
 
+      // add prometheus interceptor
+      if (parameter.isMetricsPrometheusEnable()) {
+        serverBuilder.intercept(prometheusInterceptor);
+      }
+
       if (parameter.isRpcReflectionServiceEnable()) {
         serverBuilder.addService(ProtoReflectionService.newInstance());
       }

diff --git a/framework/src/main/java/org/tron/core/services/ratelimiter/PrometheusInterceptor.java b/framework/src/main/java/org/tron/core/services/ratelimiter/PrometheusInterceptor.java
@@ -0,0 +1,46 @@
+
+package org.tron.core.services.ratelimiter;
+
+import io.grpc.ForwardingServerCall;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.Status;
+import io.prometheus.client.Histogram;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.tron.common.prometheus.MetricKeys;
+import org.tron.common.prometheus.Metrics;
+
+/**
+ * A {@link ServerInterceptor} which sends latency stats about incoming grpc calls to Prometheus.
+ */
+@Slf4j(topic = "metrics")
+@Component
+public class PrometheusInterceptor implements ServerInterceptor {
+
+  @Override
+  public <R, S> ServerCall.Listener<R> interceptCall(
+      ServerCall<R, S> call, Metadata requestMetadata, ServerCallHandler<R, S> next) {
+    return next.startCall(new MonitoringServerCall<>(call), requestMetadata);
+  }
+
+  static class MonitoringServerCall<R, S> extends ForwardingServerCall
+      .SimpleForwardingServerCall<R, S> {
+
+    private final Histogram.Timer requestTimer;
+
+    MonitoringServerCall(ServerCall<R, S> delegate) {
+      super(delegate);
+      this.requestTimer = Metrics.histogramStartTimer(
+          MetricKeys.Histogram.GRPC_SERVICE_LATENCY, getMethodDescriptor().getFullMethodName());
+    }
+
+    @Override
+    public void close(Status status, Metadata responseHeaders) {
+      Metrics.histogramObserve(requestTimer);
+      super.close(status, responseHeaders);
+    }
+  }
+}
diff --git a/framework/src/main/java/org/tron/core/services/ratelimiter/RpcApiAccessInterceptor.java b/framework/src/main/java/org/tron/core/services/ratelimiter/RpcApiAccessInterceptor.java
@@ -6,13 +6,10 @@
 import io.grpc.ServerCallHandler;
 import io.grpc.ServerInterceptor;
 import io.grpc.Status;
-import io.prometheus.client.Histogram;
 import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.tron.common.parameter.CommonParameter;
-import org.tron.common.prometheus.MetricKeys;
-import org.tron.common.prometheus.Metrics;
 
 @Slf4j
 @Component
@@ -32,11 +29,7 @@ public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
         return new ServerCall.Listener<ReqT>() {};
 
       } else {
-        Histogram.Timer requestTimer = Metrics.histogramStartTimer(
-            MetricKeys.Histogram.GRPC_SERVICE_LATENCY, endpoint);
-        Listener<ReqT> res = next.startCall(call, headers);
-        Metrics.histogramObserve(requestTimer);
-        return res;
+        return next.startCall(call, headers);
       }
     } catch (Exception e) {
       logger.error("check rpc api access Error: {}", e.getMessage());

diff --git a/framework/src/test/java/org/tron/core/services/RpcApiServicesTest.java b/framework/src/test/java/org/tron/core/services/RpcApiServicesTest.java
@@ -138,6 +138,8 @@ public static void init() throws IOException {
     getInstance().setRpcPort(PublicMethod.chooseRandomPort());
     getInstance().setRpcOnSolidityPort(PublicMethod.chooseRandomPort());
     getInstance().setRpcOnPBFTPort(PublicMethod.chooseRandomPort());
+    getInstance().setMetricsPrometheusPort(PublicMethod.chooseRandomPort());
+    getInstance().setMetricsPrometheusEnable(true);
     String fullNode = String.format("%s:%d", getInstance().getNodeLanIp(),
         getInstance().getRpcPort());
     String solidityNode = String.format("%s:%d", getInstance().getNodeLanIp(),