feature: add SqlMonitor and SlowSqlEntry for SQL execution monitoring

common/src/main/java/org/apache/seata/common/monitor/SlowSqlEntry.java

@@ -0,0 +1,52 @@
+/*
+ * 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.common.monitor;
+
+import java.time.Instant;
+
+public class SlowSqlEntry {
+
+    private final String sql;
+    private final long executionTimeMillis;
+    private final Instant timestamp;
+
+    public SlowSqlEntry(String sql, long executionTimeMillis, Instant timestamp) {
+        this.sql = sql;
+        this.executionTimeMillis = executionTimeMillis;
+        this.timestamp = timestamp;
+    }
+
+    public String getSql() {
+        return sql;
+    }
+
+    public long getExecutionTimeMillis() {
+        return executionTimeMillis;
+    }
+
+    public Instant getTimestamp() {
+        return timestamp;
+    }
+
+    @Override
+    public String toString() {
+        return "SlowSqlEntry{" + "sql='"
+                + sql + '\'' + ", executionTimeMillis="
+                + executionTimeMillis + ", timestamp="
+                + timestamp + '}';
+    }
+}

common/src/main/java/org/apache/seata/common/monitor/SqlMonitor.java

@@ -0,0 +1,136 @@
+/*
+ * 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.common.monitor;
+
+import java.time.Instant;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class SqlMonitor {
+
+    private static final SqlMonitor INSTANCE = new SqlMonitor();
+    private volatile long slowThreshold = 1000;
+    private volatile int maxSlowEntries = 50;
+    private final Deque<SlowSqlEntry> slowSqlQueue = new ConcurrentLinkedDeque<>();
+    private final Lock slowLock = new ReentrantLock();
+    private final Map<String, Integer> txnHistogram = new ConcurrentHashMap<>();
+    private final Map<String, Integer> holdHistogram = new ConcurrentHashMap<>();
+
+    private SqlMonitor() {}
+
+    public static SqlMonitor getInstance() {
+        return INSTANCE;
+    }
+
+    public void record(String sql, long execMs, long holdMs) {
+        // slow sql
+        if (execMs > slowThreshold) {
+            slowLock.lock();
+            try {
+                slowSqlQueue.addLast(new SlowSqlEntry(sql, execMs, Instant.now()));
+                if (slowSqlQueue.size() > maxSlowEntries) {
+                    slowSqlQueue.removeFirst();
+                }
+            } finally {
+                slowLock.unlock();
+            }
+        }
+
+        // transaction histogram
+        String bin = chooseTxnBucket(execMs);
+        txnHistogram.merge(bin, 1, Integer::sum);
+
+        // connection hold time histogram
+        String bin2 = chooseHoldBucket(holdMs);
+        holdHistogram.merge(bin2, 1, Integer::sum);
+    }
+
+    /**
+     * Set the execution time threshold for slow SQL.
+     * Intended for use by dynamic configuration (e.g. Nacos or application.yml binding).
+     */
+    public void setSlowThreshold(long threshold) {
+        this.slowThreshold = threshold;
+    }
+
+    /**
+     * Set the max entries for slow SQL
+     * Intended for use by dunamic configuration (e.g. Nacos or application.yml binding)
+     */
+    public void setMaxSlowEntries(int maxEntries) {
+        this.maxSlowEntries = maxEntries;
+    }
+
+    public List<SlowSqlEntry> getSlowSqlList() {
+        return new ArrayList<>(slowSqlQueue);
+    }
+
+    public Map<String, Integer> getTxnHistogram() {
+        return new LinkedHashMap<>(txnHistogram);
+    }
+
+    public Map<String, Integer> getHoldHistogram() {
+        return new LinkedHashMap<>(holdHistogram);
+    }
+
+    private String chooseTxnBucket(long ms) {
+        if (ms <= 50) {
+            return "0-50ms";
+        } else if (ms <= 200) {
+            return "50-200ms";
+        } else if (ms <= 500) {
+            return "200-500ms";
+        } else if (ms <= 1000) {
+            return "500ms-1s";
+        } else if (ms <= 3000) {
+            return "1s-3s";
+        } else {
+            return "3s+";
+        }
+    }
+
+    private String chooseHoldBucket(long ms) {
+        if (ms <= 50) {
+            return "0-50ms";
+        } else if (ms <= 200) {
+            return "50-200ms";
+        } else if (ms <= 500) {
+            return "200-500ms";
+        } else if (ms <= 1000) {
+            return "500ms-1s";
+        } else {
+            return "1s+";
+        }
+    }
+
+    /**
+     * Reset all internal states, only for testing purpose.
+     */
+    public void resetForTest() {
+        slowLock.lock();
+        try {
+            slowSqlQueue.clear();
+        } finally {
+            slowLock.unlock();
+        }
+        txnHistogram.clear();
+        holdHistogram.clear();
+    }
+}

common/src/test/java/org/apache/seata/common/monitor/SqlMonitorTest.java

@@ -0,0 +1,119 @@
+/*
+ * 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.common.monitor;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class SqlMonitorTest {
+    private SqlMonitor monitor;
+
+    @BeforeEach
+    public void setUp() {
+        monitor = SqlMonitor.getInstance();
+        monitor.resetForTest();
+    }
+
+    @Test
+    public void testRecordSlowSqlEntry() {
+        monitor.record("SELECT * FROM users", 1500, 100);
+        List<SlowSqlEntry> slowSqlList = monitor.getSlowSqlList();
+        assertEquals(1, slowSqlList.size());
+        SlowSqlEntry entry = slowSqlList.get(0);
+        assertEquals("SELECT * FROM users", entry.getSql());
+        assertTrue(entry.getExecutionTimeMillis() >= 1500);
+    }
+
+    @Test
+    public void testRecordMultiSlowEntry() {
+        monitor.record("SELECT * FROM student", 1200, 100);
+        monitor.record("SELECT * FROM school", 1300, 100);
+        List<SlowSqlEntry> slowSqlList = monitor.getSlowSqlList();
+        assertEquals(2, slowSqlList.size());
+        SlowSqlEntry entry1 = slowSqlList.get(0);
+        SlowSqlEntry entry2 = slowSqlList.get(1);
+        assertEquals("SELECT * FROM student", entry1.getSql());
+        assertEquals("SELECT * FROM school", entry2.getSql());
+    }
+
+    @Test
+    public void testMaxSlowSqlQueueSize() {
+        // maxSlowEntries = 50 by default
+        for (int i = 0; i < 55; i++) {
+            monitor.record("SELECT * FROM orders WHERE id = " + i, 1500, 200);
+        }
+
+        List<SlowSqlEntry> slowSqlList = monitor.getSlowSqlList();
+        assertEquals(50, slowSqlList.size());
+
+        // Should not contain the first 5 entries
+        for (int i = 0; i < 5; i++) {
+            int finalI = i;
+            assertFalse(
+                    slowSqlList.stream()
+                            .map(SlowSqlEntry::getSql)
+                            .anyMatch(sql -> sql.equals("SELECT * FROM orders WHERE id = " + finalI)),
+                    "Entry with id = " + finalI + " should have been evicted");
+        }
+    }
+
+    @Test
+    public void testRecordForFastSql() {
+        monitor.record("SELECT 1", 100, 50);
+        List<SlowSqlEntry> slowSqlList = monitor.getSlowSqlList();
+        assertTrue(slowSqlList.isEmpty());
+    }
+
+    @Test
+    public void testTxnHistogramBuckets() {
+        monitor.record("SELECT * FROM t1", 30, 0);
+        monitor.record("SELECT * FROM t2", 150, 0);
+        monitor.record("SELECT * FROM t3", 300, 0);
+        monitor.record("SELECT * FROM t4", 800, 0);
+        monitor.record("SELECT * FROM t5", 2000, 0);
+        monitor.record("SELECT * FROM t6", 4000, 0);
+
+        Map<String, Integer> histogram = monitor.getTxnHistogram();
+        assertEquals(1, histogram.get("0-50ms"));
+        assertEquals(1, histogram.get("50-200ms"));
+        assertEquals(1, histogram.get("200-500ms"));
+        assertEquals(1, histogram.get("500ms-1s"));
+        assertEquals(1, histogram.get("1s-3s"));
+        assertEquals(1, histogram.get("3s+"));
+    }
+
+    @Test
+    public void testHoldHistogramBuckets() {
+        monitor.record("SELECT * FROM hold1", 0, 30);
+        monitor.record("SELECT * FROM hold2", 0, 150);
+        monitor.record("SELECT * FROM hold3", 0, 300);
+        monitor.record("SELECT * FROM hold4", 0, 800);
+        monitor.record("SELECT * FROM hold5", 0, 2000);
+
+        Map<String, Integer> histogram = monitor.getHoldHistogram();
+        assertEquals(1, histogram.get("0-50ms"));
+        assertEquals(1, histogram.get("50-200ms"));
+        assertEquals(1, histogram.get("200-500ms"));
+        assertEquals(1, histogram.get("500ms-1s"));
+        assertEquals(1, histogram.get("1s+"));
+    }
+}