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