feat: enhance connection pool management

rm-datasource/src/main/java/org/apache/seata/rm/datasource/AbstractConnectionPoolManager.java

@@ -0,0 +1,4 @@
+package org.apache.seata.rm.datasource;
+
+public abstract class AbstractConnectionPoolManager {
+}

rm-datasource/src/main/java/org/apache/seata/rm/datasource/ConnectionPoolMetricsSender.java

@@ -0,0 +1,51 @@
+/*
+ * 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.seata.rm.datasource;
+
+import org.apache.seata.rm.datasource.entity.ConnectionPoolMetrics;
+
+import java.util.concurrent.LinkedBlockingQueue;
+
+public class ConnectionPoolMetricsSender {
+
+    private final static LinkedBlockingQueue<ConnectionPoolMetrics> queue = new LinkedBlockingQueue<>(600);
+
+    private static ConnectionPoolMetricsSender INSTANCE;
+
+    public static ConnectionPoolMetricsSender getInstance() {
+        if (INSTANCE == null) {
+            synchronized (ConnectionPoolMetricsSender.class) {
+                if (INSTANCE == null) {
+                    INSTANCE = new ConnectionPoolMetricsSender();
+                }
+            }
+        }
+        return INSTANCE;
+    }
+
+    private ConnectionPoolMetricsSender() {
+
+    }
+
+    public void offer(ConnectionPoolMetrics metrics) {
+        if (!queue.offer(metrics)) {
+            queue.poll();
+            queue.offer(metrics);
+        }
+    }
+
+}

rm-datasource/src/main/java/org/apache/seata/rm/datasource/ConnectionPoolService.java

@@ -0,0 +1,46 @@
+/*
+ * 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.seata.rm.datasource;
+
+import org.apache.seata.rm.datasource.entity.ConnectionPoolMetrics;
+
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+public class ConnectionPoolService implements Runnable {
+
+    private DruidConnectionPoolManager manager = DruidConnectionPoolManager.getInstance();
+
+    private ConnectionPoolMetricsSender sender = ConnectionPoolMetricsSender.getInstance();
+
+    public void init() throws Exception {
+        Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread thread = new Thread(r);
+            thread.setDaemon(true);
+            return thread;
+        }).scheduleAtFixedRate(this, 0, 1, TimeUnit.SECONDS);
+    }
+
+
+    @Override
+    public void run() {
+        List<ConnectionPoolMetrics> metricsList = manager.getConnectionPoolMetricsList();
+
+        metricsList.forEach(metrics -> sender.offer(metrics));
+    }
+}

rm-datasource/src/main/java/org/apache/seata/rm/datasource/DruidConnectionPoolManager.java

@@ -0,0 +1,131 @@
+/*
+ * 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.seata.rm.datasource;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import org.apache.seata.core.model.Resource;
+import org.apache.seata.rm.datasource.entity.ConnectionPoolMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+
+public class DruidConnectionPoolManager extends AbstractConnectionPoolManager {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DruidConnectionPoolManager.class);
+
+    private final Map<String, DruidDataSource> druidDataSourceMap = new ConcurrentHashMap<>();
+
+    private static DruidConnectionPoolManager INSTANCE;
+
+    public static DruidConnectionPoolManager getInstance() {
+        if (INSTANCE == null) {
+            synchronized (DruidConnectionPoolManager.class) {
+                if (INSTANCE == null) {
+                    INSTANCE = new DruidConnectionPoolManager();
+                }
+            }
+        }
+        return INSTANCE;
+    }
+
+    private DruidConnectionPoolManager() {
+
+    }
+
+    public void registerDataSource(Resource resource) {
+        if (resource instanceof DataSourceProxy) {
+            DataSourceProxy dataSourceProxy = (DataSourceProxy) resource;
+            druidDataSourceMap.put(dataSourceProxy.getResourceId(), (DruidDataSource) dataSourceProxy.getTargetDataSource());
+        }
+    }
+
+    public DruidDataSource get(String resourceId) {
+        return druidDataSourceMap.get(resourceId);
+    }
+
+    public List<ConnectionPoolMetrics> getConnectionPoolMetricsList() {
+        List<ConnectionPoolMetrics> metricsList = new ArrayList<>();
+
+        druidDataSourceMap.forEach((resourceId, druidDataSource) -> {
+            ConnectionPoolMetrics metrics = getConnectionPoolMetrics(resourceId);
+            metricsList.add(metrics);
+        });
+
+        return metricsList;
+    }
+
+    public ConnectionPoolMetrics getConnectionPoolMetrics(String resourceId) {
+        DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);
+        if (druidDataSource == null) {
+            return null;
+        }
+
+        ConnectionPoolMetrics metrics = new ConnectionPoolMetrics();
+        metrics.setResourceId(resourceId);
+        metrics.setCurrentTimeMillis(System.currentTimeMillis());
+        metrics.setActiveCount(druidDataSource.getActiveCount());
+        metrics.setMaxActive(druidDataSource.getMaxActive());
+        metrics.setIdleCount(druidDataSource.getPoolingCount());
+
+        return metrics;
+    }
+
+    public void adjustPoolConfig(String resourceId, int maxActive, int minIdle) {
+        DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);
+        if (druidDataSource == null) {
+            return;
+        }
+        if (maxActive >= 0) {
+            druidDataSource.setMaxActive(maxActive);
+        } else if (minIdle >= 0) {
+            druidDataSource.setMinIdle(minIdle);
+        }
+    }
+
+    public boolean checkPoolIsHealthy(String resourceId) {
+        DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);
+        if (druidDataSource == null) {
+            return false;
+        }
+        try (Connection conn = druidDataSource.getConnection()) {
+            try (PreparedStatement stmt = conn.prepareStatement("SELECT 1")) {
+                return stmt.execute();
+            }
+        } catch (SQLException e) {
+            return false;
+        }
+    }
+
+    public void logPoolStatus(String resourceId) {
+        DruidDataSource druidDataSource = druidDataSourceMap.get(resourceId);
+
+        LOGGER.info("{}-Active Connections: {}", resourceId, druidDataSource.getActiveCount());
+        LOGGER.info("{}-Idle Connections: {}", resourceId, druidDataSource.getPoolingCount());
+        if (druidDataSource.getActiveCount() > druidDataSource.getMaxActive() * 0.8) {
+            LOGGER.warn("{}-Warning: Active connections exceed 80% of max allowed connections!", resourceId);
+        }
+    }
+
+}

rm-datasource/src/main/java/org/apache/seata/rm/datasource/entity/ConnectionPoolMetrics.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.seata.rm.datasource.entity;
+
+
+public class ConnectionPoolMetrics {
+
+    private String resourceId;
+
+    private Long currentTimeMillis;
+
+    private Integer activeCount;
+
+    private Integer idleCount;
+
+    private Integer maxActive;
+
+    public ConnectionPoolMetrics(){
+
+    }
+
+    public String getResourceId() {
+        return resourceId;
+    }
+
+    public void setResourceId(String resourceId) {
+        this.resourceId = resourceId;
+    }
+
+    public Integer getActiveCount() {
+        return activeCount;
+    }
+
+    public void setActiveCount(Integer activeCount) {
+        this.activeCount = activeCount;
+    }
+
+    public Integer getIdleCount() {
+        return idleCount;
+    }
+
+    public void setIdleCount(Integer idleCount) {
+        this.idleCount = idleCount;
+    }
+
+    public Integer getMaxActive() {
+        return maxActive;
+    }
+
+    public void setMaxActive(Integer maxActive) {
+        this.maxActive = maxActive;
+    }
+
+    public Long getCurrentTimeMillis() {
+        return currentTimeMillis;
+    }
+
+    public void setCurrentTimeMillis(Long currentTimeMillis) {
+        this.currentTimeMillis = currentTimeMillis;
+    }
+}