[iotdb] branch QueryMetrics0.13 updated: Add metrics for session

[ja...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jackietien pushed a commit to branch QueryMetrics0.13
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/QueryMetrics0.13 by this push:
     new 96620dbf85 Add metrics for session
96620dbf85 is described below

commit 96620dbf8585aad6eade07a5adaf458fc930b287
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Sun Nov 13 10:20:56 2022 +0800

    Add metrics for session
---
 .../java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java |  3 +
 .../src/main/java/org/apache/iotdb/rpc/RpcRT.java  | 75 ++++++++++++++++++++++
 .../apache/iotdb/session/SessionConnection.java    |  4 ++
 3 files changed, 82 insertions(+)

diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java
index fe2ab91465..37d6be8e11 100644
--- a/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/IoTDBRpcDataSet.java
@@ -230,6 +230,7 @@ public class IoTDBRpcDataSet {
     rowsIndex = 0;
     TSFetchResultsReq req = new TSFetchResultsReq(sessionId, sql, fetchSize, queryId, true);
     req.setTimeout(timeout);
+    long startTime = System.nanoTime();
     try {
       TSFetchResultsResp resp = client.fetchResults(req);
 
@@ -244,6 +245,8 @@ public class IoTDBRpcDataSet {
     } catch (TException e) {
       throw new IoTDBConnectionException(
           "Cannot fetch result from server, because of network connection: {} ", e);
+    } finally {
+      RpcRT.getInstance().addCost(System.nanoTime() - startTime);
     }
   }
 
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcRT.java b/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcRT.java
new file mode 100644
index 0000000000..24309f059a
--- /dev/null
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcRT.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *     http://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 org.apache.iotdb.rpc;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class RpcRT {
+
+  private static final long RPC_RT_PRINT_INTERVAL_IN_MS = 5_000;
+
+  private static final Logger RPC_RT_LOGGER = LoggerFactory.getLogger(RpcRT.class);
+
+  private final AtomicLong totalTime;
+  private final AtomicLong totalCount;
+
+  private RpcRT() {
+    this.totalTime = new AtomicLong(0);
+    this.totalCount = new AtomicLong(0);
+
+    ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(1);
+    scheduledExecutor.scheduleAtFixedRate(
+        this::printRpcRTStatistics,
+        2 * RPC_RT_PRINT_INTERVAL_IN_MS,
+        RPC_RT_PRINT_INTERVAL_IN_MS,
+        TimeUnit.MILLISECONDS);
+  }
+
+  private void printRpcRTStatistics() {
+    long totalTime = this.totalTime.get() / 1_000;
+    long totalCount = this.totalCount.get();
+    RPC_RT_LOGGER.info(
+        "rpc total time: {}us, rpc total count: {}, rpc avg time: {}us",
+        totalTime,
+        totalCount,
+        (totalTime / totalCount));
+  }
+
+  public void addCost(long costTimeInNanos) {
+    totalTime.addAndGet(costTimeInNanos);
+    totalCount.incrementAndGet();
+  }
+
+  public static RpcRT getInstance() {
+    return RpcRTHolder.INSTANCE;
+  }
+
+  private static class RpcRTHolder {
+
+    private static final RpcRT INSTANCE = new RpcRT();
+
+    private RpcRTHolder() {}
+  }
+}
diff --git a/session/src/main/java/org/apache/iotdb/session/SessionConnection.java b/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
index 1f0c6b6919..75441a5e10 100644
--- a/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
+++ b/session/src/main/java/org/apache/iotdb/session/SessionConnection.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.session;
 
 import org.apache.iotdb.rpc.IoTDBConnectionException;
 import org.apache.iotdb.rpc.RedirectException;
+import org.apache.iotdb.rpc.RpcRT;
 import org.apache.iotdb.rpc.RpcTransportFactory;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.SessionTimeoutException;
@@ -366,6 +367,7 @@ public class SessionConnection {
     execReq.setFetchSize(session.fetchSize);
     execReq.setTimeout(timeout);
     TSExecuteStatementResp execResp;
+    long startTime = System.nanoTime();
     try {
       execReq.setEnableRedirectQuery(enableRedirect);
       execResp = client.executeQueryStatement(execReq);
@@ -383,6 +385,8 @@ public class SessionConnection {
       } else {
         throw new IoTDBConnectionException(logForReconnectionFailure());
       }
+    } finally {
+      RpcRT.getInstance().addCost(System.nanoTime() - startTime);
     }
     return new SessionDataSet(
         sql,