import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
/**
 * Let the server manager know a new regionserver has come online
 * @param ia The remote address
 * @param port The remote port
 * @param serverStartcode
 * @param serverCurrentTime The current time of the region server in ms
 * @return The ServerName we know this server as.
 * @throws IOException
 */
ServerName regionServerStartup(final InetAddress ia, final int port,
  final long serverStartcode, long serverCurrentTime)
throws IOException {
  // Test for case where we get a region startup message from a regionserver
  // that has been quickly restarted but whose znode expiration handler has
  // not yet run, or from a server whose fail we are currently processing.
  // Test its host+port combo is present in serverAddresstoServerInfo.  If it
  // is, reject the server and trigger its expiration. The next time it comes
  // in, it should have been removed from serverAddressToServerInfo and queued
  // for processing by ProcessServerShutdown.
  ServerName sn = new ServerName(ia.getHostName(), port, serverStartcode);
  checkClockSkew(sn, serverCurrentTime);
  checkIsDead(sn, "STARTUP");
  checkAlreadySameHostPort(sn);
  recordNewServer(sn, HServerLoad.EMPTY_HSERVERLOAD);
  return sn;
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
void regionServerReport(ServerName sn, HServerLoad hsl)
throws YouAreDeadException, PleaseHoldException {
  checkIsDead(sn, "REPORT");
  if (!this.onlineServers.containsKey(sn)) {
    // Already have this host+port combo and its just different start code?
    checkAlreadySameHostPort(sn);
    // Just let the server in. Presume master joining a running cluster.
    // recordNewServer is what happens at the end of reportServerStartup.
    // The only thing we are skipping is passing back to the regionserver
    // the ServerName to use. Here we presume a master has already done
    // that so we'll press on with whatever it gave us for ServerName.
    recordNewServer(sn, hsl);
  } else {
    this.onlineServers.put(sn, hsl);
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
void tryRegionServerReport() throws IOException {
  if (!keepLooping() && hbaseMaster == null) {
    // the current server is stopping
    return;
  }
  HServerLoad hsl = buildServerLoad();
  // Why we do this?
  this.requestCount.set(0);
  try {
    this.hbaseMaster.regionServerReport(this.serverNameFromMasterPOV.getVersionedBytes(), hsl);
  } catch (IOException ioe) {
    if (ioe instanceof RemoteException) {
      ioe = ((RemoteException) ioe).unwrapRemoteException();
    }
    if (ioe instanceof YouAreDeadException) {
      // This will be caught and handled as a fatal error in run()
      throw ioe;
    }
    // Couldn't connect to the master, get location from zk and reconnect
    // Method blocks until new master is found or we are stopped
    getMaster();
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
void tryRegionServerReport()
throws IOException {
  if (!keepLooping() && hbaseMaster == null) {
    // the current server is stopping
    return;
  }
  HServerLoad hsl = buildServerLoad();
  // Why we do this?
  this.requestCount.set(0);
  try {
    this.hbaseMaster.regionServerReport(this.serverNameFromMasterPOV.getVersionedBytes(), hsl);
  } catch (IOException ioe) {
    if (ioe instanceof RemoteException) {
      ioe = ((RemoteException)ioe).unwrapRemoteException();
    }
    if (ioe instanceof YouAreDeadException) {
      // This will be caught and handled as a fatal error in run()
      throw ioe;
    }
    // Couldn't connect to the master, get location from zk and reconnect
    // Method blocks until new master is found or we are stopped
    getMaster();
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
void tryRegionServerReport()
throws IOException {
  HServerLoad hsl = buildServerLoad();
  // Why we do this?
  this.requestCount.set(0);
  try {
    this.hbaseMaster.regionServerReport(this.serverNameFromMasterPOV.getVersionedBytes(), hsl);
  } catch (IOException ioe) {
    if (ioe instanceof RemoteException) {
      ioe = ((RemoteException)ioe).unwrapRemoteException();
    }
    if (ioe instanceof YouAreDeadException) {
      // This will be caught and handled as a fatal error in run()
      throw ioe;
    }
    // Couldn't connect to the master, get location from zk and reconnect
    // Method blocks until new master is found or we are stopped
    getMaster();
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
static public ARegionLoad hrlToARL(HServerLoad.RegionLoad rl) throws IOException {
  ARegionLoad arl = new ARegionLoad();
  arl.memStoreSizeMB = rl.getMemStoreSizeMB();
  arl.name = ByteBuffer.wrap(rl.getName());
  arl.storefileIndexSizeMB = rl.getStorefileIndexSizeMB();
  arl.storefiles = rl.getStorefiles();
  arl.storefileSizeMB = rl.getStorefileSizeMB();
  arl.stores = rl.getStores();
  return arl;
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
static public AServerLoad hslToASL(HServerLoad hsl) throws IOException {
   AServerLoad asl = new AServerLoad();
   asl.load = hsl.getLoad();
   asl.maxHeapMB = hsl.getMaxHeapMB();
   asl.memStoreSizeInMB = hsl.getMemStoreSizeInMB();
   asl.numberOfRegions = hsl.getNumberOfRegions();
   asl.numberOfRequests = hsl.getNumberOfRequests();

   Collection<HServerLoad.RegionLoad> regionLoads = hsl.getRegionsLoad().values();
   Schema s = Schema.createArray(ARegionLoad.SCHEMA$);
   GenericData.Array<ARegionLoad> aregionLoads = null;
   if (regionLoads != null) {
     aregionLoads = new GenericData.Array<ARegionLoad>(regionLoads.size(), s);
     for (HServerLoad.RegionLoad rl : regionLoads) {
aregionLoads.add(hrlToARL(rl));
     }
   } else {
     aregionLoads = new GenericData.Array<ARegionLoad>(0, s);
   }
   asl.regionsLoad = aregionLoads;

   asl.storefileIndexSizeInMB = hsl.getStorefileIndexSizeInMB();
   asl.storefiles = hsl.getStorefiles();
   asl.storefileSizeInMB = hsl.getStorefileSizeInMB();
   asl.usedHeapMB = hsl.getUsedHeapMB();
   return asl;
 }
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
static public AServerInfo hsiToASI(ServerName sn, HServerLoad hsl) throws IOException {
  AServerInfo asi = new AServerInfo();
  asi.infoPort = -1;
  asi.load = hslToASL(hsl);
  asi.serverAddress = hsaToASA(new HServerAddress(sn.getHostname(), sn.getPort()));
  asi.serverName = new Utf8(sn.toString());
  asi.startCode = sn.getStartcode();
  return asi;
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
private void dumpServers(HMaster master, PrintWriter out) {
  Map<ServerName, HServerLoad> servers =
    master.getServerManager().getOnlineServers();
  for (Map.Entry<ServerName, HServerLoad> e : servers.entrySet()) {
    out.println(e.getKey() + ": " + e.getValue());
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
/**
 * @param address
 * @return HServerLoad if serverName is known else null
 * @deprecated Use {@link #getLoad(HServerAddress)}
 */
public HServerLoad getLoad(final HServerAddress address) {
  ServerName sn = new ServerName(address.toString(), ServerName.NON_STARTCODE);
  ServerName actual =
    ServerName.findServerWithSameHostnamePort(this.getOnlineServersList(), sn);
  return actual == null? null: getLoad(actual);
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
/**
 * Compute the average load across all region servers.
 * Currently, this uses a very naive computation - just uses the number of
 * regions being served, ignoring stats about number of requests.
 * @return the average load
 */
public double getAverageLoad() {
  int totalLoad = 0;
  int numServers = 0;
  double averageLoad = 0.0;
  for (HServerLoad hsl: this.onlineServers.values()) {
      numServers++;
      totalLoad += hsl.getNumberOfRegions();
  }
  averageLoad = (double)totalLoad / (double)numServers;
  return averageLoad;
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
/**
 * @return Read-only map of servers to serverinfo
 */
public Map<ServerName, HServerLoad> getOnlineServers() {
  // Presumption is that iterating the returned Map is OK.
  synchronized (this.onlineServers) {
    return Collections.unmodifiableMap(this.onlineServers);
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
@Override
public void regionServerReport(final byte [] sn, final HServerLoad hsl)
throws IOException {
  this.serverManager.regionServerReport(ServerName.parseVersionedServerName(sn), hsl);
  if (hsl != null && this.metrics != null) {
    // Up our metrics.
    this.metrics.incrementRequests(hsl.getTotalNumberOfRequests());
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
@Override
public Map<String, HServerLoad> getRegionServers() {
  Map<String, HServerLoad> data = new HashMap<String, HServerLoad>();
  for (final Entry<ServerName, HServerLoad> entry :
    master.getServerManager().getOnlineServers().entrySet()) {
    data.put(entry.getKey().getServerName(),
        entry.getValue());
  }
  return data;
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
HServerLoad buildServerLoad() {
  Collection<HRegion> regions = getOnlineRegionsLocalContext();
  TreeMap<byte[], HServerLoad.RegionLoad> regionLoads =
      new TreeMap<byte[], HServerLoad.RegionLoad>(Bytes.BYTES_COMPARATOR);
  for (HRegion region : regions) {
    regionLoads.put(region.getRegionName(), createRegionLoad(region));
  }
  MemoryUsage memory = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage();
  return new HServerLoad(requestCount.get(), (int) metrics.getRequests(),
      (int) (memory.getUsed() / 1024 / 1024), (int) (memory.getMax() / 1024 / 1024), regionLoads,
      this.hlog.getCoprocessorHost().getCoprocessors());
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
private HServerLoad.RegionLoad createRegionLoad(final HRegion r) {
  byte[] name = r.getRegionName();
  int stores = 0;
  int storefiles = 0;
  int storeUncompressedSizeMB = 0;
  int storefileSizeMB = 0;
  int memstoreSizeMB = (int) (r.memstoreSize.get() / 1024 / 1024);
  int storefileIndexSizeMB = 0;
  int rootIndexSizeKB = 0;
  int totalStaticIndexSizeKB = 0;
  int totalStaticBloomSizeKB = 0;
  long totalCompactingKVs = 0;
  long currentCompactedKVs = 0;
  synchronized (r.stores) {
    stores += r.stores.size();
    for (Store store : r.stores.values()) {
      storefiles += store.getStorefilesCount();
      storeUncompressedSizeMB += (int) (store.getStoreSizeUncompressed() / 1024 / 1024);
      storefileSizeMB += (int) (store.getStorefilesSize() / 1024 / 1024);
      storefileIndexSizeMB += (int) (store.getStorefilesIndexSize() / 1024 / 1024);
      CompactionProgress progress = store.getCompactionProgress();
      if (progress != null) {
        totalCompactingKVs += progress.totalCompactingKVs;
        currentCompactedKVs += progress.currentCompactedKVs;
      }

      rootIndexSizeKB += (int) (store.getStorefilesIndexSize() / 1024);

      totalStaticIndexSizeKB += (int) (store.getTotalStaticIndexSize() / 1024);

      totalStaticBloomSizeKB += (int) (store.getTotalStaticBloomSize() / 1024);
    }
  }
  return new HServerLoad.RegionLoad(name, stores, storefiles, storeUncompressedSizeMB,
      storefileSizeMB, memstoreSizeMB, storefileIndexSizeMB, rootIndexSizeKB,
      totalStaticIndexSizeKB, totalStaticBloomSizeKB, (int) r.readRequestsCount.get(),
      (int) r.writeRequestsCount.get(), totalCompactingKVs, currentCompactedKVs);
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
private void verifyRegionServers(Map<String, HServerLoad> regions) {
  Set<String> expected = new HashSet<String>();
  for (int i = 0; i < 4; ++i) {
    HRegionServer rs = TEST_UTIL.getMiniHBaseCluster().getRegionServer(i);
    expected.add(rs.getServerName().getServerName());
  }

  int found = 0;
  for (java.util.Map.Entry<String, HServerLoad> entry : regions.entrySet()) {
    if (expected.contains(entry.getKey())) {
      ++found;
    }
  }
  Assert.assertEquals(4, found);
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
@Before
public void before() throws ZooKeeperConnectionException, IOException {
  // TODO: Make generic versions of what we do below and put up in a mocking
  // utility class or move up into HBaseTestingUtility.

  // Mock a Server.  Have it return a legit Configuration and ZooKeeperWatcher.
  // If abort is called, be sure to fail the test (don't just swallow it
  // silently as is mockito default).
  this.server = Mockito.mock(Server.class);
  Mockito.when(server.getConfiguration()).thenReturn(HTU.getConfiguration());
  this.watcher =
    new ZooKeeperWatcher(HTU.getConfiguration(), "mockedServer", this.server, true);
  Mockito.when(server.getZooKeeper()).thenReturn(this.watcher);
  Mockito.doThrow(new RuntimeException("Aborted")).
    when(server).abort(Mockito.anyString(), (Throwable)Mockito.anyObject());

  // Mock a ServerManager.  Say server SERVERNAME_{A,B} are online.  Also
  // make it so if close or open, we return 'success'.
  this.serverManager = Mockito.mock(ServerManager.class);
  Mockito.when(this.serverManager.isServerOnline(SERVERNAME_A)).thenReturn(true);
  Mockito.when(this.serverManager.isServerOnline(SERVERNAME_B)).thenReturn(true);
  final Map<ServerName, HServerLoad> onlineServers = new HashMap<ServerName, HServerLoad>();
  onlineServers.put(SERVERNAME_B, new HServerLoad());
  onlineServers.put(SERVERNAME_A, new HServerLoad());
  Mockito.when(this.serverManager.getOnlineServersList()).thenReturn(
      new ArrayList<ServerName>(onlineServers.keySet()));
  Mockito.when(this.serverManager.getOnlineServers()).thenReturn(onlineServers);
  Mockito.when(this.serverManager.sendRegionClose(SERVERNAME_A, REGIONINFO, -1)).
    thenReturn(true);
  Mockito.when(this.serverManager.sendRegionClose(SERVERNAME_B, REGIONINFO, -1)).
    thenReturn(true);
  // Ditto on open.
  Mockito.when(this.serverManager.sendRegionOpen(SERVERNAME_A, REGIONINFO, -1)).
    thenReturn(RegionOpeningState.OPENED);
  Mockito.when(this.serverManager.sendRegionOpen(SERVERNAME_B, REGIONINFO, -1)).
  thenReturn(RegionOpeningState.OPENED);
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
/**
 * Test verifies whether assignment is skipped for regions of tables in DISABLING state during
 * clean cluster startup. See HBASE-6281.
 *
 * @throws KeeperException
 * @throws IOException
 * @throws Exception
 */
@Test
public void testDisablingTableRegionsAssignmentDuringCleanClusterStartup()
    throws KeeperException, IOException, Exception {
  this.server.getConfiguration().setClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS,
      MockedLoadBalancer.class, LoadBalancer.class);
  Mockito.when(this.serverManager.getOnlineServers()).thenReturn(
      new HashMap<ServerName, HServerLoad>(0));
  List<ServerName> destServers = new ArrayList<ServerName>(1);
  destServers.add(SERVERNAME_A);
  Mockito.when(this.serverManager.getDrainingServersList()).thenReturn(destServers);
  // To avoid cast exception in DisableTableHandler process.
  //Server server = new HMaster(HTU.getConfiguration());
  AssignmentManagerWithExtrasForTesting am = setUpMockedAssignmentManager(server,
      this.serverManager);
  AtomicBoolean gate = new AtomicBoolean(false);
  if (balancer instanceof MockedLoadBalancer) {
    ((MockedLoadBalancer) balancer).setGateVariable(gate);
  }
  try{
    // set table in disabling state.
    am.getZKTable().setDisablingTable(REGIONINFO.getTableNameAsString());
    am.joinCluster();
    // should not call retainAssignment if we get empty regions in assignAllUserRegions.
    assertFalse(
        "Assign should not be invoked for disabling table regions during clean cluster startup.",
        gate.get());
    // need to change table state from disabling to disabled.
    assertTrue("Table should be disabled.",
        am.getZKTable().isDisabledTable(REGIONINFO.getTableNameAsString()));
  } finally {
    this.server.getConfiguration().setClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS,
      DefaultLoadBalancer.class, LoadBalancer.class);
    am.getZKTable().setEnabledTable(REGIONINFO.getTableNameAsString());
    am.shutdown();
  }
}
 

import org.apache.hadoop.hbase.HServerLoad; //导入依赖的package包/类
public RegionLoadAdapter(HBaseAdmin admin, Map<byte[], HRegionInfo> regionMap, Args args) throws IOException {
    long timestamp = System.currentTimeMillis();

    ClusterStatus clusterStatus = admin.getClusterStatus();
    Collection<ServerName> serverNames = clusterStatus.getServers();
    for (ServerName serverName : serverNames) {
        HServerLoad serverLoad = clusterStatus.getLoad(serverName);
        for (Map.Entry<byte[], HServerLoad.RegionLoad> entry : serverLoad.getRegionsLoad().entrySet()) {
            if (regionMap.get(entry.getKey()) != null)
                regionLoadMap.put(regionMap.get(entry.getKey()), new RegionLoadDelegator(entry.getValue()));
        }
    }

    Util.printVerboseMessage(args, "RegionLoadAdapter", timestamp);
}