[coordinator] Add rebalance progress and outcome metrics

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

@@ -47,6 +47,16 @@ public class MetricNames {
     public static final String KV_LEADER_REPLICA_CAPACITY = "kvLeaderReplicaCapacity";
     public static final String REPLICAS_TO_DELETE_COUNT = "replicasToDeleteCount";
     public static final String PENDING_LEADER_ACTIVATION_COUNT = "pendingLeaderActivationCount";
+    public static final String REBALANCE_IN_PROGRESS = "rebalanceInProgress";
+    public static final String REBALANCE_BUCKETS_PENDING = "rebalanceBucketsPending";
+    public static final String REBALANCE_BUCKETS_COMPLETED = "rebalanceBucketsCompleted";
+    public static final String REBALANCE_BUCKETS_FAILED = "rebalanceBucketsFailed";
+    public static final String REBALANCE_BUCKETS_TIMED_OUT = "rebalanceBucketsTimedOut";
+    public static final String REBALANCE_DURATION_MS = "rebalanceDurationMs";
+    public static final String INFLIGHT_BUCKET_DURATION_MS = "inflightBucketDurationMs";
+    public static final String REBALANCES_COMPLETED_TOTAL = "rebalancesCompletedTotal";
+    public static final String REBALANCES_FAILED_TOTAL = "rebalancesFailedTotal";
+    public static final String REBALANCES_CANCELED_TOTAL = "rebalancesCanceledTotal";
     // for coordinator sender (per-tablet-server control request sender threads)
     public static final String SENDER_QUEUE_SIZE = "senderQueueSize";
     public static final String SENDER_QUEUE_TIME_MS = "senderQueueTimeMs";

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -277,7 +277,11 @@ public CoordinatorEventProcessor(
         this.internalListenerName = conf.getString(ConfigOptions.INTERNAL_LISTENER_NAME);
         this.rebalanceManager =
                 new RebalanceManager(
-                        this, zooKeeperClient, coordinatorEventManager, SystemClock.getInstance());
+                        this,
+                        zooKeeperClient,
+                        coordinatorEventManager,
+                        SystemClock.getInstance(),
+                        coordinatorMetricGroup.getRebalanceMetrics());
         this.offlineLeaderRetryDelayMs =
                 conf.get(ConfigOptions.COORDINATOR_OFFLINE_LEADER_RETRY_DELAY).toMillis();
         if (offlineLeaderRetryDelayMs <= 0) {
@@ -342,10 +346,13 @@ public void startup() {
     }
 
     public void shutdown() {
-        clearOfflineLeaderRetryTask();
-        // close the event manager
-        coordinatorEventManager.close();
-        rebalanceManager.close();
+        try {
+            clearOfflineLeaderRetryTask();
+            // close the event manager
+            coordinatorEventManager.close();
+        } finally {
+            rebalanceManager.close();
+        }
         onShutdown();
         coordinatorContext.resetContext();
         updateObservedKvLeaderReplicaCount();

fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java

@@ -49,6 +49,7 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayDeque;
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -62,12 +63,15 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static org.apache.fluss.cluster.rebalance.RebalanceStatus.CANCELED;
 import static org.apache.fluss.cluster.rebalance.RebalanceStatus.COMPLETED;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FAILED;
 import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FINAL_STATUSES;
 import static org.apache.fluss.cluster.rebalance.RebalanceStatus.NOT_STARTED;
 import static org.apache.fluss.cluster.rebalance.RebalanceStatus.REBALANCING;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.TIMEOUT;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 
@@ -90,6 +94,7 @@ public class RebalanceManager {
     private final EventManager eventManager;
     private final Clock clock;
     private final ScheduledExecutorService timeoutChecker;
+    private final RebalanceMetrics metrics;
 
     /** A queue of in progress table bucket to rebalance. */
     private final Queue<TableBucket> inProgressRebalanceTasksQueue = new ArrayDeque<>();
@@ -102,8 +107,11 @@ public class RebalanceManager {
     private final Map<TableBucket, RebalanceResultForBucket> finishedRebalanceTasks =
             new ConcurrentHashMap<>();
 
+    private final Map<RebalanceStatus, AtomicLong> finishedBucketCounts = newFinishedBucketCounts();
+
     private final GoalOptimizer goalOptimizer;
     private volatile long registerTime;
+    private volatile boolean bucketResultsAvailable;
     private volatile @Nullable RebalanceStatus rebalanceStatus;
     private volatile @Nullable String currentRebalanceId;
     private volatile boolean isClosed = false;
@@ -126,12 +134,14 @@ public RebalanceManager(
             CoordinatorEventProcessor eventProcessor,
             ZooKeeperClient zkClient,
             EventManager eventManager,
-            Clock clock) {
+            Clock clock,
+            RebalanceMetrics metrics) {
         this(
                 eventProcessor,
                 zkClient,
                 eventManager,
                 clock,
+                metrics,
                 // TODO: Reuse the CoordinatorServer shared scheduler for this lightweight
                 // coordinator timeout checker instead of creating a component-owned scheduler.
                 Executors.newScheduledThreadPool(
@@ -144,17 +154,51 @@ public RebalanceManager(
             ZooKeeperClient zkClient,
             EventManager eventManager,
             Clock clock,
+            RebalanceMetrics metrics,
             ScheduledExecutorService timeoutChecker) {
         this.eventProcessor = eventProcessor;
         this.zkClient = zkClient;
         this.eventManager = eventManager;
         this.clock = clock == null ? SystemClock.getInstance() : clock;
         this.timeoutChecker = timeoutChecker;
         this.goalOptimizer = new GoalOptimizer();
+        this.metrics = checkNotNull(metrics, "metrics");
+    }
+
+    long rebalanceInProgress() {
+        return rebalanceStatus == REBALANCING ? 1L : 0L;
+    }
+
+    long pendingBucketCount() {
+        return inProgressRebalanceTasks.size();
+    }
+
+    long rebalanceDurationMs() {
+        return rebalanceStatus == REBALANCING
+                ? Math.max(0L, clock.milliseconds() - registerTime)
+                : 0L;
+    }
+
+    long finishedBucketCount(RebalanceStatus status) {
+        if (!bucketResultsAvailable) {
+            return 0L;
+        }
+        return finishedBucketCounts.get(status).get();
+    }
+
+    private boolean hasFinishedBucket(RebalanceStatus status) {
+        return finishedBucketCounts.get(status).get() > 0;
+    }
+
+    long inflightBucketDurationMs() {
+        TableBucket bucket = inflightTaskBucket;
+        long startMs = inflightTaskStartMs;
+        return bucket == null || startMs < 0 ? 0L : Math.max(0L, clock.milliseconds() - startMs);
     }
 
     public void startup() {
         LOG.info("Start up rebalance manager.");
+        metrics.bind(this);
         initialize();
     }
 
@@ -194,11 +238,13 @@ public void registerRebalance(
             Map<TableBucket, RebalancePlanForBucket> rebalancePlan,
             RebalanceStatus newStatus) {
         checkNotClosed();
-        registerTime = System.currentTimeMillis();
+        registerTime = clock.milliseconds();
         // first clear all exists tasks.
         inProgressRebalanceTasks.clear();
         inProgressRebalanceTasksQueue.clear();
         finishedRebalanceTasks.clear();
+        finishedBucketCounts.values().forEach(count -> count.set(0L));
+        bucketResultsAvailable = !FINAL_STATUSES.contains(newStatus);
         // Clear gate (bucket) first, then data (startMs).
         inflightTaskBucket = null;
         inflightTaskStartMs = -1;
@@ -240,6 +286,7 @@ public void finishRebalanceTask(TableBucket tableBucket, RebalanceStatus statusF
             finishedRebalanceTasks.put(
                     tableBucket,
                     RebalanceResultForBucket.of(resultForBucket.plan(), statusForBucket));
+            finishedBucketCounts.get(statusForBucket).incrementAndGet();
             // Clear gate (bucket) first, then data (startMs).
             inflightTaskBucket = null;
             inflightTaskStartMs = -1;
@@ -320,6 +367,9 @@ public void cancelRebalance(@Nullable String rebalanceId) {
             LOG.error("Error when delete rebalance plan from zookeeper.", e);
         }
 
+        if (rebalanceStatus == REBALANCING) {
+            metrics.incRebalancesCanceled();
+        }
         rebalanceStatus = CANCELED;
         inProgressRebalanceTasksQueue.clear();
         inProgressRebalanceTasks.clear();
@@ -343,20 +393,20 @@ public RebalanceTask generateRebalanceTask(List<Goal> goalsByPriority) {
         String rebalanceId = UUID.randomUUID().toString();
         try {
             // Generate the latest cluster model.
-            long startTime = System.currentTimeMillis();
+            long startTime = clock.milliseconds();
             ClusterModel clusterModel = buildClusterModel(eventProcessor.getCoordinatorContext());
             LOG.info(
                     "Build cluster model for rebalance id {} with {} ms.",
                     rebalanceId,
-                    System.currentTimeMillis() - startTime);
+                    clock.milliseconds() - startTime);
 
             // do optimize.
-            startTime = System.currentTimeMillis();
+            startTime = clock.milliseconds();
             rebalancePlanForBuckets = goalOptimizer.doOptimizeOnce(clusterModel, goalsByPriority);
             LOG.info(
                     "Do optimize for rebalance id {} with {} ms.",
                     rebalanceId,
-                    System.currentTimeMillis() - startTime);
+                    clock.milliseconds() - startTime);
         } catch (Exception e) {
             LOG.error("Failed to generate rebalance plan.", e);
             throw e;
@@ -407,14 +457,21 @@ private void completeRebalance() {
             LOG.error("Error when update rebalance plan from zookeeper.", e);
         }
 
+        if (bucketResultsAvailable) {
+            if (hasFinishedBucket(FAILED) || hasFinishedBucket(TIMEOUT)) {
+                metrics.incRebalancesFailed();
+            } else {
+                metrics.incRebalancesCompleted();
+            }
+        }
         rebalanceStatus = COMPLETED;
         inProgressRebalanceTasks.clear();
         inProgressRebalanceTasksQueue.clear();
 
         // Here, it will not clear finishedRebalanceTasks, because it will be used by
         // listRebalanceProgress. It will be cleared when next register.
 
-        LOG.info("Rebalance complete with {} ms.", System.currentTimeMillis() - registerTime);
+        LOG.info("Rebalance complete with {} ms.", clock.milliseconds() - registerTime);
     }
 
     private ClusterModel buildClusterModel(CoordinatorContext coordinatorContext) {
@@ -470,6 +527,14 @@ private RebalanceTask buildRebalanceTask(
         return new RebalanceTask(rebalanceId, NOT_STARTED, bucketPlan);
     }
 
+    private static Map<RebalanceStatus, AtomicLong> newFinishedBucketCounts() {
+        Map<RebalanceStatus, AtomicLong> counts = new EnumMap<>(RebalanceStatus.class);
+        for (RebalanceStatus status : RebalanceStatus.values()) {
+            counts.put(status, new AtomicLong());
+        }
+        return counts;
+    }
+
     private boolean isOfflineTagged(ServerTag serverTag) {
         return serverTag == ServerTag.PERMANENT_OFFLINE || serverTag == ServerTag.TEMPORARY_OFFLINE;
     }
@@ -518,6 +583,7 @@ private void checkNotClosed() {
 
     public void close() {
         isClosed = true;
+        metrics.unbind(this);
         timeoutChecker.shutdownNow();
     }
 

fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceMetrics.java

@@ -0,0 +1,104 @@
+/*
+ * 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.fluss.server.coordinator.rebalance;
+
+import org.apache.fluss.metrics.Counter;
+import org.apache.fluss.metrics.MetricNames;
+import org.apache.fluss.metrics.ThreadSafeSimpleCounter;
+import org.apache.fluss.metrics.groups.MetricGroup;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.function.ToLongFunction;
+
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.COMPLETED;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.FAILED;
+import static org.apache.fluss.cluster.rebalance.RebalanceStatus.TIMEOUT;
+
+/**
+ * Rebalance metrics registered for the lifetime of a coordinator server.
+ *
+ * <p>The coordinator event thread binds and unbinds the current {@link RebalanceManager} and
+ * updates counters, while metric reporter threads may concurrently read gauges via {@link
+ * #readCurrent(ToLongFunction)}.
+ */
+@ThreadSafe
+public class RebalanceMetrics {
+
+    private final Counter rebalancesCompleted = new ThreadSafeSimpleCounter();
+    private final Counter rebalancesFailed = new ThreadSafeSimpleCounter();
+    private final Counter rebalancesCanceled = new ThreadSafeSimpleCounter();
+
+    private volatile @Nullable RebalanceManager current;
+
+    /** Registers the rebalance metrics on the coordinator metric group. */
+    public RebalanceMetrics(MetricGroup metricGroup) {
+        metricGroup.counter(MetricNames.REBALANCES_COMPLETED_TOTAL, rebalancesCompleted);
+        metricGroup.counter(MetricNames.REBALANCES_FAILED_TOTAL, rebalancesFailed);
+        metricGroup.counter(MetricNames.REBALANCES_CANCELED_TOTAL, rebalancesCanceled);
+        metricGroup.gauge(
+                MetricNames.REBALANCE_IN_PROGRESS,
+                () -> readCurrent(RebalanceManager::rebalanceInProgress));
+        metricGroup.gauge(
+                MetricNames.REBALANCE_BUCKETS_PENDING,
+                () -> readCurrent(RebalanceManager::pendingBucketCount));
+        metricGroup.gauge(
+                MetricNames.REBALANCE_BUCKETS_COMPLETED,
+                () -> readCurrent(manager -> manager.finishedBucketCount(COMPLETED)));
+        metricGroup.gauge(
+                MetricNames.REBALANCE_BUCKETS_FAILED,
+                () -> readCurrent(manager -> manager.finishedBucketCount(FAILED)));
+        metricGroup.gauge(
+                MetricNames.REBALANCE_BUCKETS_TIMED_OUT,
+                () -> readCurrent(manager -> manager.finishedBucketCount(TIMEOUT)));
+        metricGroup.gauge(
+                MetricNames.REBALANCE_DURATION_MS,
+                () -> readCurrent(RebalanceManager::rebalanceDurationMs));
+        metricGroup.gauge(
+                MetricNames.INFLIGHT_BUCKET_DURATION_MS,
+                () -> readCurrent(RebalanceManager::inflightBucketDurationMs));
+    }
+
+    void bind(RebalanceManager manager) {
+        current = manager;
+    }
+
+    void unbind(RebalanceManager manager) {
+        if (current == manager) {
+            current = null;
+        }
+    }
+
+    void incRebalancesCompleted() {
+        rebalancesCompleted.inc();
+    }
+
+    void incRebalancesFailed() {
+        rebalancesFailed.inc();
+    }
+
+    void incRebalancesCanceled() {
+        rebalancesCanceled.inc();
+    }
+
+    private long readCurrent(ToLongFunction<RebalanceManager> read) {
+        RebalanceManager manager = current;
+        return manager == null ? 0L : read.applyAsLong(manager);
+    }
+}