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