feat: add observability metrics for undo log manager

rm-datasource/pom.xml

@@ -168,5 +168,10 @@
             <artifactId>json-common</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>seata-metrics-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>

rm-datasource/src/main/java/org/apache/seata/rm/datasource/undo/AbstractUndoLogManager.java

@@ -28,6 +28,8 @@
 import org.apache.seata.core.exception.BranchTransactionException;
 import org.apache.seata.core.exception.TransactionException;
 import org.apache.seata.core.rpc.processor.Pair;
+import org.apache.seata.metrics.registry.Registry;
+import org.apache.seata.metrics.registry.RegistryFactory;
 import org.apache.seata.rm.datasource.ConnectionContext;
 import org.apache.seata.rm.datasource.ConnectionProxy;
 import org.apache.seata.rm.datasource.DataSourceProxy;
@@ -47,6 +49,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.seata.common.DefaultValues.DEFAULT_CLIENT_UNDO_COMPRESS_ENABLE;
 import static org.apache.seata.common.DefaultValues.DEFAULT_CLIENT_UNDO_COMPRESS_THRESHOLD;
@@ -130,6 +133,7 @@ public static void removeCurrentSerializer() {
      */
     @Override
     public void deleteUndoLog(String xid, long branchId, Connection conn) throws SQLException {
+        long start = System.nanoTime();
         try (PreparedStatement deletePST = conn.prepareStatement(DELETE_UNDO_LOG_SQL);
                 PreparedStatement deleteSubPST = conn.prepareStatement(DELETE_SUB_UNDO_LOG_SQL)) {
             deletePST.setLong(1, branchId);
@@ -144,6 +148,14 @@ public void deleteUndoLog(String xid, long branchId, Connection conn) throws SQL
                 e = new SQLException(e);
             }
             throw (SQLException) e;
+        } finally {
+            Registry registry = getRegistry();
+            if (registry != null) {
+                registry.getTimer(UndoLogConstants.TIMER_UNDO_LOG_DELETE_LATENCY)
+                        .record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
+                registry.getCounter(UndoLogConstants.COUNTER_UNDO_LOG_DELETE_COUNT)
+                        .increase(1);
+            }
         }
     }
 
@@ -159,6 +171,8 @@ public void batchDeleteUndoLog(Set<String> xids, Set<Long> branchIds, Connection
         if (CollectionUtils.isEmpty(xids) || CollectionUtils.isEmpty(branchIds)) {
             return;
         }
+        long start = System.nanoTime();
+        int totalDeleteRows = 0;
         int xidSize = xids.size();
         int branchIdSize = branchIds.size();
         String batchDeleteSql = toBatchDeleteUndoLogSql(xidSize, branchIdSize);
@@ -178,6 +192,7 @@ public void batchDeleteUndoLog(Set<String> xids, Set<Long> branchIds, Connection
                 paramsIndex++;
             }
             int deleteRows = deletePST.executeUpdate();
+            totalDeleteRows += deleteRows;
             if (LOGGER.isDebugEnabled()) {
                 LOGGER.debug("batch delete undo log size {}", deleteRows);
             }
@@ -190,6 +205,16 @@ public void batchDeleteUndoLog(Set<String> xids, Set<Long> branchIds, Connection
                 e = new SQLException(e);
             }
             throw (SQLException) e;
+        } finally {
+            Registry registry = getRegistry();
+            if (registry != null) {
+                registry.getTimer(UndoLogConstants.TIMER_UNDO_LOG_DELETE_LATENCY)
+                        .record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
+                if (totalDeleteRows > 0) {
+                    registry.getCounter(UndoLogConstants.COUNTER_UNDO_LOG_DELETE_COUNT)
+                            .increase(totalDeleteRows);
+                }
+            }
         }
     }
 
@@ -300,6 +325,10 @@ public void flushUndoLogs(ConnectionProxy cp) throws SQLException {
         String rollbackCtx =
                 buildContext(parser.getName(), compressorType, UndoLogConstants.MAX_ALLOWED_PACKET, maxAllowedPacket);
         insertUndoLogWithNormal(xid, branchId, rollbackCtx, undoLogContent, cp.getTargetConnection());
+        Registry registry = getRegistry();
+        if (registry != null) {
+            registry.getSummary(UndoLogConstants.SUMMARY_UNDO_LOG_SIZE).increase(undoLogContent.length);
+        }
     }
 
     /**
@@ -586,4 +615,19 @@ public boolean hasUndoLogTable(Connection conn) {
     protected String getCheckUndoLogTableExistSql() {
         return CHECK_UNDO_LOG_TABLE_EXIST_SQL;
     }
+
+    /**
+     * Get the metrics registry instance.
+     * This method is protected to allow testing with Mockito spy.
+     *
+     * @return the Registry instance, or null if metrics are not enabled
+     */
+    protected Registry getRegistry() {
+        try {
+            return RegistryFactory.getInstance();
+        } catch (Throwable t) {
+            LOGGER.warn("Failed to get metrics registry: {}", t.getMessage());
+            return null;
+        }
+    }
 }

rm-datasource/src/main/java/org/apache/seata/rm/datasource/undo/UndoLogConstants.java

@@ -18,6 +18,8 @@
 
 import org.apache.seata.config.ConfigurationFactory;
 import org.apache.seata.core.constants.ConfigurationKeys;
+import org.apache.seata.metrics.Id;
+import org.apache.seata.metrics.IdConstants;
 
 import static org.apache.seata.common.DefaultValues.DEFAULT_TRANSACTION_UNDO_LOG_SERIALIZATION;
 
@@ -38,4 +40,19 @@ public interface UndoLogConstants {
     String SUB_SPLIT_KEY = ",";
 
     String MAX_ALLOWED_PACKET = "map";
+
+    Id SUMMARY_UNDO_LOG_SIZE = new Id("seata.undo.log")
+            .withTag(IdConstants.ROLE_KEY, IdConstants.ROLE_VALUE_RM)
+            .withTag(IdConstants.METER_KEY, IdConstants.METER_VALUE_SUMMARY)
+            .withTag("type", "size");
+
+    Id TIMER_UNDO_LOG_DELETE_LATENCY = new Id("seata.undo.log")
+            .withTag(IdConstants.ROLE_KEY, IdConstants.ROLE_VALUE_RM)
+            .withTag(IdConstants.METER_KEY, IdConstants.METER_VALUE_TIMER)
+            .withTag("type", "delete_latency");
+
+    Id COUNTER_UNDO_LOG_DELETE_COUNT = new Id("seata.undo.log")
+            .withTag(IdConstants.ROLE_KEY, IdConstants.ROLE_VALUE_RM)
+            .withTag(IdConstants.METER_KEY, IdConstants.METER_VALUE_COUNTER)
+            .withTag("type", "delete_count");
 }

rm-datasource/src/test/java/org/apache/seata/rm/datasource/undo/UndoLogMetricsTest.java

@@ -0,0 +1,100 @@
+/*
+ * 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.undo;
+
+import org.apache.seata.metrics.Id;
+import org.apache.seata.metrics.registry.Registry;
+import org.apache.seata.rm.datasource.ConnectionContext;
+import org.apache.seata.rm.datasource.ConnectionProxy;
+import org.apache.seata.rm.datasource.DataSourceProxy;
+import org.apache.seata.rm.datasource.undo.mysql.MySQLUndoLogManager;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class UndoLogMetricsTest {
+
+    @Test
+    public void testFlushUndoLogsMetrics() throws Exception {
+        Registry registry = mock(Registry.class);
+        org.apache.seata.metrics.Summary summary = mock(org.apache.seata.metrics.Summary.class);
+        when(registry.getSummary(any(Id.class))).thenReturn(summary);
+
+        ConnectionProxy connectionProxy = mock(ConnectionProxy.class);
+        ConnectionContext context = mock(ConnectionContext.class);
+        DataSourceProxy dataSourceProxy = mock(DataSourceProxy.class);
+
+        when(connectionProxy.getContext()).thenReturn(context);
+        when(connectionProxy.getDataSourceProxy()).thenReturn(dataSourceProxy);
+        Connection connection = mock(Connection.class);
+        PreparedStatement preparedStatement = mock(PreparedStatement.class);
+        when(connection.prepareStatement(any(String.class))).thenReturn(preparedStatement);
+        when(connectionProxy.getTargetConnection()).thenReturn(connection);
+        when(dataSourceProxy.getResourceId()).thenReturn("jdbc:mysql://localhost:3306/test");
+
+        when(context.hasUndoLog()).thenReturn(true);
+        when(context.getXid()).thenReturn("xid");
+        when(context.getBranchId()).thenReturn(123456L);
+        when(context.getUndoItems()).thenReturn(Collections.emptyList());
+
+        // Spy on the manager to mock getRegistry()
+        MySQLUndoLogManager manager = spy(new MySQLUndoLogManager());
+        doReturn(registry).when(manager).getRegistry();
+
+        manager.flushUndoLogs(connectionProxy);
+
+        verify(registry).getSummary(eq(UndoLogConstants.SUMMARY_UNDO_LOG_SIZE));
+        verify(summary).increase(anyLong());
+    }
+
+    @Test
+    public void testDeleteUndoLogMetrics() throws Exception {
+        Registry registry = mock(Registry.class);
+        org.apache.seata.metrics.Timer timer = mock(org.apache.seata.metrics.Timer.class);
+        org.apache.seata.metrics.Counter counter = mock(org.apache.seata.metrics.Counter.class);
+
+        when(registry.getTimer(any(Id.class))).thenReturn(timer);
+        when(registry.getCounter(any(Id.class))).thenReturn(counter);
+
+        Connection connection = mock(Connection.class);
+        PreparedStatement preparedStatement = mock(PreparedStatement.class);
+        when(connection.prepareStatement(any(String.class))).thenReturn(preparedStatement);
+
+        // Spy on the manager to mock getRegistry()
+        MySQLUndoLogManager manager = spy(new MySQLUndoLogManager());
+        doReturn(registry).when(manager).getRegistry();
+
+        manager.deleteUndoLog("xid", 123L, connection);
+
+        verify(registry).getTimer(eq(UndoLogConstants.TIMER_UNDO_LOG_DELETE_LATENCY));
+        verify(timer).record(anyLong(), any(TimeUnit.class));
+        verify(registry).getCounter(eq(UndoLogConstants.COUNTER_UNDO_LOG_DELETE_COUNT));
+        verify(counter).increase(1);
+    }
+}