1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.master;
20
21 import com.google.common.annotations.VisibleForTesting;
22 import com.google.protobuf.ByteString;
23 import com.google.protobuf.ServiceException;
24
25 import java.io.IOException;
26 import java.net.ConnectException;
27 import java.net.InetAddress;
28 import java.util.ArrayList;
29 import java.util.Collections;
30 import java.util.HashMap;
31 import java.util.HashSet;
32 import java.util.Iterator;
33 import java.util.List;
34 import java.util.Map;
35 import java.util.Map.Entry;
36 import java.util.Set;
37 import java.util.concurrent.ConcurrentHashMap;
38 import java.util.concurrent.ConcurrentNavigableMap;
39 import java.util.concurrent.ConcurrentSkipListMap;
40 import java.util.concurrent.CopyOnWriteArrayList;
41 import org.apache.commons.logging.Log;
42 import org.apache.commons.logging.LogFactory;
43 import org.apache.hadoop.conf.Configuration;
44 import org.apache.hadoop.hbase.ClockOutOfSyncException;
45 import org.apache.hadoop.hbase.HConstants;
46 import org.apache.hadoop.hbase.HRegionInfo;
47 import org.apache.hadoop.hbase.NotServingRegionException;
48 import org.apache.hadoop.hbase.RegionLoad;
49 import org.apache.hadoop.hbase.Server;
50 import org.apache.hadoop.hbase.ServerLoad;
51 import org.apache.hadoop.hbase.ServerName;
52 import org.apache.hadoop.hbase.YouAreDeadException;
53 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
54 import org.apache.hadoop.hbase.classification.InterfaceAudience;
55 import org.apache.hadoop.hbase.client.ClusterConnection;
56 import org.apache.hadoop.hbase.client.ConnectionFactory;
57 import org.apache.hadoop.hbase.client.RetriesExhaustedException;
58 import org.apache.hadoop.hbase.ipc.FailedServerException;
59 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
60 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
61 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
62 import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
63 import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
64 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
65 import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
66 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
67 import org.apache.hadoop.hbase.protobuf.RequestConverter;
68 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
69 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
70 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest;
71 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse;
72 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ServerInfo;
73 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds;
74 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.StoreSequenceId;
75 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest;
76 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
77 import org.apache.hadoop.hbase.regionserver.HRegionServer;
78 import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
79 import org.apache.hadoop.hbase.security.User;
80 import org.apache.hadoop.hbase.util.Bytes;
81 import org.apache.hadoop.hbase.util.RetryCounter;
82 import org.apache.hadoop.hbase.util.RetryCounterFactory;
83 import org.apache.hadoop.hbase.util.Triple;
84 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
85 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
86 import org.apache.zookeeper.KeeperException;
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110 @InterfaceAudience.Private
111 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="JLM_JSR166_UTILCONCURRENT_MONITORENTER",
112 justification="Synchronization on concurrent map is intended")
113 public class ServerManager {
114 public static final String WAIT_ON_REGIONSERVERS_MAXTOSTART =
115 "hbase.master.wait.on.regionservers.maxtostart";
116
117 public static final String WAIT_ON_REGIONSERVERS_MINTOSTART =
118 "hbase.master.wait.on.regionservers.mintostart";
119
120 public static final String WAIT_ON_REGIONSERVERS_TIMEOUT =
121 "hbase.master.wait.on.regionservers.timeout";
122
123 public static final String WAIT_ON_REGIONSERVERS_INTERVAL =
124 "hbase.master.wait.on.regionservers.interval";
125
126 private static final Log LOG = LogFactory.getLog(ServerManager.class);
127
128
129 private volatile boolean clusterShutdown = false;
130
131
132
133
134 private final ConcurrentNavigableMap<byte[], Long> flushedSequenceIdByRegion =
135 new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
136
137
138
139
140 private final ConcurrentNavigableMap<byte[], ConcurrentNavigableMap<byte[], Long>>
141 storeFlushedSequenceIdsByRegion =
142 new ConcurrentSkipListMap<byte[], ConcurrentNavigableMap<byte[], Long>>(Bytes.BYTES_COMPARATOR);
143
144
145 private final ConcurrentNavigableMap<ServerName, ServerLoad> onlineServers =
146 new ConcurrentSkipListMap<ServerName, ServerLoad>();
147
148
149
150
151
152 private final Map<ServerName, AdminService.BlockingInterface> rsAdmins =
153 new HashMap<ServerName, AdminService.BlockingInterface>();
154
155
156
157
158
159 private final ArrayList<ServerName> drainingServers =
160 new ArrayList<ServerName>();
161
162 private final Server master;
163 private final MasterServices services;
164 private final ClusterConnection connection;
165
166 private final DeadServer deadservers = new DeadServer();
167
168 private final long maxSkew;
169 private final long warningSkew;
170
171 private final RetryCounterFactory pingRetryCounterFactory;
172 private final RpcControllerFactory rpcControllerFactory;
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190 private Set<ServerName> queuedDeadServers = new HashSet<ServerName>();
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207 private Map<ServerName, Boolean> requeuedDeadServers
208 = new ConcurrentHashMap<ServerName, Boolean>();
209
210
211 private List<ServerListener> listeners = new CopyOnWriteArrayList<ServerListener>();
212
213
214
215
216
217
218
219 public ServerManager(final Server master, final MasterServices services)
220 throws IOException {
221 this(master, services, true);
222 }
223
224 ServerManager(final Server master, final MasterServices services,
225 final boolean connect) throws IOException {
226 this.master = master;
227 this.services = services;
228 Configuration c = master.getConfiguration();
229 maxSkew = c.getLong("hbase.master.maxclockskew", 30000);
230 warningSkew = c.getLong("hbase.master.warningclockskew", 10000);
231 this.connection = connect ? (ClusterConnection)ConnectionFactory.createConnection(c) : null;
232 int pingMaxAttempts = Math.max(1, master.getConfiguration().getInt(
233 "hbase.master.maximum.ping.server.attempts", 10));
234 int pingSleepInterval = Math.max(1, master.getConfiguration().getInt(
235 "hbase.master.ping.server.retry.sleep.interval", 100));
236 this.pingRetryCounterFactory = new RetryCounterFactory(pingMaxAttempts, pingSleepInterval);
237 this.rpcControllerFactory = this.connection == null
238 ? null
239 : connection.getRpcControllerFactory();
240 }
241
242
243
244
245
246 public void registerListener(final ServerListener listener) {
247 this.listeners.add(listener);
248 }
249
250
251
252
253
254 public boolean unregisterListener(final ServerListener listener) {
255 return this.listeners.remove(listener);
256 }
257
258
259
260
261
262
263
264
265 ServerName regionServerStartup(RegionServerStartupRequest request, InetAddress ia)
266 throws IOException {
267
268
269
270
271
272
273
274
275 final String hostname = request.hasUseThisHostnameInstead() ?
276 request.getUseThisHostnameInstead() :ia.getHostName();
277 ServerName sn = ServerName.valueOf(hostname, request.getPort(),
278 request.getServerStartCode());
279 checkClockSkew(sn, request.getServerCurrentTime());
280 checkIsDead(sn, "STARTUP");
281 if (!checkAndRecordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD)) {
282 LOG.warn("THIS SHOULD NOT HAPPEN, RegionServerStartup"
283 + " could not record the server: " + sn);
284 }
285 return sn;
286 }
287
288 private ConcurrentNavigableMap<byte[], Long> getOrCreateStoreFlushedSequenceId(
289 byte[] regionName) {
290 ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
291 storeFlushedSequenceIdsByRegion.get(regionName);
292 if (storeFlushedSequenceId != null) {
293 return storeFlushedSequenceId;
294 }
295 storeFlushedSequenceId = new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
296 ConcurrentNavigableMap<byte[], Long> alreadyPut =
297 storeFlushedSequenceIdsByRegion.putIfAbsent(regionName, storeFlushedSequenceId);
298 return alreadyPut == null ? storeFlushedSequenceId : alreadyPut;
299 }
300
301
302
303
304
305 private void updateLastFlushedSequenceIds(ServerName sn, ServerLoad hsl) {
306 Map<byte[], RegionLoad> regionsLoad = hsl.getRegionsLoad();
307 for (Entry<byte[], RegionLoad> entry : regionsLoad.entrySet()) {
308 byte[] encodedRegionName = Bytes.toBytes(HRegionInfo.encodeRegionName(entry.getKey()));
309 Long existingValue = flushedSequenceIdByRegion.get(encodedRegionName);
310 long l = entry.getValue().getCompleteSequenceId();
311
312 if (LOG.isTraceEnabled()) {
313 LOG.trace(Bytes.toString(encodedRegionName) + ", existingValue=" + existingValue +
314 ", completeSequenceId=" + l);
315 }
316 if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue)) {
317 flushedSequenceIdByRegion.put(encodedRegionName, l);
318 } else if (l != HConstants.NO_SEQNUM && l < existingValue) {
319 LOG.warn("RegionServer " + sn + " indicates a last flushed sequence id ("
320 + l + ") that is less than the previous last flushed sequence id ("
321 + existingValue + ") for region " + Bytes.toString(entry.getKey()) + " Ignoring.");
322 }
323 ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
324 getOrCreateStoreFlushedSequenceId(encodedRegionName);
325 for (StoreSequenceId storeSeqId : entry.getValue().getStoreCompleteSequenceId()) {
326 byte[] family = storeSeqId.getFamilyName().toByteArray();
327 existingValue = storeFlushedSequenceId.get(family);
328 l = storeSeqId.getSequenceId();
329 if (LOG.isTraceEnabled()) {
330 LOG.trace(Bytes.toString(encodedRegionName) + ", family=" + Bytes.toString(family) +
331 ", existingValue=" + existingValue + ", completeSequenceId=" + l);
332 }
333
334 if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue.longValue())) {
335 storeFlushedSequenceId.put(family, l);
336 }
337 }
338 }
339 }
340
341 void regionServerReport(ServerName sn,
342 ServerLoad sl) throws YouAreDeadException {
343 checkIsDead(sn, "REPORT");
344 if (null == this.onlineServers.replace(sn, sl)) {
345
346
347
348
349
350
351 if (!checkAndRecordNewServer(sn, sl)) {
352 LOG.info("RegionServerReport ignored, could not record the server: " + sn);
353 return;
354 }
355 }
356 updateLastFlushedSequenceIds(sn, sl);
357 }
358
359
360
361
362
363
364
365
366
367 boolean checkAndRecordNewServer(
368 final ServerName serverName, final ServerLoad sl) {
369 ServerName existingServer = null;
370 synchronized (this.onlineServers) {
371 existingServer = findServerWithSameHostnamePortWithLock(serverName);
372 if (existingServer != null && (existingServer.getStartcode() > serverName.getStartcode())) {
373 LOG.info("Server serverName=" + serverName + " rejected; we already have "
374 + existingServer.toString() + " registered with same hostname and port");
375 return false;
376 }
377 recordNewServerWithLock(serverName, sl);
378 }
379
380
381 if (!this.listeners.isEmpty()) {
382 for (ServerListener listener : this.listeners) {
383 listener.serverAdded(serverName);
384 }
385 }
386
387
388
389 if (existingServer != null && (existingServer.getStartcode() < serverName.getStartcode())) {
390 LOG.info("Triggering server recovery; existingServer " +
391 existingServer + " looks stale, new server:" + serverName);
392 expireServer(existingServer);
393 }
394 return true;
395 }
396
397
398
399
400
401
402
403
404
405 private void checkClockSkew(final ServerName serverName, final long serverCurrentTime)
406 throws ClockOutOfSyncException {
407 long skew = Math.abs(System.currentTimeMillis() - serverCurrentTime);
408 if (skew > maxSkew) {
409 String message = "Server " + serverName + " has been " +
410 "rejected; Reported time is too far out of sync with master. " +
411 "Time difference of " + skew + "ms > max allowed of " + maxSkew + "ms";
412 LOG.warn(message);
413 throw new ClockOutOfSyncException(message);
414 } else if (skew > warningSkew){
415 String message = "Reported time for server " + serverName + " is out of sync with master " +
416 "by " + skew + "ms. (Warning threshold is " + warningSkew + "ms; " +
417 "error threshold is " + maxSkew + "ms)";
418 LOG.warn(message);
419 }
420 }
421
422
423
424
425
426
427
428
429
430 private void checkIsDead(final ServerName serverName, final String what)
431 throws YouAreDeadException {
432 if (this.deadservers.isDeadServer(serverName)) {
433
434
435 String message = "Server " + what + " rejected; currently processing " +
436 serverName + " as dead server";
437 LOG.debug(message);
438 throw new YouAreDeadException(message);
439 }
440
441
442 if ((this.services == null || ((HMaster) this.services).isInitialized())
443 && this.deadservers.cleanPreviousInstance(serverName)) {
444
445
446 LOG.debug(what + ":" + " Server " + serverName + " came back up," +
447 " removed it from the dead servers list");
448 }
449 }
450
451
452
453
454
455 private ServerName findServerWithSameHostnamePortWithLock(
456 final ServerName serverName) {
457 ServerName end = ServerName.valueOf(serverName.getHostname(), serverName.getPort(),
458 Long.MAX_VALUE);
459
460 ServerName r = onlineServers.lowerKey(end);
461 if (r != null) {
462 if (ServerName.isSameHostnameAndPort(r, serverName)) {
463 return r;
464 }
465 }
466 return null;
467 }
468
469
470
471
472
473
474
475 @VisibleForTesting
476 void recordNewServerWithLock(final ServerName serverName, final ServerLoad sl) {
477 LOG.info("Registering server=" + serverName);
478 this.onlineServers.put(serverName, sl);
479 this.rsAdmins.remove(serverName);
480 }
481
482 public RegionStoreSequenceIds getLastFlushedSequenceId(byte[] encodedRegionName) {
483 RegionStoreSequenceIds.Builder builder = RegionStoreSequenceIds.newBuilder();
484 Long seqId = flushedSequenceIdByRegion.get(encodedRegionName);
485 builder.setLastFlushedSequenceId(seqId != null ? seqId.longValue() : HConstants.NO_SEQNUM);
486 Map<byte[], Long> storeFlushedSequenceId =
487 storeFlushedSequenceIdsByRegion.get(encodedRegionName);
488 if (storeFlushedSequenceId != null) {
489 for (Map.Entry<byte[], Long> entry : storeFlushedSequenceId.entrySet()) {
490 builder.addStoreSequenceId(StoreSequenceId.newBuilder()
491 .setFamilyName(ByteString.copyFrom(entry.getKey()))
492 .setSequenceId(entry.getValue().longValue()).build());
493 }
494 }
495 return builder.build();
496 }
497
498
499
500
501
502 public ServerLoad getLoad(final ServerName serverName) {
503 return this.onlineServers.get(serverName);
504 }
505
506
507
508
509
510
511
512 public double getAverageLoad() {
513 int totalLoad = 0;
514 int numServers = 0;
515 for (ServerLoad sl: this.onlineServers.values()) {
516 numServers++;
517 totalLoad += sl.getNumberOfRegions();
518 }
519 return numServers == 0 ? 0 :
520 (double)totalLoad / (double)numServers;
521 }
522
523
524 public int countOfRegionServers() {
525
526 return this.onlineServers.size();
527 }
528
529
530
531
532 public Map<ServerName, ServerLoad> getOnlineServers() {
533
534 synchronized (this.onlineServers) {
535 return Collections.unmodifiableMap(this.onlineServers);
536 }
537 }
538
539 public DeadServer getDeadServers() {
540 return this.deadservers;
541 }
542
543
544
545
546
547 public boolean areDeadServersInProgress() {
548 return this.deadservers.areDeadServersInProgress();
549 }
550
551 void letRegionServersShutdown() {
552 long previousLogTime = 0;
553 ServerName sn = master.getServerName();
554 ZooKeeperWatcher zkw = master.getZooKeeper();
555 int onlineServersCt;
556 while ((onlineServersCt = onlineServers.size()) > 0){
557
558 if (System.currentTimeMillis() > (previousLogTime + 1000)) {
559 Set<ServerName> remainingServers = onlineServers.keySet();
560 synchronized (onlineServers) {
561 if (remainingServers.size() == 1 && remainingServers.contains(sn)) {
562
563 return;
564 }
565 }
566 StringBuilder sb = new StringBuilder();
567
568 for (ServerName key : remainingServers) {
569 if (sb.length() > 0) {
570 sb.append(", ");
571 }
572 sb.append(key);
573 }
574 LOG.info("Waiting on regionserver(s) to go down " + sb.toString());
575 previousLogTime = System.currentTimeMillis();
576 }
577
578 try {
579 List<String> servers = ZKUtil.listChildrenNoWatch(zkw, zkw.rsZNode);
580 if (servers == null || servers.size() == 0 || (servers.size() == 1
581 && servers.contains(sn.toString()))) {
582 LOG.info("ZK shows there is only the master self online, exiting now");
583
584 break;
585 }
586 } catch (KeeperException ke) {
587 LOG.warn("Failed to list regionservers", ke);
588
589 break;
590 }
591 synchronized (onlineServers) {
592 try {
593 if (onlineServersCt == onlineServers.size()) onlineServers.wait(100);
594 } catch (InterruptedException ignored) {
595
596 }
597 }
598 }
599 }
600
601 private List<String> getRegionServersInZK(final ZooKeeperWatcher zkw)
602 throws KeeperException {
603 return ZKUtil.listChildrenNoWatch(zkw, zkw.rsZNode);
604 }
605
606
607
608
609
610 public synchronized void expireServer(final ServerName serverName) {
611 if (serverName.equals(master.getServerName())) {
612 if (!(master.isAborted() || master.isStopped())) {
613 master.stop("We lost our znode?");
614 }
615 return;
616 }
617 if (!services.isServerCrashProcessingEnabled()) {
618 LOG.info("Master doesn't enable ServerShutdownHandler during initialization, "
619 + "delay expiring server " + serverName);
620 this.queuedDeadServers.add(serverName);
621 return;
622 }
623 if (this.deadservers.isDeadServer(serverName)) {
624
625 LOG.warn("Expiration of " + serverName +
626 " but server shutdown already in progress");
627 return;
628 }
629 moveFromOnlineToDeadServers(serverName);
630
631
632
633 if (this.clusterShutdown) {
634 LOG.info("Cluster shutdown set; " + serverName +
635 " expired; onlineServers=" + this.onlineServers.size());
636 if (this.onlineServers.isEmpty()) {
637 master.stop("Cluster shutdown set; onlineServer=0");
638 }
639 return;
640 }
641
642 boolean carryingMeta = services.getAssignmentManager().isCarryingMeta(serverName) ==
643 AssignmentManager.ServerHostRegion.HOSTING_REGION;
644 ProcedureExecutor<MasterProcedureEnv> procExec = this.services.getMasterProcedureExecutor();
645 procExec.submitProcedure(new ServerCrashProcedure(
646 procExec.getEnvironment(), serverName, true, carryingMeta));
647 LOG.debug("Added=" + serverName +
648 " to dead servers, submitted shutdown handler to be executed meta=" + carryingMeta);
649
650
651 if (!this.listeners.isEmpty()) {
652 for (ServerListener listener : this.listeners) {
653 listener.serverRemoved(serverName);
654 }
655 }
656 }
657
658 @VisibleForTesting
659 public void moveFromOnlineToDeadServers(final ServerName sn) {
660 synchronized (onlineServers) {
661 if (!this.onlineServers.containsKey(sn)) {
662 LOG.warn("Expiration of " + sn + " but server not online");
663 }
664
665
666
667 this.deadservers.add(sn);
668 this.onlineServers.remove(sn);
669 onlineServers.notifyAll();
670 }
671 this.rsAdmins.remove(sn);
672 }
673
674 public synchronized void processDeadServer(final ServerName serverName, boolean shouldSplitWal) {
675
676
677
678
679
680
681
682
683 if (!services.getAssignmentManager().isFailoverCleanupDone()) {
684 requeuedDeadServers.put(serverName, shouldSplitWal);
685 return;
686 }
687
688 this.deadservers.add(serverName);
689 ProcedureExecutor<MasterProcedureEnv> procExec = this.services.getMasterProcedureExecutor();
690 procExec.submitProcedure(new ServerCrashProcedure(
691 procExec.getEnvironment(), serverName, shouldSplitWal, false));
692 }
693
694
695
696
697
698 synchronized void processQueuedDeadServers() {
699 if (!services.isServerCrashProcessingEnabled()) {
700 LOG.info("Master hasn't enabled ServerShutdownHandler");
701 }
702 Iterator<ServerName> serverIterator = queuedDeadServers.iterator();
703 while (serverIterator.hasNext()) {
704 ServerName tmpServerName = serverIterator.next();
705 expireServer(tmpServerName);
706 serverIterator.remove();
707 requeuedDeadServers.remove(tmpServerName);
708 }
709
710 if (!services.getAssignmentManager().isFailoverCleanupDone()) {
711 LOG.info("AssignmentManager hasn't finished failover cleanup; waiting");
712 }
713 for (Map.Entry<ServerName, Boolean> entry : requeuedDeadServers.entrySet()) {
714 processDeadServer(entry.getKey(), entry.getValue());
715 }
716 requeuedDeadServers.clear();
717 }
718
719
720
721
722 public boolean removeServerFromDrainList(final ServerName sn) {
723
724
725
726 if (!this.isServerOnline(sn)) {
727 LOG.warn("Server " + sn + " is not currently online. " +
728 "Removing from draining list anyway, as requested.");
729 }
730
731 return this.drainingServers.remove(sn);
732 }
733
734
735
736
737 public boolean addServerToDrainList(final ServerName sn) {
738
739
740
741 if (!this.isServerOnline(sn)) {
742 LOG.warn("Server " + sn + " is not currently online. " +
743 "Ignoring request to add it to draining list.");
744 return false;
745 }
746
747
748 if (this.drainingServers.contains(sn)) {
749 LOG.warn("Server " + sn + " is already in the draining server list." +
750 "Ignoring request to add it again.");
751 return false;
752 }
753 LOG.info("Server " + sn + " added to draining server list.");
754 return this.drainingServers.add(sn);
755 }
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770 public RegionOpeningState sendRegionOpen(final ServerName server,
771 HRegionInfo region, int versionOfOfflineNode, List<ServerName> favoredNodes)
772 throws IOException {
773 AdminService.BlockingInterface admin = getRsAdmin(server);
774 if (admin == null) {
775 LOG.warn("Attempting to send OPEN RPC to server " + server.toString() +
776 " failed because no RPC connection found to this server");
777 return RegionOpeningState.FAILED_OPENING;
778 }
779 OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(server,
780 region, versionOfOfflineNode, favoredNodes,
781 (RecoveryMode.LOG_REPLAY == this.services.getMasterFileSystem().getLogRecoveryMode()));
782 try {
783 OpenRegionResponse response = admin.openRegion(null, request);
784 return ResponseConverter.getRegionOpeningState(response);
785 } catch (ServiceException se) {
786 checkForRSznode(server, se);
787 throw ProtobufUtil.getRemoteException(se);
788 }
789 }
790
791
792
793
794
795
796
797
798
799
800
801
802 private void checkForRSznode(final ServerName serverName, final ServiceException se) {
803 if (se.getCause() == null) return;
804 Throwable t = se.getCause();
805 if (t instanceof ConnectException) {
806
807 } else {
808
809 if (!(t instanceof IOException)) return;
810 if (t.getCause() == null) return;
811 if (!(t.getCause() instanceof FailedServerException)) return;
812
813 }
814 if (!isServerOnline(serverName)) return;
815
816
817
818
819 List<String> servers = null;
820 try {
821 servers = getRegionServersInZK(this.master.getZooKeeper());
822 } catch (KeeperException ke) {
823 LOG.warn("Failed to list regionservers", ke);
824
825 }
826 boolean found = false;
827 if (servers != null) {
828 for (String serverNameAsStr: servers) {
829 ServerName sn = ServerName.valueOf(serverNameAsStr);
830 if (sn.equals(serverName)) {
831
832 found = true;
833 break;
834 }
835 }
836 }
837 if (!found) {
838 LOG.warn("Online server " + serverName.toString() + " has no corresponding " +
839 "ephemeral znode (Did it die before registering in zk?); " +
840 "calling expire to clean it up!");
841 expireServer(serverName);
842 }
843 }
844
845
846
847
848
849
850
851
852
853
854 public List<RegionOpeningState> sendRegionOpen(ServerName server,
855 List<Triple<HRegionInfo, Integer, List<ServerName>>> regionOpenInfos)
856 throws IOException {
857 AdminService.BlockingInterface admin = getRsAdmin(server);
858 if (admin == null) {
859 LOG.warn("Attempting to send OPEN RPC to server " + server.toString() +
860 " failed because no RPC connection found to this server");
861 return null;
862 }
863
864 OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(server, regionOpenInfos,
865 (RecoveryMode.LOG_REPLAY == this.services.getMasterFileSystem().getLogRecoveryMode()));
866 try {
867 OpenRegionResponse response = admin.openRegion(null, request);
868 return ResponseConverter.getRegionOpeningStateList(response);
869 } catch (ServiceException se) {
870 checkForRSznode(server, se);
871 throw ProtobufUtil.getRemoteException(se);
872 }
873 }
874
875 private HBaseRpcController newRpcController() {
876 return rpcControllerFactory == null ? null : rpcControllerFactory.newController();
877 }
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893 public boolean sendRegionClose(ServerName server, HRegionInfo region,
894 int versionOfClosingNode, ServerName dest, boolean transitionInZK) throws IOException {
895 if (server == null) throw new NullPointerException("Passed server is null");
896 AdminService.BlockingInterface admin = getRsAdmin(server);
897 if (admin == null) {
898 throw new IOException("Attempting to send CLOSE RPC to server " +
899 server.toString() + " for region " +
900 region.getRegionNameAsString() +
901 " failed because no RPC connection found to this server");
902 }
903 HBaseRpcController controller = newRpcController();
904 return ProtobufUtil.closeRegion(controller, admin, server, region.getRegionName(),
905 versionOfClosingNode, dest, transitionInZK);
906 }
907
908 public boolean sendRegionClose(ServerName server,
909 HRegionInfo region, int versionOfClosingNode) throws IOException {
910 return sendRegionClose(server, region, versionOfClosingNode, null, true);
911 }
912
913
914
915
916
917
918
919
920
921 public void sendRegionWarmup(ServerName server,
922 HRegionInfo region) {
923 if (server == null) return;
924 try {
925 AdminService.BlockingInterface admin = getRsAdmin(server);
926 HBaseRpcController controller = newRpcController();
927 ProtobufUtil.warmupRegion(controller, admin, region);
928 } catch (IOException e) {
929 LOG.error("Received exception in RPC for warmup server:" +
930 server + "region: " + region +
931 "exception: " + e);
932 }
933 }
934
935
936
937
938
939 public static void closeRegionSilentlyAndWait(ClusterConnection connection,
940 ServerName server, HRegionInfo region, long timeout) throws IOException, InterruptedException {
941 AdminService.BlockingInterface rs = connection.getAdmin(server);
942 HBaseRpcController controller = connection.getRpcControllerFactory().newController();
943 try {
944 ProtobufUtil.closeRegion(controller, rs, server, region.getRegionName(), false);
945 } catch (IOException e) {
946 LOG.warn("Exception when closing region: " + region.getRegionNameAsString(), e);
947 }
948 long expiration = timeout + System.currentTimeMillis();
949 while (System.currentTimeMillis() < expiration) {
950 controller.reset();
951 try {
952 HRegionInfo rsRegion =
953 ProtobufUtil.getRegionInfo(controller, rs, region.getRegionName());
954 if (rsRegion == null) return;
955 } catch (IOException ioe) {
956 if (ioe instanceof NotServingRegionException)
957 return;
958 LOG.warn("Exception when retrieving regioninfo from: "
959 + region.getRegionNameAsString(), ioe);
960 }
961 Thread.sleep(1000);
962 }
963 throw new IOException("Region " + region + " failed to close within"
964 + " timeout " + timeout);
965 }
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980 public void sendRegionsMerge(ServerName server, HRegionInfo region_a,
981 HRegionInfo region_b, boolean forcible, User user) throws IOException {
982 if (server == null)
983 throw new NullPointerException("Passed server is null");
984 if (region_a == null || region_b == null)
985 throw new NullPointerException("Passed region is null");
986 AdminService.BlockingInterface admin = getRsAdmin(server);
987 if (admin == null) {
988 throw new IOException("Attempting to send MERGE REGIONS RPC to server "
989 + server.toString() + " for region "
990 + region_a.getRegionNameAsString() + ","
991 + region_b.getRegionNameAsString()
992 + " failed because no RPC connection found to this server");
993 }
994 HBaseRpcController controller = newRpcController();
995 ProtobufUtil.mergeRegions(controller, admin, region_a, region_b, forcible, user);
996 }
997
998
999
1000
1001 public boolean isServerReachable(ServerName server) {
1002 if (server == null) throw new NullPointerException("Passed server is null");
1003
1004
1005 RetryCounter retryCounter = pingRetryCounterFactory.create();
1006 while (retryCounter.shouldRetry()) {
1007 synchronized (this.onlineServers) {
1008 if (this.deadservers.isDeadServer(server)) {
1009 return false;
1010 }
1011 }
1012 try {
1013 HBaseRpcController controller = newRpcController();
1014 AdminService.BlockingInterface admin = getRsAdmin(server);
1015 if (admin != null) {
1016 ServerInfo info = ProtobufUtil.getServerInfo(controller, admin);
1017 return info != null && info.hasServerName()
1018 && server.getStartcode() == info.getServerName().getStartCode();
1019 }
1020 } catch (IOException ioe) {
1021 if (LOG.isDebugEnabled()) {
1022 LOG.debug("Couldn't reach " + server + ", try=" + retryCounter.getAttemptTimes() + " of "
1023 + retryCounter.getMaxAttempts(), ioe);
1024 }
1025 try {
1026 retryCounter.sleepUntilNextRetry();
1027 } catch(InterruptedException ie) {
1028 Thread.currentThread().interrupt();
1029 break;
1030 }
1031 }
1032 }
1033 return false;
1034 }
1035
1036
1037
1038
1039
1040
1041
1042 private AdminService.BlockingInterface getRsAdmin(final ServerName sn)
1043 throws IOException {
1044 AdminService.BlockingInterface admin = this.rsAdmins.get(sn);
1045 if (admin == null) {
1046 LOG.debug("New admin connection to " + sn.toString());
1047 if (sn.equals(master.getServerName()) && master instanceof HRegionServer) {
1048
1049 admin = ((HRegionServer)master).getRSRpcServices();
1050 } else {
1051 admin = this.connection.getAdmin(sn);
1052 }
1053 this.rsAdmins.put(sn, admin);
1054 }
1055 return admin;
1056 }
1057
1058
1059
1060
1061
1062
1063 private int getMinToStart() {
1064
1065 int requiredMinToStart = 1;
1066 if (BaseLoadBalancer.tablesOnMaster(master.getConfiguration())) {
1067 if (!BaseLoadBalancer.userTablesOnMaster(master.getConfiguration())) {
1068
1069
1070
1071 requiredMinToStart = 2;
1072 }
1073 }
1074 int minToStart = this.master.getConfiguration().getInt(WAIT_ON_REGIONSERVERS_MINTOSTART, -1);
1075
1076 return minToStart == -1 || minToStart < requiredMinToStart? requiredMinToStart: minToStart;
1077 }
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092 public void waitForRegionServers(MonitoredTask status)
1093 throws InterruptedException {
1094 final long interval = this.master.getConfiguration().
1095 getLong(WAIT_ON_REGIONSERVERS_INTERVAL, 1500);
1096 final long timeout = this.master.getConfiguration().
1097 getLong(WAIT_ON_REGIONSERVERS_TIMEOUT, 4500);
1098
1099 int minToStart = getMinToStart();
1100 int maxToStart = this.master.getConfiguration().
1101 getInt(WAIT_ON_REGIONSERVERS_MAXTOSTART, Integer.MAX_VALUE);
1102 if (maxToStart < minToStart) {
1103 LOG.warn(String.format("The value of '%s' (%d) is set less than '%s' (%d), ignoring.",
1104 WAIT_ON_REGIONSERVERS_MAXTOSTART, maxToStart,
1105 WAIT_ON_REGIONSERVERS_MINTOSTART, minToStart));
1106 maxToStart = Integer.MAX_VALUE;
1107 }
1108
1109 long now = System.currentTimeMillis();
1110 final long startTime = now;
1111 long slept = 0;
1112 long lastLogTime = 0;
1113 long lastCountChange = startTime;
1114 int count = countOfRegionServers();
1115 int oldCount = 0;
1116
1117
1118
1119
1120
1121
1122
1123 for (ServerListener listener: this.listeners) {
1124 listener.waiting();
1125 }
1126 while (!this.master.isStopped() && count < maxToStart &&
1127 ((lastCountChange + interval) > now || timeout > slept || count < minToStart)) {
1128
1129 if (oldCount != count || lastLogTime + interval < now) {
1130 lastLogTime = now;
1131 String msg =
1132 "Waiting on RegionServer count=" + count + " to settle; waited="+
1133 slept + "ms, expecting min=" + minToStart + " server(s), max="+ getStrForMax(maxToStart) +
1134 " server(s), " + "timeout=" + timeout + "ms, lastChange=" + (lastCountChange - now) + "ms";
1135 LOG.info(msg);
1136 status.setStatus(msg);
1137 }
1138
1139
1140 final long sleepTime = 50;
1141 Thread.sleep(sleepTime);
1142 now = System.currentTimeMillis();
1143 slept = now - startTime;
1144
1145 oldCount = count;
1146 count = countOfRegionServers();
1147 if (count != oldCount) {
1148 lastCountChange = now;
1149 }
1150 }
1151
1152 LOG.info("Finished wait on RegionServer count=" + count + "; waited=" + slept + "ms," +
1153 " expected min=" + minToStart + " server(s), max=" + getStrForMax(maxToStart) + " server(s),"+
1154 " master is "+ (this.master.isStopped() ? "stopped.": "running")
1155 );
1156 }
1157
1158 private String getStrForMax(final int max) {
1159 return max == Integer.MAX_VALUE? "NO_LIMIT": Integer.toString(max);
1160 }
1161
1162
1163
1164
1165 public List<ServerName> getOnlineServersList() {
1166
1167
1168 return new ArrayList<ServerName>(this.onlineServers.keySet());
1169 }
1170
1171
1172
1173
1174 public List<ServerName> getDrainingServersList() {
1175 return new ArrayList<ServerName>(this.drainingServers);
1176 }
1177
1178
1179
1180
1181 Set<ServerName> getDeadNotExpiredServers() {
1182 return new HashSet<ServerName>(this.queuedDeadServers);
1183 }
1184
1185
1186
1187
1188
1189
1190 void removeRequeuedDeadServers() {
1191 requeuedDeadServers.clear();
1192 }
1193
1194
1195
1196
1197
1198 Map<ServerName, Boolean> getRequeuedDeadServers() {
1199 return Collections.unmodifiableMap(this.requeuedDeadServers);
1200 }
1201
1202 public boolean isServerOnline(ServerName serverName) {
1203 return serverName != null && onlineServers.containsKey(serverName);
1204 }
1205
1206
1207
1208
1209
1210 public boolean isServerWithSameHostnamePortOnline(final ServerName serverName) {
1211 return findServerWithSameHostnamePortWithLock(serverName) != null;
1212 }
1213
1214
1215
1216
1217
1218
1219
1220 public synchronized boolean isServerDead(ServerName serverName) {
1221 return serverName == null || deadservers.isDeadServer(serverName)
1222 || queuedDeadServers.contains(serverName)
1223 || requeuedDeadServers.containsKey(serverName);
1224 }
1225
1226 public void shutdownCluster() {
1227 this.clusterShutdown = true;
1228 this.master.stop("Cluster shutdown requested");
1229 }
1230
1231 public boolean isClusterShutdown() {
1232 return this.clusterShutdown;
1233 }
1234
1235
1236
1237
1238 public void stop() {
1239 if (connection != null) {
1240 try {
1241 connection.close();
1242 } catch (IOException e) {
1243 LOG.error("Attempt to close connection to master failed", e);
1244 }
1245 }
1246 }
1247
1248
1249
1250
1251
1252
1253 public List<ServerName> createDestinationServersList(final List<ServerName> serversToExclude){
1254 final List<ServerName> destServers = getOnlineServersList();
1255
1256 if (serversToExclude != null){
1257 destServers.removeAll(serversToExclude);
1258 }
1259
1260
1261 final List<ServerName> drainingServersCopy = getDrainingServersList();
1262 if (!drainingServersCopy.isEmpty()) {
1263 for (final ServerName server: drainingServersCopy) {
1264 destServers.remove(server);
1265 }
1266 }
1267
1268
1269 removeDeadNotExpiredServers(destServers);
1270 return destServers;
1271 }
1272
1273
1274
1275
1276 public List<ServerName> createDestinationServersList(){
1277 return createDestinationServersList(null);
1278 }
1279
1280
1281
1282
1283
1284
1285
1286 void removeDeadNotExpiredServers(List<ServerName> servers) {
1287 Set<ServerName> deadNotExpiredServersCopy = this.getDeadNotExpiredServers();
1288 if (!deadNotExpiredServersCopy.isEmpty()) {
1289 for (ServerName server : deadNotExpiredServersCopy) {
1290 LOG.debug("Removing dead but not expired server: " + server
1291 + " from eligible server pool.");
1292 servers.remove(server);
1293 }
1294 }
1295 }
1296
1297
1298
1299
1300 void clearDeadServersWithSameHostNameAndPortOfOnlineServer() {
1301 for (ServerName serverName : getOnlineServersList()) {
1302 deadservers.cleanAllPreviousInstances(serverName);
1303 }
1304 }
1305
1306
1307
1308
1309 public void removeRegion(final HRegionInfo regionInfo) {
1310 final byte[] encodedName = regionInfo.getEncodedNameAsBytes();
1311 storeFlushedSequenceIdsByRegion.remove(encodedName);
1312 flushedSequenceIdByRegion.remove(encodedName);
1313 }
1314
1315 @VisibleForTesting
1316 public boolean isRegionInServerManagerStates(final HRegionInfo hri) {
1317 final byte[] encodedName = hri.getEncodedNameAsBytes();
1318 return (storeFlushedSequenceIdsByRegion.containsKey(encodedName)
1319 || flushedSequenceIdByRegion.containsKey(encodedName));
1320 }
1321
1322
1323
1324
1325 public void removeRegions(final List<HRegionInfo> regions) {
1326 for (HRegionInfo hri: regions) {
1327 removeRegion(hri);
1328 }
1329 }
1330 }