HBASE-22709 Add a chore thread in master to do hbck checking

Author: infraio

hbase-server/src/main/jamon/org/apache/hadoop/hbase/tmpl/master/MasterStatusTmpl.jamon

@@ -149,7 +149,8 @@ AssignmentManager assignmentManager = master.getAssignmentManager();
                 <li class="active"><a href="/master-status">Home</a></li>
                 <li><a href="/tablesDetailed.jsp">Table Details</a></li>
                 <%if master.isActiveMaster() %>
-                <li><a href="/procedures.jsp">Procedures &amp; Locks</a></li>
+                    <li><a href="/procedures.jsp">Procedures &amp; Locks</a></li>
+                    <li><a href="/hbck.jsp">HBCK Report</a></li>
                 </%if>
                 <li><a href="/processMaster.jsp">Process Metrics</a></li>
                 <li><a href="/logs/">Local Logs</a></li>

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

@@ -385,6 +385,7 @@ public void run() {
   private ClusterStatusPublisher clusterStatusPublisherChore = null;
   private SnapshotCleanerChore snapshotCleanerChore = null;
 
+  private HbckChecker hbckChecker;
   CatalogJanitor catalogJanitorChore;
   private LogCleaner logCleaner;
   private HFileCleaner hfileCleaner;
@@ -1108,6 +1109,8 @@ private void finishActiveMasterInitialization(MonitoredTask status) throws IOExc
     getChoreService().scheduleChore(normalizerChore);
     this.catalogJanitorChore = new CatalogJanitor(this);
     getChoreService().scheduleChore(catalogJanitorChore);
+    this.hbckChecker = new HbckChecker(this);
+    getChoreService().scheduleChore(hbckChecker);
     this.serverManager.startChore();
 
     // Only for rolling upgrade, where we need to migrate the data in namespace table to meta table.
@@ -1587,6 +1590,7 @@ private void stopChores() {
       choreService.cancelChore(this.hfileCleaner);
       choreService.cancelChore(this.replicationBarrierCleaner);
       choreService.cancelChore(this.snapshotCleanerChore);
+      choreService.cancelChore(this.hbckChecker);
     }
   }
 
@@ -3756,4 +3760,8 @@ public Map<String, ReplicationStatus> getWalGroupsReplicationStatus() {
     }
     return super.getWalGroupsReplicationStatus();
   }
+
+  public HbckChecker getHbckChecker() {
+    return this.hbckChecker;
+  }
 }

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

@@ -0,0 +1,220 @@
+/**
+ * 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 java.io.IOException;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.ScheduledChore;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.util.FSUtils;
+import org.apache.hadoop.hbase.util.HbckRegionInfo;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.yetus.audience.InterfaceStability;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
+
+/**
+ * Used to do the hbck checking job at master side.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class HbckChecker extends ScheduledChore {
+  private static final Logger LOG = LoggerFactory.getLogger(HbckChecker.class.getName());
+
+  private static final String HBCK_CHECKER_INTERVAL = "hbase.master.hbck.checker.interval";
+  private static final int DEFAULT_HBCK_CHECKER_INTERVAL = 60 * 60 * 1000;
+
+  private final MasterServices master;
+
+  /**
+   * This map contains the state of all hbck items.  It maps from encoded region
+   * name to HbckRegionInfo structure.  The information contained in HbckRegionInfo is used
+   * to detect and correct consistency (hdfs/meta/deployment) problems.
+   */
+  private final Map<String, HbckRegionInfo> regionInfoMap = new HashMap<>();
+
+  private final Map<String, ServerName> orphanRegionsOnRS = new HashMap<>();
+  private final List<String> orphanRegionsOnFS = new LinkedList<>();
+  private final Map<String, Pair<ServerName, List<ServerName>>> inconsistentRegions =
+      new HashMap<>();
+
+  private final Map<String, ServerName> orphanRegionsOnRSSnapshot = new HashMap<>();
+  private final List<String> orphanRegionsOnFSSnapshot = new LinkedList<>();
+  private final Map<String, Pair<ServerName, List<ServerName>>> inconsistentRegionsSnapshot =
+      new HashMap<>();
+
+  private volatile boolean running = false;
+
+  public HbckChecker(MasterServices master) {
+    super("HbckChecker-", master,
+        master.getConfiguration().getInt(HBCK_CHECKER_INTERVAL, DEFAULT_HBCK_CHECKER_INTERVAL));
+    this.master = master;
+  }
+
+  @Override
+  protected void chore() {
+    running = true;
+    regionInfoMap.clear();
+    orphanRegionsOnRS.clear();
+    orphanRegionsOnFS.clear();
+    inconsistentRegions.clear();
+    loadRegionsFromInMemoryState();
+    loadRegionsFromRSReport();
+    try {
+      loadRegionsFromFS();
+    } catch (IOException e) {
+      LOG.warn("Faile to load the regions from filesystem", e);
+    }
+    saveCheckResultToSnapshot();
+    running = false;
+  }
+
+  private void saveCheckResultToSnapshot() {
+    synchronized (orphanRegionsOnRSSnapshot) {
+      orphanRegionsOnRSSnapshot.clear();
+      orphanRegionsOnRS.entrySet()
+          .forEach(e -> orphanRegionsOnRSSnapshot.put(e.getKey(), e.getValue()));
+    }
+    synchronized (orphanRegionsOnFSSnapshot) {
+      orphanRegionsOnFSSnapshot.clear();
+      orphanRegionsOnFSSnapshot.addAll(orphanRegionsOnFS);
+    }
+    synchronized (inconsistentRegionsSnapshot) {
+      inconsistentRegionsSnapshot.clear();
+      inconsistentRegions.entrySet()
+          .forEach(e -> inconsistentRegionsSnapshot.put(e.getKey(), e.getValue()));
+    }
+  }
+
+  private void loadRegionsFromInMemoryState() {
+    List<RegionState> regionStates =
+        master.getAssignmentManager().getRegionStates().getRegionStates();
+    for (RegionState regionState : regionStates) {
+      RegionInfo regionInfo = regionState.getRegion();
+      HbckRegionInfo.MetaEntry metaEntry =
+          new HbckRegionInfo.MetaEntry(regionInfo, regionState.getServerName(),
+              regionState.getStamp());
+      regionInfoMap.put(regionInfo.getEncodedName(), new HbckRegionInfo(metaEntry));
+    }
+  }
+
+  private void loadRegionsFromRSReport() {
+    Map<ServerName, Set<byte[]>> rsReports = master.getAssignmentManager().getRSReports();
+    for (Map.Entry<ServerName, Set<byte[]>> entry : rsReports.entrySet()) {
+      ServerName serverName = entry.getKey();
+      for (byte[] regionName : entry.getValue()) {
+        String encodedRegionName = RegionInfo.encodeRegionName(regionName);
+        HbckRegionInfo hri = regionInfoMap.get(encodedRegionName);
+        if (hri == null) {
+          orphanRegionsOnRS.put(encodedRegionName, serverName);
+          continue;
+        }
+        hri.addServer(hri.getMetaEntry(), serverName);
+      }
+    }
+
+    for (Map.Entry<String, HbckRegionInfo> entry : regionInfoMap.entrySet()) {
+      String encodedRegionName = entry.getKey();
+      HbckRegionInfo hri = entry.getValue();
+      ServerName locationInMeta = hri.getMetaEntry().getRegionServer();
+      if (hri.getDeployedOn().size() == 0) {
+        // Master thought this region opened, but no regionserver reported it.
+        inconsistentRegions.put(encodedRegionName, new Pair<>(locationInMeta, new LinkedList<>()));
+      } else if (hri.getDeployedOn().size() > 1) {
+        // More than one regionserver reported opened this region
+        inconsistentRegions.put(encodedRegionName, new Pair<>(locationInMeta, hri.getDeployedOn()));
+      } else if (!hri.getDeployedOn().get(0).equals(locationInMeta)) {
+        // Master thought this region opened on Server1, but regionserver reported Server2
+        inconsistentRegions.put(encodedRegionName, new Pair<>(locationInMeta, hri.getDeployedOn()));
+      }
+    }
+  }
+
+  private void loadRegionsFromFS() throws IOException {
+    Path rootDir = master.getMasterFileSystem().getRootDir();
+    FileSystem fs = master.getMasterFileSystem().getFileSystem();
+
+    // list all tables from HDFS
+    List<FileStatus> tableDirs = Lists.newArrayList();
+    List<Path> paths = FSUtils.getTableDirs(fs, rootDir);
+    for (Path path : paths) {
+      tableDirs.add(fs.getFileStatus(path));
+    }
+
+    for (FileStatus tableDir : tableDirs) {
+      FileStatus[] regionDirs = fs.listStatus(tableDir.getPath());
+      for (FileStatus regionDir : regionDirs) {
+        String encodedRegionName = regionDir.getPath().getName();
+        HbckRegionInfo hri = regionInfoMap.get(encodedRegionName);
+        if (hri == null) {
+          orphanRegionsOnFS.add(encodedRegionName);
+          continue;
+        }
+        HbckRegionInfo.HdfsEntry hdfsEntry =
+            new HbckRegionInfo.HdfsEntry(regionDir.getPath(), regionDir.getModificationTime());
+        hri.setHdfsEntry(hdfsEntry);
+      }
+    }
+  }
+
+  /**
+   * When running, the HBCK report may be changed later.
+   */
+  public boolean isRunning() {
+    return running;
+  }
+
+  public Map<String, ServerName> getOrphanRegionsOnRS() {
+    synchronized (orphanRegionsOnRSSnapshot) {
+      return this.orphanRegionsOnRSSnapshot;
+    }
+  }
+
+  public List<String> getOrphanRegionsOnFS() {
+    synchronized (orphanRegionsOnFSSnapshot) {
+      return this.orphanRegionsOnFSSnapshot;
+    }
+  }
+
+  /**
+   * Found the inconsistent regions. There are three case:
+   * case 1. Master thought this region opened, but no regionserver reported it.
+   * case 2. Master thought this region opened on Server1, but regionserver reported Server2
+   * case 3. More than one regionservers reported opened this region
+   *
+   * @return the map of inconsistent regions. Key is the region name. Value is a pair of location in
+   *         meta and the regionservers which reported opened this region.
+   */
+  public Map<String, Pair<ServerName, List<ServerName>>> getInconsistentRegions() {
+    synchronized (inconsistentRegionsSnapshot) {
+      return this.inconsistentRegionsSnapshot;
+    }
+  }
+}

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

@@ -1467,6 +1467,12 @@ public long submitServerCrash(ServerName serverName, boolean shouldSplitWal) {
       LOG.info("Skip to add SCP for {} since this server should be OFFLINE already", serverName);
       return -1;
     }
+
+    // Remove the in-memory rsReports result
+    synchronized (rsReports) {
+      rsReports.remove(serverName);
+    }
+
     // we hold the write lock here for fencing on reportRegionStateTransition. Once we set the
     // server state to CRASHED, we will no longer accept the reportRegionStateTransition call from
     // this server. This is used to simplify the implementation for TRSP and SCP, where we can make
@@ -2084,4 +2090,15 @@ public Map<String, Pair<ServerName, Set<ServerName>>> getProblematicRegions() {
 
     return problematicRegions;
   }
+
+  /**
+   * @return a snapshot of rsReports
+   */
+  public Map<ServerName, Set<byte[]>> getRSReports() {
+    Map<ServerName, Set<byte[]>> rsReportsSnapshot = new HashMap<>();
+    synchronized (rsReports) {
+      rsReports.entrySet().forEach(e -> rsReportsSnapshot.put(e.getKey(), e.getValue()));
+    }
+    return rsReportsSnapshot;
+  }
 }