[Enhancement] Make exec_plan_fragment rpc timeout configable and add more metrics

fe/fe-core/src/main/java/com/starrocks/common/Config.java

@@ -843,6 +843,21 @@ public class Config extends ConfigBase {
     @ConfField
     public static int brpc_idle_wait_max_time = 10000;
 
+    @ConfField(mutable = true)
+    public static int brpc_send_plan_fragment_timeout_ms = 60000;
+
+    @ConfField
+    public static boolean brpc_reuse_addr = true;
+
+    @ConfField
+    public static int brpc_min_evictable_idle_time_ms = 120000;
+
+    @ConfField
+    public static boolean brpc_short_connection = false;
+
+    @ConfField
+    public static boolean brpc_inner_reuse_pool = true;
+
     /**
      * FE mysql server port
      */

fe/fe-core/src/main/java/com/starrocks/metric/MetricRepo.java

@@ -79,14 +79,20 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.lang.management.ManagementFactory;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
+import javax.management.Attribute;
+import javax.management.AttributeList;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
 
 public final class MetricRepo {
     private static final Logger LOG = LogManager.getLogger(MetricRepo.class);
@@ -162,13 +168,18 @@ public final class MetricRepo {
     public static LongCounterMetric COUNTER_ROUTINE_LOAD_PAUSED;
     public static LongCounterMetric COUNTER_SHORTCIRCUIT_QUERY;
     public static LongCounterMetric COUNTER_SHORTCIRCUIT_RPC;
+    public static LongCounterMetric COUNTER_BACKEND_SERVICE_RPC;
+    public static LongCounterMetric COUNTER_LAKE_SERVICE_RPC;
+    public static LongCounterMetric COUNTER_BRPC_EXEC_PLAN_FRAGMENT;
+    public static LongCounterMetric COUNTER_BRPC_EXEC_PLAN_FRAGMENT_ERROR;
 
     public static Histogram HISTO_QUERY_LATENCY;
     public static Histogram HISTO_EDIT_LOG_WRITE_LATENCY;
     public static Histogram HISTO_JOURNAL_WRITE_LATENCY;
     public static Histogram HISTO_JOURNAL_WRITE_BATCH;
     public static Histogram HISTO_JOURNAL_WRITE_BYTES;
     public static Histogram HISTO_SHORTCIRCUIT_RPC_LATENCY;
+    public static Histogram HISTO_DEPLOY_PLAN_FRAGMENTS_LATENCY;
 
     // following metrics will be updated by metric calculator
     public static GaugeMetricImpl<Double> GAUGE_QUERY_PER_SECOND;
@@ -492,6 +503,24 @@ public Long getValue() {
         STARROCKS_METRIC_REGISTER.addMetric(COUNTER_SHORTCIRCUIT_QUERY);
         COUNTER_SHORTCIRCUIT_RPC = new LongCounterMetric("shortcircuit_rpc", MetricUnit.REQUESTS, "total shortcircuit rpc");
         STARROCKS_METRIC_REGISTER.addMetric(COUNTER_SHORTCIRCUIT_RPC);
+        COUNTER_BACKEND_SERVICE_RPC = new LongCounterMetric("brpc_backend_service", MetricUnit.REQUESTS,
+                "total backend service rpc");
+        STARROCKS_METRIC_REGISTER.addMetric(COUNTER_BACKEND_SERVICE_RPC);
+        COUNTER_LAKE_SERVICE_RPC = new LongCounterMetric("brpc_lake_service", MetricUnit.REQUESTS, "total lake service rpc");
+        STARROCKS_METRIC_REGISTER.addMetric(COUNTER_LAKE_SERVICE_RPC);
+        GaugeMetric<Long> brpcTotal = (GaugeMetric<Long>) new GaugeMetric<Long>("brpc_total", MetricUnit.REQUESTS, "total brpc") {
+            @Override
+            public Long getValue() {
+                return COUNTER_BACKEND_SERVICE_RPC.getValue() + COUNTER_LAKE_SERVICE_RPC.getValue();
+            }
+        };
+        STARROCKS_METRIC_REGISTER.addMetric(brpcTotal);
+        COUNTER_BRPC_EXEC_PLAN_FRAGMENT = new LongCounterMetric(
+                "brpc_exec_plan_fragment", MetricUnit.REQUESTS, "total brpc exec plan fragment");
+        STARROCKS_METRIC_REGISTER.addMetric(COUNTER_BRPC_EXEC_PLAN_FRAGMENT);
+        COUNTER_BRPC_EXEC_PLAN_FRAGMENT_ERROR = new LongCounterMetric(
+                "brpc_exec_plan_fragment_error", MetricUnit.REQUESTS, "total brpc exec plan fragment error");
+        STARROCKS_METRIC_REGISTER.addMetric(COUNTER_BRPC_EXEC_PLAN_FRAGMENT_ERROR);
 
         COUNTER_QUERY_ANALYSIS_ERR = new LongCounterMetric("query_analysis_err", MetricUnit.REQUESTS,
                                                            "total analysis error query");
@@ -556,6 +585,8 @@ public Long getValue() {
         HISTO_JOURNAL_WRITE_BYTES =
                 METRIC_REGISTER.histogram(MetricRegistry.name("journal", "write", "bytes"));
         HISTO_SHORTCIRCUIT_RPC_LATENCY = METRIC_REGISTER.histogram(MetricRegistry.name("shortcircuit", "latency", "ms"));
+        HISTO_DEPLOY_PLAN_FRAGMENTS_LATENCY = METRIC_REGISTER.histogram(
+                MetricRegistry.name("deploy_plan_fragments", "latency", "ms"));
 
         // init system metrics
         initSystemMetrics();
@@ -853,6 +884,9 @@ public static synchronized String getMetric(MetricVisitor visitor, MetricsAction
         // collect starmgr related metrics as well
         StarMgrServer.getCurrentState().visitMetrics(visitor);
 
+        // collect brpc pool metrics
+        collectBrpcMetrics(visitor);
+
         // node info
         visitor.getNodeInfo();
         return visitor.build();
@@ -941,6 +975,49 @@ public Long getValue() {
         visitor.visit(databaseNum);
     }
 
+    private static void collectBrpcMetrics(MetricVisitor visitor) {
+        try {
+            MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
+            ObjectName pattern = new ObjectName("org.apache.commons.pool2:type=GenericObjectPool,name=*");
+            Set<ObjectName> objectNames = mBeanServer.queryNames(pattern, null);
+            if (objectNames.size() == 0) {
+                LOG.warn("failed to get GenericObjectPoolMXBean");
+                return;
+            }
+            String[] attrNames = new String[] {"NumActive", "NumIdle", "NumWaiters", "BorrowedCount",
+                    "ReturnedCount", "CreatedCount", "DestroyedCount", "MeanActiveTimeMillis", "MeanIdleTimeMillis",
+                    "MeanBorrowWaitTimeMillis"};
+            for (ObjectName objectName : objectNames) {
+                String name = objectName.getKeyProperty("name");
+                AttributeList attrs = mBeanServer.getAttributes(objectName, attrNames);
+                if (attrs.size() != attrNames.length) {
+                    LOG.warn("failed to get GenericObjectPoolMXBean attributes, attrs.size={}, attrNames.size={}",
+                            attrs.size(), attrNames.length);
+                    return;
+                }
+                for (int i = 0; i < attrs.size(); i++) {
+                    String attrName = attrNames[i];
+                    Object attr = ((Attribute) attrs.get(i)).getValue();
+                    if (attr instanceof Integer) {
+                        GaugeMetricImpl<Integer> metric = new GaugeMetricImpl<>(
+                                "brpc_pool_" + attrName.toLowerCase(), MetricUnit.NOUNIT, "brpc pool " + attrName);
+                        metric.addLabel(new MetricLabel("name", name));
+                        metric.setValue((Integer) attr);
+                        visitor.visit(metric);
+                    } else if (attr instanceof Long) {
+                        GaugeMetricImpl<Long> metric = new GaugeMetricImpl<>(
+                                "brpc_pool_" + attrName.toLowerCase(), MetricUnit.NOUNIT, "brpc pool " + attrName);
+                        metric.addLabel(new MetricLabel("name", name));
+                        metric.setValue((Long) attr);
+                        visitor.visit(metric);
+                    }
+                }
+            }
+        } catch (Throwable e) {
+            LOG.warn("failed to collect brpc metrics", e);
+        }
+    }
+
     private static void collectRoutineLoadProcessMetrics(MetricVisitor visitor) {
         for (GaugeMetricImpl<Long> metric : GAUGE_ROUTINE_LOAD_LAGS) {
             visitor.visit(metric);

fe/fe-core/src/main/java/com/starrocks/qe/DefaultCoordinator.java

@@ -59,6 +59,7 @@
 import com.starrocks.connector.exception.GlobalDictNotMatchException;
 import com.starrocks.connector.exception.RemoteFileNotFoundException;
 import com.starrocks.datacache.DataCacheSelectMetrics;
+import com.starrocks.metric.MetricRepo;
 import com.starrocks.mysql.MysqlCommand;
 import com.starrocks.planner.PlanFragment;
 import com.starrocks.planner.PlanFragmentId;
@@ -671,6 +672,7 @@ private void deliverExecFragments(ScheduleOption option) throws RpcException, St
                             this::handleErrorExecution, option.doDeploy);
             scheduler.prepareSchedule(this, deployer, executionDAG);
             this.scheduler.schedule(option);
+            MetricRepo.HISTO_DEPLOY_PLAN_FRAGMENTS_LATENCY.update(ignored.getTotalTime());
             queryProfile.attachExecutionProfiles(executionDAG.getExecutions());
         } finally {
             unlock();

fe/fe-core/src/main/java/com/starrocks/qe/scheduler/dag/FragmentInstanceExecState.java

@@ -19,6 +19,7 @@
 import com.starrocks.common.Status;
 import com.starrocks.common.util.DebugUtil;
 import com.starrocks.common.util.RuntimeProfile;
+import com.starrocks.metric.MetricRepo;
 import com.starrocks.planner.PlanFragmentId;
 import com.starrocks.proto.PExecPlanFragmentResult;
 import com.starrocks.proto.PPlanFragmentCancelReason;
@@ -290,9 +291,11 @@ public DeploymentResult waitForDeploymentCompletion(long deployTimeoutMs) {
             failure = e;
         }
 
+        MetricRepo.COUNTER_BRPC_EXEC_PLAN_FRAGMENT.increase(1L);
         if (code == TStatusCode.OK) {
             transitionState(State.DEPLOYING, State.EXECUTING);
         } else {
+            MetricRepo.COUNTER_BRPC_EXEC_PLAN_FRAGMENT_ERROR.increase(1L);
             transitionState(State.DEPLOYING, State.FAILED);
 
             if (errMsg == null) {

fe/fe-core/src/main/java/com/starrocks/rpc/BackendServiceClient.java

@@ -34,6 +34,7 @@
 
 package com.starrocks.rpc;
 
+import com.baidu.jprotobuf.pbrpc.utils.TalkTimeoutController;
 import com.google.common.base.Preconditions;
 import com.starrocks.common.Config;
 import com.starrocks.common.profile.Timer;
@@ -85,6 +86,7 @@ private Future<PExecPlanFragmentResult> sendPlanFragmentAsync(TNetworkAddress ad
         Tracers.count(Tracers.Module.SCHEDULER, "DeployDataSize", pRequest.serializedRequest.length);
         try (Timer ignored = Tracers.watchScope(Tracers.Module.SCHEDULER, "DeployAsyncSendTime")) {
             final PBackendService service = BrpcProxy.getBackendService(address);
+            TalkTimeoutController.setTalkTimeout(Config.brpc_send_plan_fragment_timeout_ms);
             return service.execPlanFragmentAsync(pRequest);
         } catch (NoSuchElementException e) {
             try {

fe/fe-core/src/main/java/com/starrocks/rpc/BrpcProxy.java

@@ -42,6 +42,11 @@ public BrpcProxy() {
         // Therefore, MaxIdleSize shouldn't less than MaxTotal for the async requests.
         rpcOptions.setMaxIdleSize(Config.brpc_connection_pool_size);
         rpcOptions.setMaxWait(Config.brpc_idle_wait_max_time);
+        rpcOptions.setJmxEnabled(true);
+        rpcOptions.setReuseAddress(Config.brpc_reuse_addr);
+        rpcOptions.setMinEvictableIdleTime(Config.brpc_min_evictable_idle_time_ms);
+        rpcOptions.setShortConnection(Config.brpc_short_connection);
+        rpcOptions.setInnerResuePool(Config.brpc_inner_reuse_pool);
 
         rpcClient = new RpcClient(rpcOptions);
         backendServiceMap = new ConcurrentHashMap<>();
@@ -87,14 +92,14 @@ private PBackendService createBackendService(TNetworkAddress address) {
         ProtobufRpcProxy<PBackendService> proxy = new ProtobufRpcProxy<>(rpcClient, PBackendService.class);
         proxy.setHost(address.getHostname());
         proxy.setPort(address.getPort());
-        return proxy.proxy();
+        return new PBackendServiceWithMetrics(proxy.proxy());
     }
 
     private LakeService createLakeService(TNetworkAddress address) {
         ProtobufRpcProxy<LakeService> proxy = new ProtobufRpcProxy<>(rpcClient, LakeService.class);
         proxy.setHost(address.getHostname());
         proxy.setPort(address.getPort());
-        return proxy.proxy();
+        return new LakeServiceWithMetrics(proxy.proxy());
     }
 
     private static class SingletonHolder {

fe/fe-core/src/main/java/com/starrocks/rpc/LakeServiceWithMetrics.java

@@ -0,0 +1,158 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// 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 com.starrocks.rpc;
+
+import com.starrocks.metric.MetricRepo;
+import com.starrocks.proto.AbortCompactionRequest;
+import com.starrocks.proto.AbortCompactionResponse;
+import com.starrocks.proto.AbortTxnRequest;
+import com.starrocks.proto.AbortTxnResponse;
+import com.starrocks.proto.CompactRequest;
+import com.starrocks.proto.CompactResponse;
+import com.starrocks.proto.DeleteDataRequest;
+import com.starrocks.proto.DeleteDataResponse;
+import com.starrocks.proto.DeleteTabletRequest;
+import com.starrocks.proto.DeleteTabletResponse;
+import com.starrocks.proto.DeleteTxnLogRequest;
+import com.starrocks.proto.DeleteTxnLogResponse;
+import com.starrocks.proto.DropTableRequest;
+import com.starrocks.proto.DropTableResponse;
+import com.starrocks.proto.LockTabletMetadataRequest;
+import com.starrocks.proto.LockTabletMetadataResponse;
+import com.starrocks.proto.PublishLogVersionBatchRequest;
+import com.starrocks.proto.PublishLogVersionRequest;
+import com.starrocks.proto.PublishLogVersionResponse;
+import com.starrocks.proto.PublishVersionRequest;
+import com.starrocks.proto.PublishVersionResponse;
+import com.starrocks.proto.RestoreSnapshotsRequest;
+import com.starrocks.proto.RestoreSnapshotsResponse;
+import com.starrocks.proto.TabletStatRequest;
+import com.starrocks.proto.TabletStatResponse;
+import com.starrocks.proto.UnlockTabletMetadataRequest;
+import com.starrocks.proto.UnlockTabletMetadataResponse;
+import com.starrocks.proto.UploadSnapshotsRequest;
+import com.starrocks.proto.UploadSnapshotsResponse;
+import com.starrocks.proto.VacuumRequest;
+import com.starrocks.proto.VacuumResponse;
+
+import java.util.concurrent.Future;
+
+public class LakeServiceWithMetrics implements LakeService {
+    final LakeService lakeService;
+    public LakeServiceWithMetrics(LakeService lakeService) {
+        this.lakeService = lakeService;
+    }
+
+    private static void increaseMetrics() {
+        if (MetricRepo.COUNTER_LAKE_SERVICE_RPC != null) {
+            MetricRepo.COUNTER_LAKE_SERVICE_RPC.increase(1L);
+        }
+    }
+
+    @Override
+    public Future<PublishVersionResponse> publishVersion(PublishVersionRequest request) {
+        increaseMetrics();
+        return lakeService.publishVersion(request);
+    }
+
+    @Override
+    public Future<AbortTxnResponse> abortTxn(AbortTxnRequest request) {
+        increaseMetrics();
+        return lakeService.abortTxn(request);
+    }
+
+    @Override
+    public Future<CompactResponse> compact(CompactRequest request) {
+        increaseMetrics();
+        return lakeService.compact(request);
+    }
+
+    @Override
+    public Future<DeleteTabletResponse> deleteTablet(DeleteTabletRequest request) {
+        increaseMetrics();
+        return lakeService.deleteTablet(request);
+    }
+
+    @Override
+    public Future<DeleteDataResponse> deleteData(DeleteDataRequest request) {
+        increaseMetrics();
+        return lakeService.deleteData(request);
+    }
+
+    @Override
+    public Future<DeleteTxnLogResponse> deleteTxnLog(DeleteTxnLogRequest request) {
+        increaseMetrics();
+        return lakeService.deleteTxnLog(request);
+    }
+
+    @Override
+    public Future<TabletStatResponse> getTabletStats(TabletStatRequest request) {
+        increaseMetrics();
+        return lakeService.getTabletStats(request);
+    }
+
+    @Override
+    public Future<DropTableResponse> dropTable(DropTableRequest request) {
+        increaseMetrics();
+        return lakeService.dropTable(request);
+    }
+
+    @Override
+    public Future<PublishLogVersionResponse> publishLogVersion(PublishLogVersionRequest request) {
+        increaseMetrics();
+        return lakeService.publishLogVersion(request);
+    }
+
+    @Override
+    public Future<PublishLogVersionResponse> publishLogVersionBatch(PublishLogVersionBatchRequest request) {
+        increaseMetrics();
+        return lakeService.publishLogVersionBatch(request);
+    }
+
+    @Override
+    public Future<LockTabletMetadataResponse> lockTabletMetadata(LockTabletMetadataRequest request) {
+        increaseMetrics();
+        return lakeService.lockTabletMetadata(request);
+    }
+
+    @Override
+    public Future<UnlockTabletMetadataResponse> unlockTabletMetadata(UnlockTabletMetadataRequest request) {
+        increaseMetrics();
+        return lakeService.unlockTabletMetadata(request);
+    }
+
+    @Override
+    public Future<UploadSnapshotsResponse> uploadSnapshots(UploadSnapshotsRequest request) {
+        increaseMetrics();
+        return lakeService.uploadSnapshots(request);
+    }
+
+    @Override
+    public Future<RestoreSnapshotsResponse> restoreSnapshots(RestoreSnapshotsRequest request) {
+        increaseMetrics();
+        return lakeService.restoreSnapshots(request);
+    }
+
+    @Override
+    public Future<AbortCompactionResponse> abortCompaction(AbortCompactionRequest request) {
+        increaseMetrics();
+        return lakeService.abortCompaction(request);
+    }
+
+    @Override
+    public Future<VacuumResponse> vacuum(VacuumRequest request) {
+        increaseMetrics();
+        return lakeService.vacuum(request);
+    }
+}