[lake] Introduce lake lag metrics for datalake enabled table (#1729)

Author: caozhen1937

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -133,6 +133,8 @@ public class MetricNames {
     public static final String LOG_NUM_SEGMENTS = "numSegments";
     public static final String LOG_END_OFFSET = "endOffset";
     public static final String REMOTE_LOG_SIZE = "size";
+    public static final String LOG_LAKE_PENDING_RECORDS = "pendingRecords";
+    public static final String LOG_LAKE_TIMESTAMP_LAG = "timestampLag";
 
     // for logic storage
     public static final String LOCAL_STORAGE_LOG_SIZE = "logSize";

fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java

@@ -201,6 +201,8 @@ public final class Replica {
     private Counter isrExpands;
     private Counter failedIsrUpdates;
 
+    private MetricGroup lakeTieringMetricGroup;
+
     public Replica(
             PhysicalTablePath physicalPath,
             TableBucket tableBucket,
@@ -536,6 +538,10 @@ public LogOffsetSnapshot fetchOffsetSnapshot(boolean fetchOnlyFromLeader) throws
     private void onBecomeNewLeader() {
         updateLeaderEndOffsetSnapshot();
 
+        if (isDataLakeEnabled()) {
+            registerLakeTieringMetrics();
+        }
+
         if (isKvTable()) {
             // if it's become new leader, we must
             // first destroy the old kv tablet
@@ -546,11 +552,30 @@ private void onBecomeNewLeader() {
         }
     }
 
+    private void registerLakeTieringMetrics() {
+        lakeTieringMetricGroup = bucketMetricGroup.addGroup("lakeTiering");
+        lakeTieringMetricGroup.gauge(
+                MetricNames.LOG_LAKE_PENDING_RECORDS,
+                () ->
+                        getLakeLogEndOffset() < 0L
+                                ? -1
+                                : getLogHighWatermark() - getLakeLogEndOffset());
+        lakeTieringMetricGroup.gauge(
+                MetricNames.LOG_LAKE_TIMESTAMP_LAG,
+                () ->
+                        logTablet.getLakeMaxTimestamp() < 0L
+                                ? -1
+                                : logTablet.localMaxTimestamp() - logTablet.getLakeMaxTimestamp());
+    }
+
     private void onBecomeNewFollower() {
         if (isKvTable()) {
             // it should be from leader to follower, we need to destroy the kv tablet
             dropKv();
         }
+        if (lakeTieringMetricGroup != null) {
+            lakeTieringMetricGroup.close();
+        }
     }
 
     @VisibleForTesting

website/docs/maintenance/observability/monitor-metrics.md

@@ -762,6 +762,17 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
       <td>endOffset</td>
       <td>The end offset in local storage for this table bucket.</td>
       <td>Gauge</td>
+    </tr>
+     <tr>
+      <td rowspan="2">table_bucket_lakeTiering</td>
+      <td>pendingRecords</td>
+      <td>The number of records lag between local log and remote log for this table bucket.</td>
+      <td>Gauge</td>
+    </tr>
+     <tr>
+      <td>timestampLag</td>
+      <td>The timestamp lag between local log and remote log for this table bucket in milliseconds.</td>
+      <td>Gauge</td>
     </tr>
     <tr>
       <td rowspan="3">table_bucket_remoteLog</td>