[feat] support fe/be metrics system table

be/src/exec/schema_scanner.cpp

@@ -30,6 +30,7 @@
 #include "exec/schema_scanner/schema_backend_active_tasks.h"
 #include "exec/schema_scanner/schema_backend_configuration_scanner.h"
 #include "exec/schema_scanner/schema_backend_kerberos_ticket_cache.h"
+#include "exec/schema_scanner/schema_backend_metrics_scanner.h"
 #include "exec/schema_scanner/schema_catalog_meta_cache_stats_scanner.h"
 #include "exec/schema_scanner/schema_charsets_scanner.h"
 #include "exec/schema_scanner/schema_cluster_snapshot_properties_scanner.h"
@@ -43,6 +44,7 @@
 #include "exec/schema_scanner/schema_file_cache_info_scanner.h"
 #include "exec/schema_scanner/schema_file_cache_statistics.h"
 #include "exec/schema_scanner/schema_files_scanner.h"
+#include "exec/schema_scanner/schema_frontend_metrics_scanner.h"
 #include "exec/schema_scanner/schema_load_job_scanner.h"
 #include "exec/schema_scanner/schema_metadata_name_ids_scanner.h"
 #include "exec/schema_scanner/schema_partitions_scanner.h"
@@ -262,6 +264,10 @@ std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type
         return SchemaColumnDataSizesScanner::create_unique();
     case TSchemaTableType::SCH_FILE_CACHE_INFO:
         return SchemaFileCacheInfoScanner::create_unique();
+    case TSchemaTableType::SCH_BE_METRICS:
+        return SchemaBackendMetricsScanner::create_unique();
+    case TSchemaTableType::SCH_FE_METRICS:
+        return SchemaFrontendMetricsScanner::create_unique();
     default:
         return SchemaDummyScanner::create_unique();
         break;

be/src/exec/schema_scanner/schema_backend_metrics_scanner.cpp

@@ -0,0 +1,86 @@
+// 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.
+
+#include "exec/schema_scanner/schema_backend_metrics_scanner.h"
+
+#include "runtime/exec_env.h"
+#include "runtime/runtime_query_statistics_mgr.h"
+#include "runtime/runtime_state.h"
+#include "vec/common/string_ref.h"
+#include "vec/core/block.h"
+#include "vec/data_types/data_type_factory.hpp"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+std::vector<SchemaScanner::ColumnDesc> SchemaBackendMetricsScanner::_s_tbls_columns = {
+        //   name,       type,          size,        is_null
+        {"BE_ID", TYPE_BIGINT, sizeof(int64_t), true},
+        {"BE_IP", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"METRIC_NAME", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"METRIC_TYPE", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"METRIC_VALUE", TYPE_DOUBLE, sizeof(double), true},
+        {"TAG", TYPE_VARCHAR, sizeof(StringRef), true}};
+
+SchemaBackendMetricsScanner::SchemaBackendMetricsScanner()
+        : SchemaScanner(_s_tbls_columns, TSchemaTableType::SCH_BE_METRICS) {}
+
+SchemaBackendMetricsScanner::~SchemaBackendMetricsScanner() {}
+
+Status SchemaBackendMetricsScanner::start(RuntimeState* state) {
+    _block_rows_limit = state->batch_size();
+    return Status::OK();
+}
+
+Status SchemaBackendMetricsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
+    if (!_is_init) {
+        return Status::InternalError("Used before initialized.");
+    }
+
+    if (nullptr == block || nullptr == eos) {
+        return Status::InternalError("input pointer is nullptr.");
+    }
+
+    if (_backend_metrics_block == nullptr) {
+        _backend_metrics_block = vectorized::Block::create_unique();
+        for (int i = 0; i < _s_tbls_columns.size(); ++i) {
+            auto data_type = vectorized::DataTypeFactory::instance().create_data_type(
+                    _s_tbls_columns[i].type, true);
+            _backend_metrics_block->insert(vectorized::ColumnWithTypeAndName(
+                    data_type->create_column(), data_type, _s_tbls_columns[i].name));
+        }
+        _backend_metrics_block->reserve(_block_rows_limit);
+        DorisMetrics::instance()->metric_registry()->get_be_metrics_block(
+                _backend_metrics_block.get());
+        _total_rows = (int)_backend_metrics_block->rows();
+    }
+
+    if (_row_idx == _total_rows) {
+        *eos = true;
+        return Status::OK();
+    }
+
+    int current_batch_rows = std::min(_block_rows_limit, _total_rows - _row_idx);
+    vectorized::MutableBlock mblock = vectorized::MutableBlock::build_mutable_block(block);
+    RETURN_IF_ERROR(mblock.add_rows(_backend_metrics_block.get(), _row_idx, current_batch_rows));
+    _row_idx += current_batch_rows;
+
+    *eos = _row_idx == _total_rows;
+    return Status::OK();
+}
+
+} // namespace doris

be/src/exec/schema_scanner/schema_backend_metrics_scanner.h

@@ -0,0 +1,49 @@
+// 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.
+
+#pragma once
+
+#include <vector>
+
+#include "common/status.h"
+#include "exec/schema_scanner.h"
+
+namespace doris {
+class RuntimeState;
+namespace vectorized {
+class Block;
+} // namespace vectorized
+
+class SchemaBackendMetricsScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaBackendMetricsScanner);
+
+public:
+    SchemaBackendMetricsScanner();
+    ~SchemaBackendMetricsScanner() override;
+
+    Status start(RuntimeState* state) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
+
+    static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
+
+private:
+    int _block_rows_limit = 4096;
+    int _row_idx = 0;
+    int _total_rows = 0;
+    std::unique_ptr<vectorized::Block> _backend_metrics_block = nullptr;
+};
+} // namespace doris

be/src/exec/schema_scanner/schema_frontend_metrics_scanner.cpp

@@ -0,0 +1,132 @@
+// 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.
+
+#include "exec/schema_scanner/schema_frontend_metrics_scanner.h"
+
+#include <gen_cpp/FrontendService_types.h>
+
+#include <exception>
+#include <vector>
+
+#include "exec/schema_scanner/schema_helper.h"
+#include "runtime/define_primitive_type.h"
+#include "runtime/runtime_state.h"
+#include "vec/common/string_ref.h"
+#include "vec/core/block.h"
+#include "vec/data_types/data_type_factory.hpp"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+std::vector<SchemaScanner::ColumnDesc> SchemaFrontendMetricsScanner::_s_frontend_metric_columns = {
+        //   name,       type,          size,        is_null
+        {"FE", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"METRIC_NAME", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"METRIC_TYPE", TYPE_VARCHAR, sizeof(StringRef), true},
+        {"METRIC_VALUE", TYPE_DOUBLE, sizeof(double), true},
+        {"TAG", TYPE_VARCHAR, sizeof(StringRef), true}};
+
+SchemaFrontendMetricsScanner::SchemaFrontendMetricsScanner()
+        : SchemaScanner(_s_frontend_metric_columns, TSchemaTableType::SCH_FE_METRICS) {}
+
+SchemaFrontendMetricsScanner::~SchemaFrontendMetricsScanner() = default;
+
+Status SchemaFrontendMetricsScanner::start(RuntimeState* state) {
+    TFetchFeMetricsRequest request;
+
+    for (const auto& fe_addr : _param->common_param->fe_addr_list) {
+        TFetchFeMetricsResult tmp_ret;
+        RETURN_IF_ERROR(SchemaHelper::fetch_frontend_metrics(fe_addr.hostname, fe_addr.port,
+                                                             request, &tmp_ret));
+
+        _metrics_list_result.metrics_list.insert(_metrics_list_result.metrics_list.end(),
+                                                 tmp_ret.metrics_list.begin(),
+                                                 tmp_ret.metrics_list.end());
+    }
+
+    return Status::OK();
+}
+
+Status SchemaFrontendMetricsScanner::get_next_block_internal(vectorized::Block* block, bool* eos) {
+    if (!_is_init) {
+        return Status::InternalError("call this before initial.");
+    }
+    if (block == nullptr || eos == nullptr) {
+        return Status::InternalError("invalid parameter.");
+    }
+
+    *eos = true;
+    if (_metrics_list_result.metrics_list.empty()) {
+        return Status::OK();
+    }
+
+    return _fill_block_impl(block);
+}
+
+Status SchemaFrontendMetricsScanner::_fill_block_impl(vectorized::Block* block) {
+    SCOPED_TIMER(_fill_block_timer);
+
+    const auto& metrics_list = _metrics_list_result.metrics_list;
+    size_t row_num = metrics_list.size();
+    if (row_num == 0) {
+        return Status::OK();
+    }
+
+    for (size_t col_idx = 0; col_idx < _s_frontend_metric_columns.size(); ++col_idx) {
+        std::vector<StringRef> str_refs(row_num);
+        std::vector<double> double_vals(row_num);
+        std::vector<void*> datas(row_num);
+        std::vector<std::string> column_values(
+                row_num); // Store the strings to ensure their lifetime
+
+        for (size_t row_idx = 0; row_idx < row_num; ++row_idx) {
+            const auto& row = metrics_list[row_idx];
+            if (row.size() != _s_frontend_metric_columns.size()) {
+                return Status::InternalError(
+                        "process list meet invalid schema, schema_size={}, input_data_size={}",
+                        _s_frontend_metric_columns.size(), row.size());
+            }
+
+            // Fetch and store the column value based on its index
+            std::string& column_value =
+                    column_values[row_idx]; // Reference to the actual string in the vector
+            column_value = row[col_idx];
+
+            if (_s_frontend_metric_columns[col_idx].type == TYPE_DOUBLE) {
+                try {
+                    double val = !column_value.empty() ? std::stod(column_value) : 0;
+                    double_vals[row_idx] = val;
+                } catch (const std::exception& e) {
+                    return Status::InternalError(
+                            "process list meet invalid data, column={}, data={}, reason={}",
+                            _s_frontend_metric_columns[col_idx].name, column_value, e.what());
+                }
+                datas[row_idx] = &double_vals[row_idx];
+            } else {
+                str_refs[row_idx] =
+                        StringRef(column_values[row_idx].data(), column_values[row_idx].size());
+                datas[row_idx] = &str_refs[row_idx];
+            }
+        }
+
+        RETURN_IF_ERROR(fill_dest_column_for_range(block, col_idx, datas));
+    }
+
+    return Status::OK();
+}
+
+} // namespace doris

be/src/exec/schema_scanner/schema_frontend_metrics_scanner.h

@@ -0,0 +1,53 @@
+// 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.
+
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "common/status.h"
+#include "exec/schema_scanner.h"
+#include "gen_cpp/FrontendService_types.h"
+
+namespace doris {
+
+class RuntimeState;
+
+namespace vectorized {
+class Block;
+}
+
+class SchemaFrontendMetricsScanner : public SchemaScanner {
+    ENABLE_FACTORY_CREATOR(SchemaFrontendMetricsScanner);
+
+public:
+    SchemaFrontendMetricsScanner();
+    ~SchemaFrontendMetricsScanner() override;
+
+    Status start(RuntimeState* state) override;
+    Status get_next_block_internal(vectorized::Block* block, bool* eos) override;
+
+    static std::vector<SchemaScanner::ColumnDesc> _s_tbls_columns;
+
+private:
+    Status _fill_block_impl(vectorized::Block* block);
+
+    TFetchFeMetricsResult _metrics_list_result;
+};
+
+} // namespace doris

be/src/exec/schema_scanner/schema_helper.cpp

@@ -178,4 +178,13 @@ Status SchemaHelper::get_master_keys(const std::string& ip, const int32_t port,
             });
 }
 
+Status SchemaHelper::fetch_frontend_metrics(const std::string& ip, const int32_t port,
+                                            const TFetchFeMetricsRequest& request,
+                                            TFetchFeMetricsResult* result) {
+    return ThriftRpcHelper::rpc<FrontendServiceClient>(
+            ip, port, [&request, &result](FrontendServiceConnection& client) {
+                client->fetchFrontendMetrics(*result, request);
+            });
+}
+
 } // namespace doris

be/src/exec/schema_scanner/schema_helper.h

@@ -46,6 +46,8 @@ class TShowProcessListRequest;
 class TShowProcessListResult;
 class TShowUserRequest;
 class TShowUserResult;
+class TFetchFeMetricsRequest;
+class TFetchFeMetricsResult;
 
 // this class is a helper for getting schema info from FE
 class SchemaHelper {
@@ -106,6 +108,10 @@ class SchemaHelper {
     static Status get_master_keys(const std::string& ip, const int32_t port,
                                   const TGetEncryptionKeysRequest& request,
                                   TGetEncryptionKeysResult* result);
+
+    static Status fetch_frontend_metrics(const std::string& ip, const int32_t port,
+                                         const TFetchFeMetricsRequest& request,
+                                         TFetchFeMetricsResult* result);
 };
 
 } // namespace doris