HBASE-27380 RitDuration histogram metric is broken (#7830)

Signed-off-by: Duo Zhang <zhangduo@apache.org>
Reviewed-by: Vaibhav Joshi <vjoshi@cloudera.com>

Author: liuxiaocs7

hbase-client/src/main/java/org/apache/hadoop/hbase/master/RegionState.java

@@ -183,8 +183,6 @@ public static State convert(ClusterStatusProtos.RegionState.State protoState) {
   private final RegionInfo hri;
   private final ServerName serverName;
   private final State state;
-  // The duration of region in transition
-  private long ritDuration;
 
   public static RegionState createForTesting(RegionInfo region, State state) {
     return new RegionState(region, state, EnvironmentEdgeManager.currentTime(), null);
@@ -195,16 +193,10 @@ public RegionState(RegionInfo region, State state, ServerName serverName) {
   }
 
   public RegionState(RegionInfo region, State state, long stamp, ServerName serverName) {
-    this(region, state, stamp, serverName, 0);
-  }
-
-  public RegionState(RegionInfo region, State state, long stamp, ServerName serverName,
-    long ritDuration) {
     this.hri = region;
     this.state = state;
     this.stamp = stamp;
     this.serverName = serverName;
-    this.ritDuration = ritDuration;
   }
 
   public State getState() {
@@ -223,19 +215,6 @@ public ServerName getServerName() {
     return serverName;
   }
 
-  public long getRitDuration() {
-    return ritDuration;
-  }
-
-  /**
-   * Update the duration of region in transition
-   * @param previousStamp previous RegionState's timestamp
-   */
-  @InterfaceAudience.Private
-  void updateRitDuration(long previousStamp) {
-    this.ritDuration += (this.stamp - previousStamp);
-  }
-
   public boolean isClosing() {
     return state == State.CLOSING;
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignmentManager.java

@@ -235,8 +235,7 @@ public class AssignmentManager {
 
   private final int forceRegionRetainmentRetries;
 
-  private final RegionInTransitionTracker regionInTransitionTracker =
-    new RegionInTransitionTracker();
+  private final RegionInTransitionTracker regionInTransitionTracker;
 
   public AssignmentManager(MasterServices master, MasterRegion masterRegion) {
     this(master, masterRegion, new RegionStateStore(master, masterRegion));
@@ -246,6 +245,7 @@ public AssignmentManager(MasterServices master, MasterRegion masterRegion) {
     this.master = master;
     this.regionStateStore = stateStore;
     this.metrics = new MetricsAssignmentManager();
+    this.regionInTransitionTracker = new RegionInTransitionTracker(metrics::updateRitDuration);
     this.masterRegion = masterRegion;
 
     final Configuration conf = master.getConfiguration();
@@ -1717,7 +1717,7 @@ public boolean isRegionTwiceOverThreshold(final RegionInfo regionInfo) {
       if (state == null) {
         return false;
       }
-      return (statTimestamp - state.getStamp()) > (ritThreshold * 2);
+      return (statTimestamp - state.getStamp()) > (ritThreshold * 2L);
     }
 
     protected void update(final AssignmentManager am) {
@@ -1747,7 +1747,7 @@ private void update(final Collection<RegionState> regions, final long currentTim
             ritsOverThreshold = new HashMap<String, RegionState>();
           }
           ritsOverThreshold.put(state.getRegion().getEncodedName(), state);
-          totalRITsTwiceThreshold += (ritTime > (ritThreshold * 2)) ? 1 : 0;
+          totalRITsTwiceThreshold += (ritTime > (ritThreshold * 2L)) ? 1 : 0;
         }
         if (oldestRITTime < ritTime) {
           oldestRITTime = ritTime;
@@ -1903,7 +1903,7 @@ public void visitRegionState(Result result, final RegionInfo regionInfo, final S
         if (timeOfCrash != 0) {
           regionNode.crashed(timeOfCrash);
         }
-        regionInTransitionTracker.regionCrashed(regionNode);
+        regionInTransitionTracker.regionCrashed(regionNode, timeOfCrash);
       }
     }
   };
@@ -2125,7 +2125,7 @@ public int getRegionTransitScheduledCount() {
    * Get the number of regions in transition.
    */
   public int getRegionsInTransitionCount() {
-    return regionInTransitionTracker.getRegionsInTransition().size();
+    return regionInTransitionTracker.getRegionsInTransitionCount();
   }
 
   public SortedSet<RegionState> getRegionsStateInTransition() {
@@ -2354,7 +2354,7 @@ public void markRegionsAsCrashed(List<RegionInfo> regionsOnCrashedServer,
       RegionStateNode node = regionStates.getOrCreateRegionStateNode(regionInfo);
       if (crashedServerName.equals(node.getRegionLocation())) {
         node.crashed(scp.getSubmittedTime());
-        regionInTransitionTracker.regionCrashed(node);
+        regionInTransitionTracker.regionCrashed(node, scp.getSubmittedTime());
       } else {
         LOG.warn("Region {} should be on crashed region server {} but is recorded on {}",
           regionInfo, crashedServerName, node.getRegionLocation());

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionInTransitionTracker.java

@@ -17,14 +17,19 @@
  */
 package org.apache.hadoop.hbase.master.assignment;
 
+import com.google.errorprone.annotations.RestrictedApi;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.LongConsumer;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.TableState;
 import org.apache.hadoop.hbase.master.RegionState;
 import org.apache.hadoop.hbase.master.TableStateManager;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -36,31 +41,50 @@
 public class RegionInTransitionTracker {
   private static final Logger LOG = LoggerFactory.getLogger(RegionInTransitionTracker.class);
 
-  private final List<RegionState.State> DISABLE_TABLE_REGION_STATE =
+  private static final List<RegionState.State> DISABLE_TABLE_REGION_STATE =
     List.of(RegionState.State.OFFLINE, RegionState.State.CLOSED);
 
-  private final List<RegionState.State> ENABLE_TABLE_REGION_STATE = List.of(RegionState.State.OPEN);
+  private static final List<RegionState.State> ENABLE_TABLE_REGION_STATE =
+    List.of(RegionState.State.OPEN);
 
-  // DO NOT USE containsKey()/remove() on regionInTransition with a different RegionInfo instance:
-  // this map is ordered by RegionInfo.COMPARATOR, and that comparator includes the offline flag.
-  // Lookups can therefore fail if the RegionInfo used as the key has a different offline value,
-  // even when it refers to the same region. Offline value changes with splitting.
-  private final ConcurrentSkipListMap<RegionInfo, RegionStateNode> regionInTransition =
-    new ConcurrentSkipListMap<>(RegionInfo.COMPARATOR);
+  // DO NOT USE containsKey()/remove() on regionInTransition with a RegionInfo instance whose
+  // offline flag differs from the one stored as the key: RegionInfo#equals and #hashCode both
+  // include the offline flag, so such a lookup misses even when it refers to the same region.
+  // Offline value changes with splitting.
+  private final ConcurrentHashMap<RegionInfo, Pair<RegionStateNode, Long>> regionInTransition =
+    new ConcurrentHashMap<>();
 
+  private final LongConsumer ritDurationConsumer;
   private TableStateManager tableStateManager;
 
+  public RegionInTransitionTracker(LongConsumer ritDurationConsumer) {
+    this.ritDurationConsumer = Objects.requireNonNull(ritDurationConsumer);
+  }
+
+  @RestrictedApi(explanation = "Should only be called in tests", link = "",
+      allowedOnPath = ".*/src/test/.*")
+  boolean isRegionInTransition(final RegionInfo regionInfo) {
+    return regionInTransition.containsKey(regionInfo);
+  }
+
   /**
    * Handles a region whose hosting RegionServer has crashed. When a RegionServer fails, all regions
    * it was hosting are automatically added to the RIT list since they need to be reassigned to
    * other servers.
+   * @param regionStateNode the region whose hosting server crashed
+   * @param crashTime       the RIT start time to use when the region is not already in transition
+   *                        (an existing entry keeps its earlier start). Passed explicitly rather
+   *                        than read from {@link RegionStateNode#getLastUpdate()}, which a stale
+   *                        procedure can mask. A non-positive value falls back to the node's last
+   *                        update.
    */
-  public void regionCrashed(RegionStateNode regionStateNode) {
+  public void regionCrashed(RegionStateNode regionStateNode, long crashTime) {
     if (isReplica(regionStateNode)) {
       return;
     }
 
-    if (addRegionInTransition(regionStateNode)) {
+    long startTime = crashTime > 0 ? crashTime : regionStateNode.getLastUpdate();
+    if (addRegionInTransition(regionStateNode, startTime)) {
       LOG.debug("{} added to RIT list because hosting region server is crashed ",
         regionStateNode.getRegionInfo().getEncodedName());
     }
@@ -128,11 +152,26 @@ public void handleRegionDelete(RegionInfo regionInfo) {
   }
 
   private boolean addRegionInTransition(final RegionStateNode regionStateNode) {
-    return regionInTransition.putIfAbsent(regionStateNode.getRegionInfo(), regionStateNode) == null;
+    return addRegionInTransition(regionStateNode, regionStateNode.getLastUpdate());
+  }
+
+  private boolean addRegionInTransition(final RegionStateNode regionStateNode, long startTime) {
+    if (startTime <= 0) {
+      startTime = EnvironmentEdgeManager.currentTime();
+    }
+    return regionInTransition.putIfAbsent(regionStateNode.getRegionInfo(),
+      Pair.newPair(regionStateNode, startTime)) == null;
   }
 
   private boolean removeRegionInTransition(final RegionInfo regionInfo) {
-    return regionInTransition.remove(regionInfo) != null;
+    Pair<RegionStateNode, Long> removed = regionInTransition.remove(regionInfo);
+    if (removed != null) {
+      long duration = EnvironmentEdgeManager.currentTime() - removed.getSecond();
+      if (duration >= 0) {
+        ritDurationConsumer.accept(duration);
+      }
+    }
+    return removed != null;
   }
 
   public void stop() {
@@ -143,8 +182,16 @@ public boolean hasRegionsInTransition() {
     return !regionInTransition.isEmpty();
   }
 
+  public int getRegionsInTransitionCount() {
+    return regionInTransition.size();
+  }
+
   public List<RegionStateNode> getRegionsInTransition() {
-    return new ArrayList<>(regionInTransition.values());
+    List<RegionStateNode> regions = new ArrayList<>(regionInTransition.size());
+    for (Pair<RegionStateNode, Long> entry : regionInTransition.values()) {
+      regions.add(entry.getFirst());
+    }
+    return regions;
   }
 
   public void setTableStateManager(TableStateManager tableStateManager) {
@@ -154,5 +201,4 @@ public void setTableStateManager(TableStateManager tableStateManager) {
   private static boolean isReplica(RegionStateNode regionStateNode) {
     return regionStateNode.getRegionInfo().getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID;
   }
-
 }

hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestAssignmentManagerRitDurationMetrics.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.hadoop.hbase.master;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.master.assignment.AssignmentTestingUtil;
+import org.apache.hadoop.hbase.testclassification.MasterTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+@Tag(MasterTests.TAG)
+@Tag(MediumTests.TAG)
+public class TestAssignmentManagerRitDurationMetrics {
+
+  private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+  private static final byte[] FAMILY = Bytes.toBytes("family");
+  private static final long WAIT_TIMEOUT_MS = 10_000L;
+
+  private static HMaster MASTER;
+  private static final String RIT_DURATION_NUM_OPS_METRIC = "RitDuration_num_ops";
+
+  @BeforeAll
+  public static void startCluster() throws Exception {
+    TEST_UTIL.startMiniCluster(2);
+    MASTER = TEST_UTIL.getMiniHBaseCluster().getMaster();
+  }
+
+  @AfterAll
+  public static void after() throws Exception {
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @Test
+  public void testRitDurationHistogramMetric(TestInfo testInfo) throws Exception {
+    TableName tableName = TableName.valueOf(testInfo.getTestMethod().orElseThrow().getName());
+    try (Table table = TEST_UTIL.createTable(tableName, FAMILY)) {
+      RegionInfo regionInfo =
+        MASTER.getAssignmentManager().getRegionStates().getRegionsOfTable(tableName).get(0);
+      TEST_UTIL.waitFor(WAIT_TIMEOUT_MS,
+        () -> !AssignmentTestingUtil.isRegionInTransition(regionInfo, MASTER.getAssignmentManager())
+          && MASTER.getAssignmentManager().getRegionStates().getRegionServerOfRegion(regionInfo)
+              != null);
+
+      MetricsAssignmentManagerSource amSource =
+        MASTER.getAssignmentManager().getAssignmentManagerMetrics().getMetricsProcSource();
+      long ritDurationNumOps =
+        getMetricValue(snapshotMetrics(amSource), RIT_DURATION_NUM_OPS_METRIC);
+
+      ServerName current =
+        MASTER.getAssignmentManager().getRegionStates().getRegionServerOfRegion(regionInfo);
+      ServerName target = MASTER.getServerManager().getOnlineServersList().stream()
+        .filter(sn -> !sn.equals(current)).findFirst()
+        .orElseThrow(() -> new IllegalStateException("Need at least two regionservers"));
+
+      TEST_UTIL.getAdmin().move(regionInfo.getEncodedNameAsBytes(), target);
+      TEST_UTIL.waitFor(WAIT_TIMEOUT_MS, () -> target
+        .equals(MASTER.getAssignmentManager().getRegionStates().getRegionServerOfRegion(regionInfo))
+        && !AssignmentTestingUtil.isRegionInTransition(regionInfo, MASTER.getAssignmentManager()));
+
+      // num_ops is cumulative (never reset on snapshot); an increase proves the histogram is now
+      // fed on RIT completion. Use >= not ==: background RIT may also add. _max is not asserted --
+      // snapshot() resets it on every read, racing the metrics2 sampler.
+      long ritDurationNumOpsAfter =
+        getMetricValue(snapshotMetrics(amSource), RIT_DURATION_NUM_OPS_METRIC);
+      assertTrue(ritDurationNumOpsAfter >= ritDurationNumOps + 1,
+        "RitDuration histogram num_ops should increase after a region transition");
+    }
+  }
+
+  private MetricsRecord snapshotMetrics(MetricsAssignmentManagerSource amSource) {
+    MetricsCollectorImpl collector = new MetricsCollectorImpl();
+    assertInstanceOf(MetricsSource.class, amSource,
+      "MetricsAssignmentManagerSource should also implement MetricsSource");
+    ((MetricsSource) amSource).getMetrics(collector, true);
+    assertEquals(1, collector.getRecords().size());
+    return collector.getRecords().get(0);
+  }
+
+  private long getMetricValue(MetricsRecord record, String metricName) {
+    for (AbstractMetric metric : record.metrics()) {
+      if (metricName.equals(metric.name())) {
+        return metric.value().longValue();
+      }
+    }
+    throw new AssertionError("Metric not found: " + metricName);
+  }
+}