[Proc] Add stream load info to system info of web site (#6970)

#6969

fe/fe-core/src/main/java/org/apache/doris/common/proc/ProcService.java

@@ -51,6 +51,7 @@ public final class ProcService {
         root.register("current_backend_instances", new CurrentQueryBackendInstanceProcDir());
         root.register("cluster_balance", new ClusterBalanceProcDir());
         root.register("routine_loads", new RoutineLoadsProcDir());
+        root.register("stream_loads", new StreamLoadProcNode());
         root.register("colocation_group", new ColocationGroupProcDir());
         root.register("bdbje", new BDBJEProcDir());
     }

fe/fe-core/src/main/java/org/apache/doris/common/proc/StreamLoadProcNode.java

@@ -0,0 +1,58 @@
+// 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.doris.common.proc;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.load.StreamLoadRecordMgr;
+
+import java.util.List;
+
+/*
+    SHOW PROC "/stream_loads"
+    show statistic of all of running loads
+
+    RESULT:
+    Label | DbId | FinishTime
+ */
+public class StreamLoadProcNode implements ProcNodeInterface {
+
+    private static final ImmutableList<String> TITLE_NAMES =
+            new ImmutableList.Builder<String>()
+                    .add("Label")
+                    .add("DbId")
+                    .add("FinishTime")
+                    .build();
+
+    @Override
+    public ProcResult fetchResult() throws AnalysisException {
+        BaseProcResult baseProcResult = new BaseProcResult();
+        baseProcResult.setNames(TITLE_NAMES);
+        StreamLoadRecordMgr streamLoadRecordMgr = Catalog.getCurrentCatalog().getStreamLoadRecordMgr();
+        try {
+            List<StreamLoadRecordMgr.StreamLoadItem> streamLoadJobList = streamLoadRecordMgr.getStreamLoadRecords();
+            for (StreamLoadRecordMgr.StreamLoadItem streamLoadItem : streamLoadJobList) {
+                baseProcResult.addRow(streamLoadItem.getStatistics());
+            }
+        } catch (Exception e) {
+            throw new AnalysisException("failed to get all of stream load job");
+        }
+        return baseProcResult;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadRecordMgr.java

@@ -51,6 +51,7 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.text.SimpleDateFormat;
+import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -67,7 +68,7 @@ import com.google.gson.annotations.SerializedName;
 public class StreamLoadRecordMgr extends MasterDaemon {
     private static final Logger LOG = LogManager.getLogger(StreamLoadRecordMgr.class);
 
-    private class StreamLoadItem {
+    public class StreamLoadItem {
         private String label;
         private long dbId;
         private String finishTime;
@@ -89,6 +90,14 @@ public class StreamLoadRecordMgr extends MasterDaemon {
         public String getFinishTime() {
             return finishTime;
         }
+
+        public List<String> getStatistics() {
+            List<String> row = Lists.newArrayList();
+            row.add(label);
+            row.add(String.valueOf(dbId));
+            row.add(finishTime);
+            return row;
+        }
     }
 
     class StreamLoadComparator implements Comparator<StreamLoadItem> {
@@ -139,6 +148,10 @@ public class StreamLoadRecordMgr extends MasterDaemon {
         writeUnlock();
     }
 
+    public List<StreamLoadItem> getStreamLoadRecords() {
+        return new ArrayList<>(streamLoadRecordHeap);
+    }
+
     public List<List<Comparable>> getStreamLoadRecordByDb(long dbId, String label, boolean accurateMatch, StreamLoadState state) {
         LinkedList<List<Comparable>> streamLoadRecords = new LinkedList<List<Comparable>>();