[feature] (Metric) add jni metrics for jdbc connection

be/src/common/config.cpp

@@ -587,6 +587,9 @@ DEFINE_mBool(enable_token_check, "true");
 // to open/close system metrics
 DEFINE_Bool(enable_system_metrics, "true");
 
+// to open/close jni metrics
+DEFINE_Bool(enable_jni_metrics, "true");
+
 // Number of cores Doris will used, this will effect only when it's greater than 0.
 // Otherwise, Doris will use all cores returned from "/proc/cpuinfo".
 DEFINE_Int32(num_cores, "0");

be/src/common/config.h

@@ -637,6 +637,9 @@ DECLARE_mBool(enable_token_check);
 // to open/close system metrics
 DECLARE_Bool(enable_system_metrics);
 
+// to open/close jni metrics
+DECLARE_Bool(enable_jni_metrics);
+
 // Number of cores Doris will used, this will effect only when it's greater than 0.
 // Otherwise, Doris will use all cores returned from "/proc/cpuinfo".
 DECLARE_Int32(num_cores);

be/src/util/doris_metrics.cpp

@@ -336,6 +336,12 @@ void DorisMetrics::init_jvm_metrics(JNIEnv* env) {
     _jvm_metrics.reset(new JvmMetrics(&_metric_registry, env));
 }
 
+void DorisMetrics::init_jni_metrics() {
+    if (config::enable_jni_metrics) {
+        _jni_metrics.reset(new JniMetrics(&_metric_registry));
+    }
+}
+
 void DorisMetrics::_update() {
     _update_process_thread_num();
     _update_process_fd_num();

be/src/util/doris_metrics.h

@@ -24,6 +24,7 @@
 #include <string>
 #include <vector>
 
+#include "util/jni_metrics.h"
 #include "util/jvm_metrics.h"
 #include "util/metrics.h"
 #include "util/system_metrics.h"
@@ -257,7 +258,9 @@ class DorisMetrics {
     SystemMetrics* system_metrics() { return _system_metrics.get(); }
     MetricEntity* server_entity() { return _server_metric_entity.get(); }
     JvmMetrics* jvm_metrics() { return _jvm_metrics.get(); }
+    JniMetrics* jni_metrics() { return _jni_metrics.get(); }
     void init_jvm_metrics(JNIEnv* env);
+    void init_jni_metrics();
 
 private:
     // Don't allow constructor
@@ -275,6 +278,7 @@ class DorisMetrics {
 
     std::unique_ptr<SystemMetrics> _system_metrics;
     std::unique_ptr<JvmMetrics> _jvm_metrics;
+    std::unique_ptr<JniMetrics> _jni_metrics;
 
     std::shared_ptr<MetricEntity> _server_metric_entity;
 };

be/src/util/jni-util.cpp

@@ -604,6 +604,7 @@ Status JniUtil::Init() {
     RETURN_IF_ERROR(init_jni_scanner_loader(env));
     jvm_inited_ = true;
     DorisMetrics::instance()->init_jvm_metrics(env);
+    DorisMetrics::instance()->init_jni_metrics();
     return Status::OK();
 }
 

be/src/util/jni_metrics.cpp

@@ -0,0 +1,112 @@
+// 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 "jni_metrics.h"
+
+#include <memory>
+#include <utility>
+
+#include "util/jni-util.h"
+#include "util/metrics.h"
+
+namespace doris {
+
+DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(jdbc_scan_connection_percent, MetricUnit::PERCENT);
+struct JdbcConnectionMetrics {
+    JdbcConnectionMetrics(std::shared_ptr<MetricEntity> entity) : entity(entity) {
+        INT_GAUGE_METRIC_REGISTER(entity, jdbc_scan_connection_percent);
+    }
+    void upate(int value) { jdbc_scan_connection_percent->set_value(value); }
+    std::shared_ptr<MetricEntity> entity;
+    IntGauge* jdbc_scan_connection_percent;
+};
+
+const char* JniMetrics::_s_hook_name = "jni_metrics";
+
+JniMetrics::JniMetrics(MetricRegistry* registry) {
+    DCHECK(registry != nullptr);
+    _registry = registry;
+    _server_entity = _registry->register_entity("server");
+    DCHECK(_server_entity != nullptr);
+    Status st = _init();
+    if (!st.ok()) {
+        LOG(WARNING) << "init jni metric failed. " << st.to_string();
+    }
+}
+
+Status JniMetrics::_init() {
+    JNIEnv* env = nullptr;
+    RETURN_IF_ERROR(JniUtil::GetJNIEnv(&env));
+    RETURN_IF_ERROR(JniUtil::get_jni_scanner_class(env, "org/apache/doris/jdbc/JdbcDataSource",
+                                        &_jdbc_data_source_clz));
+    JNI_CALL_METHOD_CHECK_EXCEPTION(
+         , _get_connection_percent_id, env,
+         GetStaticMethodID(_jdbc_data_source_clz, "getConnectionPercent",
+                           "()Ljava/util/Map;"));
+    _is_init = true;
+    _server_entity->register_hook(_s_hook_name,
+                                  std::bind(&JniMetrics::update_jdbc_connection_metrics, this));
+    LOG(INFO) << "jni metrics inited successfully";
+    return Status::OK();
+}
+
+Status JniMetrics::update_jdbc_connection_metrics() {
+    JNIEnv* env = nullptr;
+    RETURN_IF_ERROR(JniUtil::GetJNIEnv(&env));
+    JNI_CALL_METHOD_CHECK_EXCEPTION_DELETE_REF(
+         jobject, metrics, env,
+         CallStaticObjectMethod(_jdbc_data_source_clz, _get_connection_percent_id));
+    std::map<std::string, std::string> result = JniUtil::convert_to_cpp_map(env, metrics);
+    for (auto item : result) {
+        std::string catalog_id = item.first;
+        int percent = std::stoi(item.second);
+        auto iter = _jdbc_connection_metrics.find(catalog_id);
+        if (iter == _jdbc_connection_metrics.end()) {
+            auto entity =
+                    _registry->register_entity_unlocked(catalog_id, {{"catalog", catalog_id}});
+            _jdbc_connection_metrics.emplace(catalog_id,
+                                             std::make_shared<JdbcConnectionMetrics>(entity));
+        }
+        _jdbc_connection_metrics[catalog_id]->upate(percent);
+    }
+    // remove unused catalog
+    for (auto& item : _jdbc_connection_metrics) {
+        auto iter = result.find(item.first);
+        if (iter == result.end()) {
+            _jdbc_connection_metrics.erase(item.first);
+            _registry->deregister_entity_unlocked(item.second->entity);
+            LOG(INFO) << "catalog id : " << item.first << " unused, removed.";
+        }
+    }
+    return Status::OK();
+}
+
+JniMetrics::~JniMetrics() {
+    JNIEnv* env = nullptr;
+    Status st = JniUtil::GetJNIEnv(&env);
+    if (_is_init && st.ok()) {
+        env->DeleteGlobalRef(_jdbc_data_source_clz);
+    } else {
+        LOG(WARNING) << "get jni env failed. " << st.to_string();
+    }
+    _server_entity->deregister_hook(_s_hook_name);
+    for (auto& it : _jdbc_connection_metrics) {
+        _registry->deregister_entity(it.second->entity);
+    }
+}
+
+} // namespace doris

be/src/util/jni_metrics.h

@@ -0,0 +1,50 @@
+// 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 "jni.h"
+#include "util/metrics.h"
+
+namespace doris {
+
+struct JdbcConnectionMetrics;
+
+class JniMetrics {
+public:
+    JniMetrics(MetricRegistry* registry);
+    ~JniMetrics();
+    Status update_jdbc_connection_metrics();
+
+private:
+    Status _init();
+
+private:
+    static const char* _s_hook_name;
+    bool _is_init = false;
+    MetricRegistry* _registry = nullptr;
+    std::shared_ptr<MetricEntity> _server_entity;
+
+    jclass _jdbc_data_source_clz;
+    jmethodID _get_connection_percent_id;
+    std::unordered_map<std::string, std::shared_ptr<JdbcConnectionMetrics>>
+            _jdbc_connection_metrics;
+};
+
+} // namespace doris

be/src/util/metrics.cpp

@@ -275,8 +275,14 @@ MetricRegistry::~MetricRegistry() {}
 std::shared_ptr<MetricEntity> MetricRegistry::register_entity(const std::string& name,
                                                               const Labels& labels,
                                                               MetricEntityType type) {
-    std::shared_ptr<MetricEntity> entity = std::make_shared<MetricEntity>(type, name, labels);
     std::lock_guard<std::mutex> l(_lock);
+    return register_entity_unlocked(name, labels, type);
+}
+
+std::shared_ptr<MetricEntity> MetricRegistry::register_entity_unlocked(const std::string& name,
+                                                                       const Labels& labels,
+                                                                       MetricEntityType type) {
+    std::shared_ptr<MetricEntity> entity = std::make_shared<MetricEntity>(type, name, labels);
     auto inserted_entity = _entities.insert(std::make_pair(entity, 1));
     if (!inserted_entity.second) {
         // If exist, increase the registered count
@@ -287,6 +293,10 @@ std::shared_ptr<MetricEntity> MetricRegistry::register_entity(const std::string&
 
 void MetricRegistry::deregister_entity(const std::shared_ptr<MetricEntity>& entity) {
     std::lock_guard<std::mutex> l(_lock);
+    deregister_entity_unlocked(entity);
+}
+
+void MetricRegistry::deregister_entity_unlocked(const std::shared_ptr<MetricEntity>& entity) {
     auto found_entity = _entities.find(entity);
     if (found_entity != _entities.end()) {
         // Decrease the registered count

be/src/util/metrics.h

@@ -407,7 +407,11 @@ class MetricRegistry {
     std::shared_ptr<MetricEntity> register_entity(
             const std::string& name, const Labels& labels = {},
             MetricEntityType type = MetricEntityType::kServer);
+    std::shared_ptr<MetricEntity> register_entity_unlocked(
+            const std::string& name, const Labels& labels = {},
+            MetricEntityType type = MetricEntityType::kServer);
     void deregister_entity(const std::shared_ptr<MetricEntity>& entity);
+    void deregister_entity_unlocked(const std::shared_ptr<MetricEntity>& entity);
     std::shared_ptr<MetricEntity> get_entity(const std::string& name, const Labels& labels = {},
                                              MetricEntityType type = MetricEntityType::kServer);
 

fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/JdbcDataSource.java

@@ -20,6 +20,7 @@
 import com.zaxxer.hikari.HikariDataSource;
 import org.apache.log4j.Logger;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -93,4 +94,22 @@ private void startCleanupTask() {
             restartCleanupTask();
         }
     }
+
+    /**
+     * used for jni metrics
+     * get jdbc scan connection percent
+     * percent = activeConnections * 100 / maximumPoolSize
+     */
+    public static Map<String, String> getConnectionPercent() {
+        Map<String, String> result = new HashMap<>();
+        for (String key : getDataSource().lastAccessTimeMap.keySet()) {
+            HikariDataSource ds = getDataSource().sourcesMap.get(key);
+            int percent = 0;
+            if (ds != null) {
+                percent = ds.getHikariPoolMXBean().getActiveConnections() * 100 / ds.getMaximumPoolSize();
+            }
+            result.put(key.split("jdbc")[0], String.valueOf(percent));
+        }
+        return result;
+    }
 }

regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy

@@ -2127,7 +2127,7 @@ class Suite implements GroovyInterceptable {
         def js = jsonOutput.toJson(map)
         log.info("get be metric req: ${js} ".toString())
 
-        def ret = 0;
+        def ret = Integer.MAX_VALUE
         metric_api.call(js) {
             respCode, body ->
                 log.info("get be metric resp: ${respCode}".toString())