001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.master; 019 020import java.io.IOException; 021import java.io.InterruptedIOException; 022import java.util.HashSet; 023import java.util.Iterator; 024import java.util.List; 025import java.util.Set; 026import java.util.concurrent.ExecutorService; 027import java.util.concurrent.Executors; 028import java.util.stream.Collectors; 029import org.apache.hadoop.hbase.ServerMetrics; 030import org.apache.hadoop.hbase.ServerMetricsBuilder; 031import org.apache.hadoop.hbase.ServerName; 032import org.apache.hadoop.hbase.client.VersionInfoUtil; 033import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure; 034import org.apache.hadoop.hbase.util.Pair; 035import org.apache.hadoop.hbase.zookeeper.ZKListener; 036import org.apache.hadoop.hbase.zookeeper.ZKUtil; 037import org.apache.hadoop.hbase.zookeeper.ZKWatcher; 038import org.apache.hadoop.hbase.zookeeper.ZNodePaths; 039import org.apache.yetus.audience.InterfaceAudience; 040import org.apache.zookeeper.KeeperException; 041import org.slf4j.Logger; 042import org.slf4j.LoggerFactory; 043 044import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder; 045 046import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 047import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.RegionServerInfo; 048 049/** 050 * Tracks the online region servers via ZK. 051 * <p/> 052 * Handling of new RSs checking in is done via RPC. This class is only responsible for watching for 053 * expired nodes. It handles listening for changes in the RS node list. The only exception is when 054 * master restart, we will use the list fetched from zk to construct the initial set of live region 055 * servers. 056 * <p/> 057 * If an RS node gets deleted, this automatically handles calling of 058 * {@link ServerManager#expireServer(ServerName)} 059 */ 060@InterfaceAudience.Private 061public class RegionServerTracker extends ZKListener { 062 private static final Logger LOG = LoggerFactory.getLogger(RegionServerTracker.class); 063 private final Set<ServerName> regionServers = new HashSet<>(); 064 private final ServerManager serverManager; 065 private final MasterServices server; 066 // As we need to send request to zk when processing the nodeChildrenChanged event, we'd better 067 // move the operation to a single threaded thread pool in order to not block the zk event 068 // processing since all the zk listener across HMaster will be called in one thread sequentially. 069 private final ExecutorService executor; 070 071 public RegionServerTracker(ZKWatcher watcher, MasterServices server, 072 ServerManager serverManager) { 073 super(watcher); 074 this.server = server; 075 this.serverManager = serverManager; 076 this.executor = Executors.newSingleThreadExecutor( 077 new ThreadFactoryBuilder().setDaemon(true).setNameFormat("RegionServerTracker-%d").build()); 078 } 079 080 private Pair<ServerName, RegionServerInfo> getServerInfo(String name) 081 throws KeeperException, IOException { 082 ServerName serverName = ServerName.parseServerName(name); 083 String nodePath = ZNodePaths.joinZNode(watcher.getZNodePaths().rsZNode, name); 084 byte[] data; 085 try { 086 data = ZKUtil.getData(watcher, nodePath); 087 } catch (InterruptedException e) { 088 throw (InterruptedIOException) new InterruptedIOException().initCause(e); 089 } 090 if (data == null) { 091 // we should receive a children changed event later and then we will expire it, so we still 092 // need to add it to the region server set. 093 LOG.warn("Server node {} does not exist, already dead?", name); 094 return Pair.newPair(serverName, null); 095 } 096 if (data.length == 0 || !ProtobufUtil.isPBMagicPrefix(data)) { 097 // this should not happen actually, unless we have bugs or someone has messed zk up. 098 LOG.warn("Invalid data for region server node {} on zookeeper, data length = {}", name, 099 data.length); 100 return Pair.newPair(serverName, null); 101 } 102 RegionServerInfo.Builder builder = RegionServerInfo.newBuilder(); 103 int magicLen = ProtobufUtil.lengthOfPBMagic(); 104 ProtobufUtil.mergeFrom(builder, data, magicLen, data.length - magicLen); 105 return Pair.newPair(serverName, builder.build()); 106 } 107 108 /** 109 * Starts the tracking of online RegionServers. All RSes will be tracked after this method is 110 * called. 111 * <p/> 112 * In this method, we will also construct the region server sets in {@link ServerManager}. If a 113 * region server is dead between the crash of the previous master instance and the start of the 114 * current master instance, we will schedule a SCP for it. This is done in 115 * {@link ServerManager#findDeadServersAndProcess(Set, Set)}, we call it here under the lock 116 * protection to prevent concurrency issues with server expiration operation. 117 * @param deadServersFromPE the region servers which already have SCP associated. 118 * @param liveServersFromWALDir the live region servers from wal directory. 119 * @param splittingServersFromWALDir Servers whose WALs are being actively 'split'. 120 */ 121 public void start(Set<ServerCrashProcedure> deadServersFromPE, Set<ServerName> liveServersFromWALDir, 122 Set<ServerName> splittingServersFromWALDir) 123 throws KeeperException, IOException { 124 LOG.info("Starting RegionServerTracker; {} have existing ServerCrashProcedures, {} " + 125 "possibly 'live' servers, and {} 'splitting'.", deadServersFromPE.size(), 126 liveServersFromWALDir.size(), splittingServersFromWALDir.size()); 127 // deadServersFromPE is made from a list of outstanding ServerCrashProcedures. 128 // splittingServersFromWALDir are being actively split -- the directory in the FS ends in 129 // '-SPLITTING'. Each splitting server should have a corresponding SCP. Log if not. 130 Set<ServerName> deadServerNames = deadServersFromPE.stream() 131 .map(s -> s.getServerName()).collect(Collectors.toSet()); 132 splittingServersFromWALDir.stream().filter(s -> !deadServerNames.contains(s)). 133 forEach(s -> LOG.error("{} has no matching ServerCrashProcedure", s)); 134 watcher.registerListener(this); 135 synchronized (this) { 136 List<String> servers = 137 ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.getZNodePaths().rsZNode); 138 for (String n : servers) { 139 Pair<ServerName, RegionServerInfo> pair = getServerInfo(n); 140 ServerName serverName = pair.getFirst(); 141 RegionServerInfo info = pair.getSecond(); 142 regionServers.add(serverName); 143 ServerMetrics serverMetrics = info != null ? ServerMetricsBuilder.of(serverName, 144 VersionInfoUtil.getVersionNumber(info.getVersionInfo()), 145 info.getVersionInfo().getVersion()) : ServerMetricsBuilder.of(serverName); 146 serverManager.checkAndRecordNewServer(serverName, serverMetrics); 147 } 148 serverManager.findDeadServersAndProcess(deadServersFromPE, liveServersFromWALDir); 149 } 150 } 151 152 public void stop() { 153 executor.shutdownNow(); 154 } 155 156 private synchronized void refresh() { 157 List<String> names; 158 try { 159 names = ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.getZNodePaths().rsZNode); 160 } catch (KeeperException e) { 161 // here we need to abort as we failed to set watcher on the rs node which means that we can 162 // not track the node deleted evetnt any more. 163 server.abort("Unexpected zk exception getting RS nodes", e); 164 return; 165 } 166 Set<ServerName> servers = 167 names.stream().map(ServerName::parseServerName).collect(Collectors.toSet()); 168 for (Iterator<ServerName> iter = regionServers.iterator(); iter.hasNext();) { 169 ServerName sn = iter.next(); 170 if (!servers.contains(sn)) { 171 LOG.info("RegionServer ephemeral node deleted, processing expiration [{}]", sn); 172 serverManager.expireServer(sn); 173 iter.remove(); 174 } 175 } 176 // here we do not need to parse the region server info as it is useless now, we only need the 177 // server name. 178 boolean newServerAdded = false; 179 for (ServerName sn : servers) { 180 if (regionServers.add(sn)) { 181 newServerAdded = true; 182 LOG.info("RegionServer ephemeral node created, adding [" + sn + "]"); 183 } 184 } 185 if (newServerAdded && server.isInitialized()) { 186 // Only call the check to move servers if a RegionServer was added to the cluster; in this 187 // case it could be a server with a new version so it makes sense to run the check. 188 server.checkIfShouldMoveSystemRegionAsync(); 189 } 190 } 191 192 @Override 193 public void nodeChildrenChanged(String path) { 194 if (path.equals(watcher.getZNodePaths().rsZNode) && !server.isAborted() && 195 !server.isStopped()) { 196 executor.execute(this::refresh); 197 } 198 } 199}