View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.master;
20  
21  import java.io.IOException;
22  import java.net.InetAddress;
23  import java.util.ArrayList;
24  import java.util.Collections;
25  import java.util.HashMap;
26  import java.util.HashSet;
27  import java.util.Iterator;
28  import java.util.List;
29  import java.util.Map;
30  import java.util.Map.Entry;
31  import java.util.Set;
32  import java.util.concurrent.ConcurrentHashMap;
33  import java.util.concurrent.ConcurrentNavigableMap;
34  import java.util.concurrent.ConcurrentSkipListMap;
35  import java.util.concurrent.CopyOnWriteArrayList;
36  
37  import org.apache.commons.logging.Log;
38  import org.apache.commons.logging.LogFactory;
39  import org.apache.hadoop.conf.Configuration;
40  import org.apache.hadoop.hbase.ClockOutOfSyncException;
41  import org.apache.hadoop.hbase.HConstants;
42  import org.apache.hadoop.hbase.HRegionInfo;
43  import org.apache.hadoop.hbase.NotServingRegionException;
44  import org.apache.hadoop.hbase.RegionLoad;
45  import org.apache.hadoop.hbase.Server;
46  import org.apache.hadoop.hbase.ServerLoad;
47  import org.apache.hadoop.hbase.ServerName;
48  import org.apache.hadoop.hbase.YouAreDeadException;
49  import org.apache.hadoop.hbase.ZooKeeperConnectionException;
50  import org.apache.hadoop.hbase.classification.InterfaceAudience;
51  import org.apache.hadoop.hbase.client.ClusterConnection;
52  import org.apache.hadoop.hbase.client.RetriesExhaustedException;
53  import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController;
54  import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
55  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
56  import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
57  import org.apache.hadoop.hbase.monitoring.MonitoredTask;
58  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
59  import org.apache.hadoop.hbase.protobuf.RequestConverter;
60  import org.apache.hadoop.hbase.protobuf.ResponseConverter;
61  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
62  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest;
63  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse;
64  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ServerInfo;
65  import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest;
66  import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds;
67  import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.StoreSequenceId;
68  import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
69  import org.apache.hadoop.hbase.regionserver.HRegionServer;
70  import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
71  import org.apache.hadoop.hbase.security.User;
72  import org.apache.hadoop.hbase.util.Bytes;
73  import org.apache.hadoop.hbase.util.Pair;
74  import org.apache.hadoop.hbase.util.RetryCounter;
75  import org.apache.hadoop.hbase.util.RetryCounterFactory;
76  import org.apache.hadoop.hbase.zookeeper.ZKUtil;
77  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
78  import org.apache.zookeeper.KeeperException;
79  
80  import com.google.common.annotations.VisibleForTesting;
81  import com.google.protobuf.ByteString;
82  import com.google.protobuf.ServiceException;
83  
84  /**
85   * The ServerManager class manages info about region servers.
86   * <p>
87   * Maintains lists of online and dead servers.  Processes the startups,
88   * shutdowns, and deaths of region servers.
89   * <p>
90   * Servers are distinguished in two different ways.  A given server has a
91   * location, specified by hostname and port, and of which there can only be one
92   * online at any given time.  A server instance is specified by the location
93   * (hostname and port) as well as the startcode (timestamp from when the server
94   * was started).  This is used to differentiate a restarted instance of a given
95   * server from the original instance.
96   * <p>
97   * If a sever is known not to be running any more, it is called dead. The dead
98   * server needs to be handled by a ServerShutdownHandler.  If the handler is not
99   * enabled yet, the server can't be handled right away so it is queued up.
100  * After the handler is enabled, the server will be submitted to a handler to handle.
101  * However, the handler may be just partially enabled.  If so,
102  * the server cannot be fully processed, and be queued up for further processing.
103  * A server is fully processed only after the handler is fully enabled
104  * and has completed the handling.
105  */
106 @InterfaceAudience.Private
107 public class ServerManager {
108   public static final String WAIT_ON_REGIONSERVERS_MAXTOSTART =
109       "hbase.master.wait.on.regionservers.maxtostart";
110 
111   public static final String WAIT_ON_REGIONSERVERS_MINTOSTART =
112       "hbase.master.wait.on.regionservers.mintostart";
113 
114   public static final String WAIT_ON_REGIONSERVERS_TIMEOUT =
115       "hbase.master.wait.on.regionservers.timeout";
116 
117   public static final String WAIT_ON_REGIONSERVERS_INTERVAL =
118       "hbase.master.wait.on.regionservers.interval";
119 
120   private static final Log LOG = LogFactory.getLog(ServerManager.class);
121 
122   // Set if we are to shutdown the cluster.
123   private volatile boolean clusterShutdown = false;
124 
125   /**
126    * The last flushed sequence id for a region.
127    */
128   private final ConcurrentNavigableMap<byte[], Long> flushedSequenceIdByRegion =
129     new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
130 
131   /**
132    * The last flushed sequence id for a store in a region.
133    */
134   private final ConcurrentNavigableMap<byte[], ConcurrentNavigableMap<byte[], Long>>
135     storeFlushedSequenceIdsByRegion =
136     new ConcurrentSkipListMap<byte[], ConcurrentNavigableMap<byte[], Long>>(Bytes.BYTES_COMPARATOR);
137 
138   /** Map of registered servers to their current load */
139   private final ConcurrentHashMap<ServerName, ServerLoad> onlineServers =
140     new ConcurrentHashMap<ServerName, ServerLoad>();
141 
142   /**
143    * Map of admin interfaces per registered regionserver; these interfaces we use to control
144    * regionservers out on the cluster
145    */
146   private final Map<ServerName, AdminService.BlockingInterface> rsAdmins =
147     new HashMap<ServerName, AdminService.BlockingInterface>();
148 
149   /** List of region servers that should not get any more new regions. */
150   private final ArrayList<ServerName> drainingServers =
151     new ArrayList<ServerName>();
152 
153   private final Server master;
154   private final MasterServices services;
155   private final ClusterConnection connection;
156 
157   private final DeadServer deadservers = new DeadServer();
158 
159   private final long maxSkew;
160   private final long warningSkew;
161 
162   private final RetryCounterFactory pingRetryCounterFactory;
163   private final RpcControllerFactory rpcControllerFactory;
164 
165   /**
166    * Set of region servers which are dead but not processed immediately. If one
167    * server died before master enables ServerShutdownHandler, the server will be
168    * added to this set and will be processed through calling
169    * {@link ServerManager#processQueuedDeadServers()} by master.
170    * <p>
171    * A dead server is a server instance known to be dead, not listed in the /hbase/rs
172    * znode any more. It may have not been submitted to ServerShutdownHandler yet
173    * because the handler is not enabled.
174    * <p>
175    * A dead server, which has been submitted to ServerShutdownHandler while the
176    * handler is not enabled, is queued up.
177    * <p>
178    * So this is a set of region servers known to be dead but not submitted to
179    * ServerShutdownHandler for processing yet.
180    */
181   private Set<ServerName> queuedDeadServers = new HashSet<ServerName>();
182 
183   /**
184    * Set of region servers which are dead and submitted to ServerShutdownHandler to process but not
185    * fully processed immediately.
186    * <p>
187    * If one server died before assignment manager finished the failover cleanup, the server will be
188    * added to this set and will be processed through calling
189    * {@link ServerManager#processQueuedDeadServers()} by assignment manager.
190    * <p>
191    * The Boolean value indicates whether log split is needed inside ServerShutdownHandler
192    * <p>
193    * ServerShutdownHandler processes a dead server submitted to the handler after the handler is
194    * enabled. It may not be able to complete the processing because meta is not yet online or master
195    * is currently in startup mode. In this case, the dead server will be parked in this set
196    * temporarily.
197    */
198   private Map<ServerName, Boolean> requeuedDeadServers
199     = new ConcurrentHashMap<ServerName, Boolean>();
200 
201   /** Listeners that are called on server events. */
202   private List<ServerListener> listeners = new CopyOnWriteArrayList<ServerListener>();
203 
204   /**
205    * Constructor.
206    * @param master
207    * @param services
208    * @throws ZooKeeperConnectionException
209    */
210   public ServerManager(final Server master, final MasterServices services)
211       throws IOException {
212     this(master, services, true);
213   }
214 
215   ServerManager(final Server master, final MasterServices services,
216       final boolean connect) throws IOException {
217     this.master = master;
218     this.services = services;
219     Configuration c = master.getConfiguration();
220     maxSkew = c.getLong("hbase.master.maxclockskew", 30000);
221     warningSkew = c.getLong("hbase.master.warningclockskew", 10000);
222     this.connection = connect ? master.getClusterConnection() : null;
223     int pingMaxAttempts = Math.max(1, master.getConfiguration().getInt(
224       "hbase.master.maximum.ping.server.attempts", 10));
225     int pingSleepInterval = Math.max(1, master.getConfiguration().getInt(
226       "hbase.master.ping.server.retry.sleep.interval", 100));
227     this.pingRetryCounterFactory = new RetryCounterFactory(pingMaxAttempts, pingSleepInterval);
228     this.rpcControllerFactory = this.connection == null
229         ? null
230         : connection.getRpcControllerFactory();
231   }
232 
233   /**
234    * Add the listener to the notification list.
235    * @param listener The ServerListener to register
236    */
237   public void registerListener(final ServerListener listener) {
238     this.listeners.add(listener);
239   }
240 
241   /**
242    * Remove the listener from the notification list.
243    * @param listener The ServerListener to unregister
244    */
245   public boolean unregisterListener(final ServerListener listener) {
246     return this.listeners.remove(listener);
247   }
248 
249   /**
250    * Let the server manager know a new regionserver has come online
251    * @param request the startup request
252    * @param ia the InetAddress from which request is received
253    * @return The ServerName we know this server as.
254    * @throws IOException
255    */
256   ServerName regionServerStartup(RegionServerStartupRequest request, InetAddress ia)
257       throws IOException {
258     // Test for case where we get a region startup message from a regionserver
259     // that has been quickly restarted but whose znode expiration handler has
260     // not yet run, or from a server whose fail we are currently processing.
261     // Test its host+port combo is present in serverAddresstoServerInfo.  If it
262     // is, reject the server and trigger its expiration. The next time it comes
263     // in, it should have been removed from serverAddressToServerInfo and queued
264     // for processing by ProcessServerShutdown.
265 
266     final String hostname = request.hasUseThisHostnameInstead() ?
267         request.getUseThisHostnameInstead() :ia.getHostName();
268     ServerName sn = ServerName.valueOf(hostname, request.getPort(),
269       request.getServerStartCode());
270     checkClockSkew(sn, request.getServerCurrentTime());
271     checkIsDead(sn, "STARTUP");
272     if (!checkAndRecordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD)) {
273       LOG.warn("THIS SHOULD NOT HAPPEN, RegionServerStartup"
274         + " could not record the server: " + sn);
275     }
276     return sn;
277   }
278 
279   private ConcurrentNavigableMap<byte[], Long> getOrCreateStoreFlushedSequenceId(
280     byte[] regionName) {
281     ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
282         storeFlushedSequenceIdsByRegion.get(regionName);
283     if (storeFlushedSequenceId != null) {
284       return storeFlushedSequenceId;
285     }
286     storeFlushedSequenceId = new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
287     ConcurrentNavigableMap<byte[], Long> alreadyPut =
288         storeFlushedSequenceIdsByRegion.putIfAbsent(regionName, storeFlushedSequenceId);
289     return alreadyPut == null ? storeFlushedSequenceId : alreadyPut;
290   }
291   /**
292    * Updates last flushed sequence Ids for the regions on server sn
293    * @param sn
294    * @param hsl
295    */
296   private void updateLastFlushedSequenceIds(ServerName sn, ServerLoad hsl) {
297     Map<byte[], RegionLoad> regionsLoad = hsl.getRegionsLoad();
298     for (Entry<byte[], RegionLoad> entry : regionsLoad.entrySet()) {
299       byte[] encodedRegionName = Bytes.toBytes(HRegionInfo.encodeRegionName(entry.getKey()));
300       Long existingValue = flushedSequenceIdByRegion.get(encodedRegionName);
301       long l = entry.getValue().getCompleteSequenceId();
302       // Don't let smaller sequence ids override greater sequence ids.
303       if (LOG.isTraceEnabled()) {
304         LOG.trace(Bytes.toString(encodedRegionName) + ", existingValue=" + existingValue +
305           ", completeSequenceId=" + l);
306       }
307       if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue)) {
308         flushedSequenceIdByRegion.put(encodedRegionName, l);
309       } else if (l != HConstants.NO_SEQNUM && l < existingValue) {
310         LOG.warn("RegionServer " + sn + " indicates a last flushed sequence id ("
311             + l + ") that is less than the previous last flushed sequence id ("
312             + existingValue + ") for region " + Bytes.toString(entry.getKey()) + " Ignoring.");
313       }
314       ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
315           getOrCreateStoreFlushedSequenceId(encodedRegionName);
316       for (StoreSequenceId storeSeqId : entry.getValue().getStoreCompleteSequenceId()) {
317         byte[] family = storeSeqId.getFamilyName().toByteArray();
318         existingValue = storeFlushedSequenceId.get(family);
319         l = storeSeqId.getSequenceId();
320         if (LOG.isTraceEnabled()) {
321           LOG.trace(Bytes.toString(encodedRegionName) + ", family=" + Bytes.toString(family) +
322             ", existingValue=" + existingValue + ", completeSequenceId=" + l);
323         }
324         // Don't let smaller sequence ids override greater sequence ids.
325         if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue.longValue())) {
326           storeFlushedSequenceId.put(family, l);
327         }
328       }
329     }
330   }
331 
332   void regionServerReport(ServerName sn,
333       ServerLoad sl) throws YouAreDeadException {
334     checkIsDead(sn, "REPORT");
335     if (null == this.onlineServers.replace(sn, sl)) {
336       // Already have this host+port combo and its just different start code?
337       // Just let the server in. Presume master joining a running cluster.
338       // recordNewServer is what happens at the end of reportServerStartup.
339       // The only thing we are skipping is passing back to the regionserver
340       // the ServerName to use. Here we presume a master has already done
341       // that so we'll press on with whatever it gave us for ServerName.
342       if (!checkAndRecordNewServer(sn, sl)) {
343         LOG.info("RegionServerReport ignored, could not record the server: " + sn);
344         return; // Not recorded, so no need to move on
345       }
346     }
347     updateLastFlushedSequenceIds(sn, sl);
348   }
349 
350   /**
351    * Check is a server of same host and port already exists,
352    * if not, or the existed one got a smaller start code, record it.
353    *
354    * @param serverName the server to check and record
355    * @param sl the server load on the server
356    * @return true if the server is recorded, otherwise, false
357    */
358   boolean checkAndRecordNewServer(
359       final ServerName serverName, final ServerLoad sl) {
360     ServerName existingServer = null;
361     synchronized (this.onlineServers) {
362       existingServer = findServerWithSameHostnamePortWithLock(serverName);
363       if (existingServer != null && (existingServer.getStartcode() > serverName.getStartcode())) {
364         LOG.info("Server serverName=" + serverName + " rejected; we already have "
365             + existingServer.toString() + " registered with same hostname and port");
366         return false;
367       }
368       recordNewServerWithLock(serverName, sl);
369     }
370 
371     // Tell our listeners that a server was added
372     if (!this.listeners.isEmpty()) {
373       for (ServerListener listener : this.listeners) {
374         listener.serverAdded(serverName);
375       }
376     }
377 
378     // Note that we assume that same ts means same server, and don't expire in that case.
379     //  TODO: ts can theoretically collide due to clock shifts, so this is a bit hacky.
380     if (existingServer != null && (existingServer.getStartcode() < serverName.getStartcode())) {
381       LOG.info("Triggering server recovery; existingServer " +
382           existingServer + " looks stale, new server:" + serverName);
383       expireServer(existingServer);
384     }
385     return true;
386   }
387 
388   /**
389    * Checks if the clock skew between the server and the master. If the clock skew exceeds the
390    * configured max, it will throw an exception; if it exceeds the configured warning threshold,
391    * it will log a warning but start normally.
392    * @param serverName Incoming servers's name
393    * @param serverCurrentTime
394    * @throws ClockOutOfSyncException if the skew exceeds the configured max value
395    */
396   private void checkClockSkew(final ServerName serverName, final long serverCurrentTime)
397   throws ClockOutOfSyncException {
398     long skew = Math.abs(System.currentTimeMillis() - serverCurrentTime);
399     if (skew > maxSkew) {
400       String message = "Server " + serverName + " has been " +
401         "rejected; Reported time is too far out of sync with master.  " +
402         "Time difference of " + skew + "ms > max allowed of " + maxSkew + "ms";
403       LOG.warn(message);
404       throw new ClockOutOfSyncException(message);
405     } else if (skew > warningSkew){
406       String message = "Reported time for server " + serverName + " is out of sync with master " +
407         "by " + skew + "ms. (Warning threshold is " + warningSkew + "ms; " +
408         "error threshold is " + maxSkew + "ms)";
409       LOG.warn(message);
410     }
411   }
412 
413   /**
414    * If this server is on the dead list, reject it with a YouAreDeadException.
415    * If it was dead but came back with a new start code, remove the old entry
416    * from the dead list.
417    * @param serverName
418    * @param what START or REPORT
419    * @throws org.apache.hadoop.hbase.YouAreDeadException
420    */
421   private void checkIsDead(final ServerName serverName, final String what)
422       throws YouAreDeadException {
423     if (this.deadservers.isDeadServer(serverName)) {
424       // host name, port and start code all match with existing one of the
425       // dead servers. So, this server must be dead.
426       String message = "Server " + what + " rejected; currently processing " +
427           serverName + " as dead server";
428       LOG.debug(message);
429       throw new YouAreDeadException(message);
430     }
431     // remove dead server with same hostname and port of newly checking in rs after master
432     // initialization.See HBASE-5916 for more information.
433     if ((this.services == null || this.services.isInitialized())
434         && this.deadservers.cleanPreviousInstance(serverName)) {
435       // This server has now become alive after we marked it as dead.
436       // We removed it's previous entry from the dead list to reflect it.
437       LOG.debug(what + ":" + " Server " + serverName + " came back up," +
438           " removed it from the dead servers list");
439     }
440   }
441 
442   /**
443    * Assumes onlineServers is locked.
444    * @return ServerName with matching hostname and port.
445    */
446   private ServerName findServerWithSameHostnamePortWithLock(
447       final ServerName serverName) {
448     for (ServerName sn: this.onlineServers.keySet()) {
449       if (ServerName.isSameHostnameAndPort(serverName, sn)) return sn;
450     }
451     return null;
452   }
453 
454   /**
455    * Adds the onlineServers list. onlineServers should be locked.
456    * @param serverName The remote servers name.
457    * @param sl
458    * @return Server load from the removed server, if any.
459    */
460   @VisibleForTesting
461   void recordNewServerWithLock(final ServerName serverName, final ServerLoad sl) {
462     LOG.info("Registering server=" + serverName);
463     this.onlineServers.put(serverName, sl);
464     this.rsAdmins.remove(serverName);
465   }
466 
467   public RegionStoreSequenceIds getLastFlushedSequenceId(byte[] encodedRegionName) {
468     RegionStoreSequenceIds.Builder builder = RegionStoreSequenceIds.newBuilder();
469     Long seqId = flushedSequenceIdByRegion.get(encodedRegionName);
470     builder.setLastFlushedSequenceId(seqId != null ? seqId.longValue() : HConstants.NO_SEQNUM);
471     Map<byte[], Long> storeFlushedSequenceId =
472         storeFlushedSequenceIdsByRegion.get(encodedRegionName);
473     if (storeFlushedSequenceId != null) {
474       for (Map.Entry<byte[], Long> entry : storeFlushedSequenceId.entrySet()) {
475         builder.addStoreSequenceId(StoreSequenceId.newBuilder()
476             .setFamilyName(ByteString.copyFrom(entry.getKey()))
477             .setSequenceId(entry.getValue().longValue()).build());
478       }
479     }
480     return builder.build();
481   }
482 
483   /**
484    * @param serverName
485    * @return ServerLoad if serverName is known else null
486    */
487   public ServerLoad getLoad(final ServerName serverName) {
488     return this.onlineServers.get(serverName);
489   }
490 
491   /**
492    * Compute the average load across all region servers.
493    * Currently, this uses a very naive computation - just uses the number of
494    * regions being served, ignoring stats about number of requests.
495    * @return the average load
496    */
497   public double getAverageLoad() {
498     int totalLoad = 0;
499     int numServers = 0;
500     for (ServerLoad sl: this.onlineServers.values()) {
501         numServers++;
502         totalLoad += sl.getNumberOfRegions();
503     }
504     return numServers == 0 ? 0 :
505       (double)totalLoad / (double)numServers;
506   }
507 
508   /** @return the count of active regionservers */
509   public int countOfRegionServers() {
510     // Presumes onlineServers is a concurrent map
511     return this.onlineServers.size();
512   }
513 
514   /**
515    * @return Read-only map of servers to serverinfo
516    */
517   public Map<ServerName, ServerLoad> getOnlineServers() {
518     // Presumption is that iterating the returned Map is OK.
519     synchronized (this.onlineServers) {
520       return Collections.unmodifiableMap(this.onlineServers);
521     }
522   }
523 
524 
525   public DeadServer getDeadServers() {
526     return this.deadservers;
527   }
528 
529   /**
530    * Checks if any dead servers are currently in progress.
531    * @return true if any RS are being processed as dead, false if not
532    */
533   public boolean areDeadServersInProgress() {
534     return this.deadservers.areDeadServersInProgress();
535   }
536 
537   void letRegionServersShutdown() {
538     long previousLogTime = 0;
539     ServerName sn = master.getServerName();
540     ZooKeeperWatcher zkw = master.getZooKeeper();
541     int onlineServersCt;
542     while ((onlineServersCt = onlineServers.size()) > 0){
543 
544       if (System.currentTimeMillis() > (previousLogTime + 1000)) {
545         Set<ServerName> remainingServers = onlineServers.keySet();
546         synchronized (onlineServers) {
547           if (remainingServers.size() == 1 && remainingServers.contains(sn)) {
548             // Master will delete itself later.
549             return;
550           }
551         }
552         StringBuilder sb = new StringBuilder();
553         // It's ok here to not sync on onlineServers - merely logging
554         for (ServerName key : remainingServers) {
555           if (sb.length() > 0) {
556             sb.append(", ");
557           }
558           sb.append(key);
559         }
560         LOG.info("Waiting on regionserver(s) to go down " + sb.toString());
561         previousLogTime = System.currentTimeMillis();
562       }
563 
564       try {
565         List<String> servers = ZKUtil.listChildrenNoWatch(zkw, zkw.rsZNode);
566         if (servers == null || servers.size() == 0 || (servers.size() == 1
567             && servers.contains(sn.toString()))) {
568           LOG.info("ZK shows there is only the master self online, exiting now");
569           // Master could have lost some ZK events, no need to wait more.
570           break;
571         }
572       } catch (KeeperException ke) {
573         LOG.warn("Failed to list regionservers", ke);
574         // ZK is malfunctioning, don't hang here
575         break;
576       }
577       synchronized (onlineServers) {
578         try {
579           if (onlineServersCt == onlineServers.size()) onlineServers.wait(100);
580         } catch (InterruptedException ignored) {
581           // continue
582         }
583       }
584     }
585   }
586 
587   /*
588    * Expire the passed server.  Add it to list of dead servers and queue a
589    * shutdown processing.
590    */
591   public synchronized void expireServer(final ServerName serverName) {
592     if (serverName.equals(master.getServerName())) {
593       if (!(master.isAborted() || master.isStopped())) {
594         master.stop("We lost our znode?");
595       }
596       return;
597     }
598     if (!services.isServerCrashProcessingEnabled()) {
599       LOG.info("Master doesn't enable ServerShutdownHandler during initialization, "
600           + "delay expiring server " + serverName);
601       this.queuedDeadServers.add(serverName);
602       return;
603     }
604     if (this.deadservers.isDeadServer(serverName)) {
605       // TODO: Can this happen?  It shouldn't be online in this case?
606       LOG.warn("Expiration of " + serverName +
607           " but server shutdown already in progress");
608       return;
609     }
610     moveFromOnelineToDeadServers(serverName);
611 
612     // If cluster is going down, yes, servers are going to be expiring; don't
613     // process as a dead server
614     if (this.clusterShutdown) {
615       LOG.info("Cluster shutdown set; " + serverName +
616         " expired; onlineServers=" + this.onlineServers.size());
617       if (this.onlineServers.isEmpty()) {
618         master.stop("Cluster shutdown set; onlineServer=0");
619       }
620       return;
621     }
622 
623     boolean carryingMeta = services.getAssignmentManager().isCarryingMeta(serverName);
624     this.services.getMasterProcedureExecutor().
625       submitProcedure(new ServerCrashProcedure(serverName, true, carryingMeta));
626     LOG.debug("Added=" + serverName +
627       " to dead servers, submitted shutdown handler to be executed meta=" + carryingMeta);
628 
629     // Tell our listeners that a server was removed
630     if (!this.listeners.isEmpty()) {
631       for (ServerListener listener : this.listeners) {
632         listener.serverRemoved(serverName);
633       }
634     }
635   }
636 
637   @VisibleForTesting
638   public void moveFromOnelineToDeadServers(final ServerName sn) {
639     synchronized (onlineServers) {
640       if (!this.onlineServers.containsKey(sn)) {
641         LOG.warn("Expiration of " + sn + " but server not online");
642       }
643       // Remove the server from the known servers lists and update load info BUT
644       // add to deadservers first; do this so it'll show in dead servers list if
645       // not in online servers list.
646       this.deadservers.add(sn);
647       this.onlineServers.remove(sn);
648       onlineServers.notifyAll();
649     }
650     this.rsAdmins.remove(sn);
651   }
652 
653   public synchronized void processDeadServer(final ServerName serverName, boolean shouldSplitWal) {
654     // When assignment manager is cleaning up the zookeeper nodes and rebuilding the
655     // in-memory region states, region servers could be down. Meta table can and
656     // should be re-assigned, log splitting can be done too. However, it is better to
657     // wait till the cleanup is done before re-assigning user regions.
658     //
659     // We should not wait in the server shutdown handler thread since it can clog
660     // the handler threads and meta table could not be re-assigned in case
661     // the corresponding server is down. So we queue them up here instead.
662     if (!services.getAssignmentManager().isFailoverCleanupDone()) {
663       requeuedDeadServers.put(serverName, shouldSplitWal);
664       return;
665     }
666 
667     this.deadservers.add(serverName);
668     this.services.getMasterProcedureExecutor().
669     submitProcedure(new ServerCrashProcedure(serverName, shouldSplitWal, false));
670   }
671 
672   /**
673    * Process the servers which died during master's initialization. It will be
674    * called after HMaster#assignMeta and AssignmentManager#joinCluster.
675    * */
676   synchronized void processQueuedDeadServers() {
677     if (!services.isServerCrashProcessingEnabled()) {
678       LOG.info("Master hasn't enabled ServerShutdownHandler");
679     }
680     Iterator<ServerName> serverIterator = queuedDeadServers.iterator();
681     while (serverIterator.hasNext()) {
682       ServerName tmpServerName = serverIterator.next();
683       expireServer(tmpServerName);
684       serverIterator.remove();
685       requeuedDeadServers.remove(tmpServerName);
686     }
687 
688     if (!services.getAssignmentManager().isFailoverCleanupDone()) {
689       LOG.info("AssignmentManager hasn't finished failover cleanup; waiting");
690     }
691 
692     for (Map.Entry<ServerName, Boolean> entry : requeuedDeadServers.entrySet()) {
693       processDeadServer(entry.getKey(), entry.getValue());
694     }
695     requeuedDeadServers.clear();
696   }
697 
698   /*
699    * Remove the server from the drain list.
700    */
701   public boolean removeServerFromDrainList(final ServerName sn) {
702     // Warn if the server (sn) is not online.  ServerName is of the form:
703     // <hostname> , <port> , <startcode>
704 
705     if (!this.isServerOnline(sn)) {
706       LOG.warn("Server " + sn + " is not currently online. " +
707                "Removing from draining list anyway, as requested.");
708     }
709     // Remove the server from the draining servers lists.
710     return this.drainingServers.remove(sn);
711   }
712 
713   /*
714    * Add the server to the drain list.
715    */
716   public boolean addServerToDrainList(final ServerName sn) {
717     // Warn if the server (sn) is not online.  ServerName is of the form:
718     // <hostname> , <port> , <startcode>
719 
720     if (!this.isServerOnline(sn)) {
721       LOG.warn("Server " + sn + " is not currently online. " +
722                "Ignoring request to add it to draining list.");
723       return false;
724     }
725     // Add the server to the draining servers lists, if it's not already in
726     // it.
727     if (this.drainingServers.contains(sn)) {
728       LOG.warn("Server " + sn + " is already in the draining server list." +
729                "Ignoring request to add it again.");
730       return false;
731     }
732     return this.drainingServers.add(sn);
733   }
734 
735   // RPC methods to region servers
736 
737   /**
738    * Sends an OPEN RPC to the specified server to open the specified region.
739    * <p>
740    * Open should not fail but can if server just crashed.
741    * <p>
742    * @param server server to open a region
743    * @param region region to open
744    * @param favoredNodes
745    */
746   public RegionOpeningState sendRegionOpen(final ServerName server,
747       HRegionInfo region, List<ServerName> favoredNodes)
748   throws IOException {
749     AdminService.BlockingInterface admin = getRsAdmin(server);
750     if (admin == null) {
751       throw new IOException("Attempting to send OPEN RPC to server " + server.toString() +
752         " failed because no RPC connection found to this server");
753     }
754     OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(server,
755       region, favoredNodes,
756       (RecoveryMode.LOG_REPLAY == this.services.getMasterWalManager().getLogRecoveryMode()));
757     try {
758       OpenRegionResponse response = admin.openRegion(null, request);
759       return ResponseConverter.getRegionOpeningState(response);
760     } catch (ServiceException se) {
761       throw ProtobufUtil.getRemoteException(se);
762     }
763   }
764 
765   /**
766    * Sends an OPEN RPC to the specified server to open the specified region.
767    * <p>
768    * Open should not fail but can if server just crashed.
769    * <p>
770    * @param server server to open a region
771    * @param regionOpenInfos info of a list of regions to open
772    * @return a list of region opening states
773    */
774   public List<RegionOpeningState> sendRegionOpen(ServerName server,
775       List<Pair<HRegionInfo, List<ServerName>>> regionOpenInfos)
776   throws IOException {
777     AdminService.BlockingInterface admin = getRsAdmin(server);
778     if (admin == null) {
779       throw new IOException("Attempting to send OPEN RPC to server " + server.toString() +
780         " failed because no RPC connection found to this server");
781     }
782 
783     OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(server, regionOpenInfos,
784       (RecoveryMode.LOG_REPLAY == this.services.getMasterWalManager().getLogRecoveryMode()));
785     try {
786       OpenRegionResponse response = admin.openRegion(null, request);
787       return ResponseConverter.getRegionOpeningStateList(response);
788     } catch (ServiceException se) {
789       throw ProtobufUtil.getRemoteException(se);
790     }
791   }
792 
793   private PayloadCarryingRpcController newRpcController() {
794     return rpcControllerFactory == null ? null : rpcControllerFactory.newController();
795   }
796 
797   /**
798    * Sends an CLOSE RPC to the specified server to close the specified region.
799    * <p>
800    * A region server could reject the close request because it either does not
801    * have the specified region or the region is being split.
802    * @param server server to open a region
803    * @param region region to open
804    * @param dest - if the region is moved to another server, the destination server. null otherwise.
805    * @throws IOException
806    */
807   public boolean sendRegionClose(ServerName server, HRegionInfo region,
808       ServerName dest) throws IOException {
809     if (server == null) throw new NullPointerException("Passed server is null");
810     AdminService.BlockingInterface admin = getRsAdmin(server);
811     if (admin == null) {
812       throw new IOException("Attempting to send CLOSE RPC to server " +
813         server.toString() + " for region " +
814         region.getRegionNameAsString() +
815         " failed because no RPC connection found to this server");
816     }
817     PayloadCarryingRpcController controller = newRpcController();
818     return ProtobufUtil.closeRegion(controller, admin, server, region.getRegionName(), dest);
819   }
820 
821   public boolean sendRegionClose(ServerName server,
822       HRegionInfo region) throws IOException {
823     return sendRegionClose(server, region, null);
824   }
825 
826   /**
827    * Sends a WARMUP RPC to the specified server to warmup the specified region.
828    * <p>
829    * A region server could reject the close request because it either does not
830    * have the specified region or the region is being split.
831    * @param server server to warmup a region
832    * @param region region to  warmup
833    */
834   public void sendRegionWarmup(ServerName server,
835       HRegionInfo region) {
836     if (server == null) return;
837     try {
838       AdminService.BlockingInterface admin = getRsAdmin(server);
839       PayloadCarryingRpcController controller = newRpcController();
840       ProtobufUtil.warmupRegion(controller, admin, region);
841     } catch (IOException e) {
842       LOG.error("Received exception in RPC for warmup server:" +
843         server + "region: " + region +
844         "exception: " + e);
845     }
846   }
847 
848   /**
849    * Contacts a region server and waits up to timeout ms
850    * to close the region.  This bypasses the active hmaster.
851    */
852   public static void closeRegionSilentlyAndWait(ClusterConnection connection,
853     ServerName server, HRegionInfo region, long timeout) throws IOException, InterruptedException {
854     AdminService.BlockingInterface rs = connection.getAdmin(server);
855     PayloadCarryingRpcController controller = connection.getRpcControllerFactory().newController();
856     try {
857       ProtobufUtil.closeRegion(controller, rs, server, region.getRegionName());
858     } catch (IOException e) {
859       LOG.warn("Exception when closing region: " + region.getRegionNameAsString(), e);
860     }
861     long expiration = timeout + System.currentTimeMillis();
862     while (System.currentTimeMillis() < expiration) {
863       try {
864         HRegionInfo rsRegion =
865           ProtobufUtil.getRegionInfo(controller, rs, region.getRegionName());
866         if (rsRegion == null) return;
867       } catch (IOException ioe) {
868         if (ioe instanceof NotServingRegionException) // no need to retry again
869           return;
870         LOG.warn("Exception when retrieving regioninfo from: "
871           + region.getRegionNameAsString(), ioe);
872       }
873       Thread.sleep(1000);
874     }
875     throw new IOException("Region " + region + " failed to close within"
876         + " timeout " + timeout);
877   }
878 
879   /**
880    * Sends an MERGE REGIONS RPC to the specified server to merge the specified
881    * regions.
882    * <p>
883    * A region server could reject the close request because it either does not
884    * have the specified region.
885    * @param server server to merge regions
886    * @param region_a region to merge
887    * @param region_b region to merge
888    * @param forcible true if do a compulsory merge, otherwise we will only merge
889    *          two adjacent regions
890    * @throws IOException
891    */
892   public void sendRegionsMerge(ServerName server, HRegionInfo region_a,
893       HRegionInfo region_b, boolean forcible, User user) throws IOException {
894     if (server == null)
895       throw new NullPointerException("Passed server is null");
896     if (region_a == null || region_b == null)
897       throw new NullPointerException("Passed region is null");
898     AdminService.BlockingInterface admin = getRsAdmin(server);
899     if (admin == null) {
900       throw new IOException("Attempting to send MERGE REGIONS RPC to server "
901           + server.toString() + " for region "
902           + region_a.getRegionNameAsString() + ","
903           + region_b.getRegionNameAsString()
904           + " failed because no RPC connection found to this server");
905     }
906     PayloadCarryingRpcController controller = newRpcController();
907     ProtobufUtil.mergeRegions(controller, admin, region_a, region_b, forcible, user);
908   }
909 
910   /**
911    * Check if a region server is reachable and has the expected start code
912    */
913   public boolean isServerReachable(ServerName server) {
914     if (server == null) throw new NullPointerException("Passed server is null");
915 
916 
917     RetryCounter retryCounter = pingRetryCounterFactory.create();
918     while (retryCounter.shouldRetry()) {
919       try {
920         PayloadCarryingRpcController controller = newRpcController();
921         AdminService.BlockingInterface admin = getRsAdmin(server);
922         if (admin != null) {
923           ServerInfo info = ProtobufUtil.getServerInfo(controller, admin);
924           return info != null && info.hasServerName()
925             && server.getStartcode() == info.getServerName().getStartCode();
926         }
927       } catch (IOException ioe) {
928         LOG.debug("Couldn't reach " + server + ", try=" + retryCounter.getAttemptTimes()
929           + " of " + retryCounter.getMaxAttempts(), ioe);
930         try {
931           retryCounter.sleepUntilNextRetry();
932         } catch(InterruptedException ie) {
933           Thread.currentThread().interrupt();
934         }
935       }
936     }
937     return false;
938   }
939 
940     /**
941     * @param sn
942     * @return Admin interface for the remote regionserver named <code>sn</code>
943     * @throws IOException
944     * @throws RetriesExhaustedException wrapping a ConnectException if failed
945     */
946   private AdminService.BlockingInterface getRsAdmin(final ServerName sn)
947   throws IOException {
948     AdminService.BlockingInterface admin = this.rsAdmins.get(sn);
949     if (admin == null) {
950       LOG.debug("New admin connection to " + sn.toString());
951       if (sn.equals(master.getServerName()) && master instanceof HRegionServer) {
952         // A master is also a region server now, see HBASE-10569 for details
953         admin = ((HRegionServer)master).getRSRpcServices();
954       } else {
955         admin = this.connection.getAdmin(sn);
956       }
957       this.rsAdmins.put(sn, admin);
958     }
959     return admin;
960   }
961 
962   /**
963    * Wait for the region servers to report in.
964    * We will wait until one of this condition is met:
965    *  - the master is stopped
966    *  - the 'hbase.master.wait.on.regionservers.maxtostart' number of
967    *    region servers is reached
968    *  - the 'hbase.master.wait.on.regionservers.mintostart' is reached AND
969    *   there have been no new region server in for
970    *      'hbase.master.wait.on.regionservers.interval' time AND
971    *   the 'hbase.master.wait.on.regionservers.timeout' is reached
972    *
973    * @throws InterruptedException
974    */
975   public void waitForRegionServers(MonitoredTask status)
976   throws InterruptedException {
977     final long interval = this.master.getConfiguration().
978       getLong(WAIT_ON_REGIONSERVERS_INTERVAL, 1500);
979     final long timeout = this.master.getConfiguration().
980       getLong(WAIT_ON_REGIONSERVERS_TIMEOUT, 4500);
981     int defaultMinToStart = 1;
982     if (BaseLoadBalancer.tablesOnMaster(master.getConfiguration())) {
983       // If we assign regions to master, we'd like to start
984       // at least another region server so that we don't
985       // assign all regions to master if other region servers
986       // don't come up in time.
987       defaultMinToStart = 2;
988     }
989     int minToStart = this.master.getConfiguration().
990       getInt(WAIT_ON_REGIONSERVERS_MINTOSTART, defaultMinToStart);
991     if (minToStart < 1) {
992       LOG.warn(String.format(
993         "The value of '%s' (%d) can not be less than 1, ignoring.",
994         WAIT_ON_REGIONSERVERS_MINTOSTART, minToStart));
995       minToStart = 1;
996     }
997     int maxToStart = this.master.getConfiguration().
998       getInt(WAIT_ON_REGIONSERVERS_MAXTOSTART, Integer.MAX_VALUE);
999     if (maxToStart < minToStart) {
1000         LOG.warn(String.format(
1001             "The value of '%s' (%d) is set less than '%s' (%d), ignoring.",
1002             WAIT_ON_REGIONSERVERS_MAXTOSTART, maxToStart,
1003             WAIT_ON_REGIONSERVERS_MINTOSTART, minToStart));
1004         maxToStart = Integer.MAX_VALUE;
1005     }
1006 
1007     long now =  System.currentTimeMillis();
1008     final long startTime = now;
1009     long slept = 0;
1010     long lastLogTime = 0;
1011     long lastCountChange = startTime;
1012     int count = countOfRegionServers();
1013     int oldCount = 0;
1014     while (!this.master.isStopped() && count < maxToStart
1015         && (lastCountChange+interval > now || timeout > slept || count < minToStart)) {
1016       // Log some info at every interval time or if there is a change
1017       if (oldCount != count || lastLogTime+interval < now){
1018         lastLogTime = now;
1019         String msg =
1020           "Waiting for region servers count to settle; currently"+
1021             " checked in " + count + ", slept for " + slept + " ms," +
1022             " expecting minimum of " + minToStart + ", maximum of "+ maxToStart+
1023             ", timeout of "+timeout+" ms, interval of "+interval+" ms.";
1024         LOG.info(msg);
1025         status.setStatus(msg);
1026       }
1027 
1028       // We sleep for some time
1029       final long sleepTime = 50;
1030       Thread.sleep(sleepTime);
1031       now =  System.currentTimeMillis();
1032       slept = now - startTime;
1033 
1034       oldCount = count;
1035       count = countOfRegionServers();
1036       if (count != oldCount) {
1037         lastCountChange = now;
1038       }
1039     }
1040 
1041     LOG.info("Finished waiting for region servers count to settle;" +
1042       " checked in " + count + ", slept for " + slept + " ms," +
1043       " expecting minimum of " + minToStart + ", maximum of "+ maxToStart+","+
1044       " master is "+ (this.master.isStopped() ? "stopped.": "running")
1045     );
1046   }
1047 
1048   /**
1049    * @return A copy of the internal list of online servers.
1050    */
1051   public List<ServerName> getOnlineServersList() {
1052     // TODO: optimize the load balancer call so we don't need to make a new list
1053     // TODO: FIX. THIS IS POPULAR CALL.
1054     return new ArrayList<ServerName>(this.onlineServers.keySet());
1055   }
1056 
1057   /**
1058    * @return A copy of the internal list of draining servers.
1059    */
1060   public List<ServerName> getDrainingServersList() {
1061     return new ArrayList<ServerName>(this.drainingServers);
1062   }
1063 
1064   /**
1065    * @return A copy of the internal set of deadNotExpired servers.
1066    */
1067   Set<ServerName> getDeadNotExpiredServers() {
1068     return new HashSet<ServerName>(this.queuedDeadServers);
1069   }
1070 
1071   /**
1072    * During startup, if we figure it is not a failover, i.e. there is
1073    * no more WAL files to split, we won't try to recover these dead servers.
1074    * So we just remove them from the queue. Use caution in calling this.
1075    */
1076   void removeRequeuedDeadServers() {
1077     requeuedDeadServers.clear();
1078   }
1079 
1080   /**
1081    * @return A copy of the internal map of requeuedDeadServers servers and their corresponding
1082    *         splitlog need flag.
1083    */
1084   Map<ServerName, Boolean> getRequeuedDeadServers() {
1085     return Collections.unmodifiableMap(this.requeuedDeadServers);
1086   }
1087 
1088   public boolean isServerOnline(ServerName serverName) {
1089     return serverName != null && onlineServers.containsKey(serverName);
1090   }
1091 
1092   /**
1093    * Check if a server is known to be dead.  A server can be online,
1094    * or known to be dead, or unknown to this manager (i.e, not online,
1095    * not known to be dead either. it is simply not tracked by the
1096    * master any more, for example, a very old previous instance).
1097    */
1098   public synchronized boolean isServerDead(ServerName serverName) {
1099     return serverName == null || deadservers.isDeadServer(serverName)
1100       || queuedDeadServers.contains(serverName)
1101       || requeuedDeadServers.containsKey(serverName);
1102   }
1103 
1104   public void shutdownCluster() {
1105     this.clusterShutdown = true;
1106     this.master.stop("Cluster shutdown requested");
1107   }
1108 
1109   public boolean isClusterShutdown() {
1110     return this.clusterShutdown;
1111   }
1112 
1113   /**
1114    * Stop the ServerManager.  Currently closes the connection to the master.
1115    */
1116   public void stop() {
1117     if (connection != null) {
1118       try {
1119         connection.close();
1120       } catch (IOException e) {
1121         LOG.error("Attempt to close connection to master failed", e);
1122       }
1123     }
1124   }
1125 
1126   /**
1127    * Creates a list of possible destinations for a region. It contains the online servers, but not
1128    *  the draining or dying servers.
1129    *  @param serverToExclude can be null if there is no server to exclude
1130    */
1131   public List<ServerName> createDestinationServersList(final ServerName serverToExclude){
1132     final List<ServerName> destServers = getOnlineServersList();
1133 
1134     if (serverToExclude != null){
1135       destServers.remove(serverToExclude);
1136     }
1137 
1138     // Loop through the draining server list and remove them from the server list
1139     final List<ServerName> drainingServersCopy = getDrainingServersList();
1140     if (!drainingServersCopy.isEmpty()) {
1141       for (final ServerName server: drainingServersCopy) {
1142         destServers.remove(server);
1143       }
1144     }
1145 
1146     // Remove the deadNotExpired servers from the server list.
1147     removeDeadNotExpiredServers(destServers);
1148     return destServers;
1149   }
1150 
1151   /**
1152    * Calls {@link #createDestinationServersList} without server to exclude.
1153    */
1154   public List<ServerName> createDestinationServersList(){
1155     return createDestinationServersList(null);
1156   }
1157 
1158     /**
1159     * Loop through the deadNotExpired server list and remove them from the
1160     * servers.
1161     * This function should be used carefully outside of this class. You should use a high level
1162     *  method such as {@link #createDestinationServersList()} instead of managing you own list.
1163     */
1164   void removeDeadNotExpiredServers(List<ServerName> servers) {
1165     Set<ServerName> deadNotExpiredServersCopy = this.getDeadNotExpiredServers();
1166     if (!deadNotExpiredServersCopy.isEmpty()) {
1167       for (ServerName server : deadNotExpiredServersCopy) {
1168         LOG.debug("Removing dead but not expired server: " + server
1169           + " from eligible server pool.");
1170         servers.remove(server);
1171       }
1172     }
1173   }
1174 
1175   /**
1176    * To clear any dead server with same host name and port of any online server
1177    */
1178   void clearDeadServersWithSameHostNameAndPortOfOnlineServer() {
1179     for (ServerName serverName : getOnlineServersList()) {
1180       deadservers.cleanAllPreviousInstances(serverName);
1181     }
1182   }
1183 }