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.io.InterruptedIOException;
23 import java.lang.reflect.Constructor;
24 import java.lang.reflect.InvocationTargetException;
25 import java.net.InetAddress;
26 import java.net.InetSocketAddress;
27 import java.net.UnknownHostException;
28 import java.util.ArrayList;
29 import java.util.Collection;
30 import java.util.Collections;
31 import java.util.Comparator;
32 import java.util.HashSet;
33 import java.util.Iterator;
34 import java.util.List;
35 import java.util.Map;
36 import java.util.Set;
37 import java.util.concurrent.TimeUnit;
38 import java.util.concurrent.atomic.AtomicReference;
39 import java.util.regex.Pattern;
40
41 import javax.servlet.ServletException;
42 import javax.servlet.http.HttpServlet;
43 import javax.servlet.http.HttpServletRequest;
44 import javax.servlet.http.HttpServletResponse;
45
46 import org.apache.commons.logging.Log;
47 import org.apache.commons.logging.LogFactory;
48 import org.apache.hadoop.conf.Configuration;
49 import org.apache.hadoop.fs.Path;
50 import org.apache.hadoop.hbase.ClusterStatus;
51 import org.apache.hadoop.hbase.CoordinatedStateException;
52 import org.apache.hadoop.hbase.CoordinatedStateManager;
53 import org.apache.hadoop.hbase.DoNotRetryIOException;
54 import org.apache.hadoop.hbase.HBaseIOException;
55 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
56 import org.apache.hadoop.hbase.HColumnDescriptor;
57 import org.apache.hadoop.hbase.HConstants;
58 import org.apache.hadoop.hbase.HRegionInfo;
59 import org.apache.hadoop.hbase.HTableDescriptor;
60 import org.apache.hadoop.hbase.MasterNotRunningException;
61 import org.apache.hadoop.hbase.MetaMigrationConvertingToPB;
62 import org.apache.hadoop.hbase.MetaTableAccessor;
63 import org.apache.hadoop.hbase.NamespaceDescriptor;
64 import org.apache.hadoop.hbase.NamespaceNotFoundException;
65 import org.apache.hadoop.hbase.PleaseHoldException;
66 import org.apache.hadoop.hbase.ProcedureInfo;
67 import org.apache.hadoop.hbase.Server;
68 import org.apache.hadoop.hbase.ServerLoad;
69 import org.apache.hadoop.hbase.ServerName;
70 import org.apache.hadoop.hbase.TableDescriptors;
71 import org.apache.hadoop.hbase.TableName;
72 import org.apache.hadoop.hbase.TableNotDisabledException;
73 import org.apache.hadoop.hbase.TableNotFoundException;
74 import org.apache.hadoop.hbase.UnknownRegionException;
75 import org.apache.hadoop.hbase.classification.InterfaceAudience;
76 import org.apache.hadoop.hbase.client.MetaScanner;
77 import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitor;
78 import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitorBase;
79 import org.apache.hadoop.hbase.client.RegionReplicaUtil;
80 import org.apache.hadoop.hbase.client.Result;
81 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
82 import org.apache.hadoop.hbase.exceptions.DeserializationException;
83 import org.apache.hadoop.hbase.executor.ExecutorType;
84 import org.apache.hadoop.hbase.ipc.RpcServer;
85 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
86 import org.apache.hadoop.hbase.master.MasterRpcServices.BalanceSwitchMode;
87 import org.apache.hadoop.hbase.master.RegionState.State;
88 import org.apache.hadoop.hbase.master.balancer.BalancerChore;
89 import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
90 import org.apache.hadoop.hbase.master.balancer.ClusterStatusChore;
91 import org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory;
92 import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
93 import org.apache.hadoop.hbase.master.cleaner.LogCleaner;
94 import org.apache.hadoop.hbase.master.handler.CreateTableHandler;
95 import org.apache.hadoop.hbase.master.handler.DeleteTableHandler;
96 import org.apache.hadoop.hbase.master.handler.DisableTableHandler;
97 import org.apache.hadoop.hbase.master.handler.DispatchMergingRegionHandler;
98 import org.apache.hadoop.hbase.master.handler.EnableTableHandler;
99 import org.apache.hadoop.hbase.master.handler.ModifyTableHandler;
100 import org.apache.hadoop.hbase.master.handler.TableAddFamilyHandler;
101 import org.apache.hadoop.hbase.master.handler.TableDeleteFamilyHandler;
102 import org.apache.hadoop.hbase.master.handler.TableModifyFamilyHandler;
103 import org.apache.hadoop.hbase.master.handler.TruncateTableHandler;
104 import org.apache.hadoop.hbase.master.procedure.AddColumnFamilyProcedure;
105 import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure;
106 import org.apache.hadoop.hbase.master.procedure.DeleteColumnFamilyProcedure;
107 import org.apache.hadoop.hbase.master.procedure.DeleteTableProcedure;
108 import org.apache.hadoop.hbase.master.procedure.DisableTableProcedure;
109 import org.apache.hadoop.hbase.master.procedure.EnableTableProcedure;
110 import org.apache.hadoop.hbase.master.procedure.MasterProcedureConstants;
111 import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
112 import org.apache.hadoop.hbase.master.procedure.ModifyColumnFamilyProcedure;
113 import org.apache.hadoop.hbase.master.procedure.ModifyTableProcedure;
114 import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch;
115 import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait;
116 import org.apache.hadoop.hbase.master.procedure.TruncateTableProcedure;
117 import org.apache.hadoop.hbase.master.snapshot.SnapshotManager;
118 import org.apache.hadoop.hbase.monitoring.MemoryBoundedLogMessageBuffer;
119 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
120 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
121 import org.apache.hadoop.hbase.procedure.MasterProcedureManagerHost;
122 import org.apache.hadoop.hbase.procedure.flush.MasterFlushTableProcedureManager;
123 import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
124 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
125 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionServerInfo;
126 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
127 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
128 import org.apache.hadoop.hbase.quotas.MasterQuotaManager;
129 import org.apache.hadoop.hbase.quotas.RegionStateListener;
130 import org.apache.hadoop.hbase.regionserver.HRegionServer;
131 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
132 import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost;
133 import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy;
134 import org.apache.hadoop.hbase.replication.regionserver.Replication;
135 import org.apache.hadoop.hbase.security.UserProvider;
136 import org.apache.hadoop.hbase.util.Addressing;
137 import org.apache.hadoop.hbase.util.Bytes;
138 import org.apache.hadoop.hbase.util.CompressionTest;
139 import org.apache.hadoop.hbase.util.ConfigUtil;
140 import org.apache.hadoop.hbase.util.EncryptionTest;
141 import org.apache.hadoop.hbase.util.FSUtils;
142 import org.apache.hadoop.hbase.util.HFileArchiveUtil;
143 import org.apache.hadoop.hbase.util.HasThread;
144 import org.apache.hadoop.hbase.util.ModifyRegionUtils;
145 import org.apache.hadoop.hbase.util.Pair;
146 import org.apache.hadoop.hbase.util.Threads;
147 import org.apache.hadoop.hbase.util.VersionInfo;
148 import org.apache.hadoop.hbase.zookeeper.DrainingServerTracker;
149 import org.apache.hadoop.hbase.zookeeper.LoadBalancerTracker;
150 import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
151 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
152 import org.apache.hadoop.hbase.zookeeper.RegionServerTracker;
153 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
154 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
155 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
156 import org.apache.zookeeper.KeeperException;
157 import org.mortbay.jetty.Connector;
158 import org.mortbay.jetty.nio.SelectChannelConnector;
159 import org.mortbay.jetty.servlet.Context;
160
161 import com.google.common.annotations.VisibleForTesting;
162 import com.google.common.collect.Maps;
163 import com.google.protobuf.Descriptors;
164 import com.google.protobuf.Service;
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
182 @SuppressWarnings("deprecation")
183 public class HMaster extends HRegionServer implements MasterServices, Server {
184 private static final Log LOG = LogFactory.getLog(HMaster.class.getName());
185
186
187
188
189
190 private static class InitializationMonitor extends HasThread {
191
192 public static final String TIMEOUT_KEY = "hbase.master.initializationmonitor.timeout";
193 public static final long TIMEOUT_DEFAULT = TimeUnit.MILLISECONDS.convert(15, TimeUnit.MINUTES);
194
195
196
197
198
199 public static final String HALT_KEY = "hbase.master.initializationmonitor.haltontimeout";
200 public static final boolean HALT_DEFAULT = false;
201
202 private final HMaster master;
203 private final long timeout;
204 private final boolean haltOnTimeout;
205
206
207 InitializationMonitor(HMaster master) {
208 super("MasterInitializationMonitor");
209 this.master = master;
210 this.timeout = master.getConfiguration().getLong(TIMEOUT_KEY, TIMEOUT_DEFAULT);
211 this.haltOnTimeout = master.getConfiguration().getBoolean(HALT_KEY, HALT_DEFAULT);
212 this.setDaemon(true);
213 }
214
215 @Override
216 public void run() {
217 try {
218 while (!master.isStopped() && master.isActiveMaster()) {
219 Thread.sleep(timeout);
220 if (master.isInitialized()) {
221 LOG.debug("Initialization completed within allotted tolerance. Monitor exiting.");
222 } else {
223 LOG.error("Master failed to complete initialization after " + timeout + "ms. Please"
224 + " consider submitting a bug report including a thread dump of this process.");
225 if (haltOnTimeout) {
226 LOG.error("Zombie Master exiting. Thread dump to stdout");
227 Threads.printThreadInfo(System.out, "Zombie HMaster");
228 System.exit(-1);
229 }
230 }
231 }
232 } catch (InterruptedException ie) {
233 LOG.trace("InitMonitor thread interrupted. Existing.");
234 }
235 }
236 }
237
238
239
240 public static final String MASTER = "master";
241
242
243 private final ActiveMasterManager activeMasterManager;
244
245 RegionServerTracker regionServerTracker;
246
247 private DrainingServerTracker drainingServerTracker;
248
249 LoadBalancerTracker loadBalancerTracker;
250
251
252 private TableNamespaceManager tableNamespaceManager;
253
254
255 final MetricsMaster metricsMaster;
256
257 private MasterFileSystem fileSystemManager;
258
259
260 volatile ServerManager serverManager;
261
262
263 AssignmentManager assignmentManager;
264
265
266
267
268 MemoryBoundedLogMessageBuffer rsFatals;
269
270
271 private volatile boolean isActiveMaster = false;
272
273
274
275 volatile boolean initialized = false;
276
277
278
279 volatile boolean serviceStarted = false;
280
281
282 private volatile boolean serverShutdownHandlerEnabled = false;
283
284 LoadBalancer balancer;
285 private BalancerChore balancerChore;
286 private ClusterStatusChore clusterStatusChore;
287 private ClusterStatusPublisher clusterStatusPublisherChore = null;
288
289 CatalogJanitor catalogJanitorChore;
290 private LogCleaner logCleaner;
291 private HFileCleaner hfileCleaner;
292
293 MasterCoprocessorHost cpHost;
294
295 private final boolean preLoadTableDescriptors;
296
297
298 private long masterActiveTime;
299
300
301 private final boolean masterCheckCompression;
302
303
304 private final boolean masterCheckEncryption;
305
306
307 private enum ProcedureConf {
308 PROCEDURE_ENABLED,
309 HANDLER_USED,
310 PROCEDURE_FULLY_DISABLED,
311 }
312 private final ProcedureConf procedureConf;
313
314 Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap();
315
316
317 SnapshotManager snapshotManager;
318
319 MasterProcedureManagerHost mpmHost;
320
321
322 private volatile MasterQuotaManager quotaManager;
323
324 private ProcedureExecutor<MasterProcedureEnv> procedureExecutor;
325 private WALProcedureStore procedureStore;
326
327
328 private volatile boolean initializationBeforeMetaAssignment = false;
329
330
331 private org.mortbay.jetty.Server masterJettyServer;
332
333 public static class RedirectServlet extends HttpServlet {
334 private static final long serialVersionUID = 2894774810058302472L;
335 private static int regionServerInfoPort;
336
337 @Override
338 public void doGet(HttpServletRequest request,
339 HttpServletResponse response) throws ServletException, IOException {
340 String redirectUrl = request.getScheme() + "://"
341 + request.getServerName() + ":" + regionServerInfoPort
342 + request.getRequestURI();
343 response.sendRedirect(redirectUrl);
344 }
345 }
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363 public HMaster(final Configuration conf, CoordinatedStateManager csm)
364 throws IOException, KeeperException, InterruptedException {
365 super(conf, csm);
366 this.rsFatals = new MemoryBoundedLogMessageBuffer(
367 conf.getLong("hbase.master.buffer.for.rs.fatals", 1*1024*1024));
368
369 LOG.info("hbase.rootdir=" + FSUtils.getRootDir(this.conf) +
370 ", hbase.cluster.distributed=" + this.conf.getBoolean(HConstants.CLUSTER_DISTRIBUTED, false));
371
372
373 this.conf.setBoolean(HConstants.USE_META_REPLICAS, false);
374
375 Replication.decorateMasterConfiguration(this.conf);
376
377
378
379 if (this.conf.get("mapreduce.task.attempt.id") == null) {
380 this.conf.set("mapreduce.task.attempt.id", "hb_m_" + this.serverName.toString());
381 }
382
383
384 this.masterCheckCompression = conf.getBoolean("hbase.master.check.compression", true);
385
386
387 this.masterCheckEncryption = conf.getBoolean("hbase.master.check.encryption", true);
388
389 this.metricsMaster = new MetricsMaster(new MetricsMasterWrapperImpl(this));
390
391
392
393
394 String procedureConfString = conf.get("hbase.master.procedure.tableddl", "enabled");
395 if (procedureConfString.equalsIgnoreCase("disabled")) {
396 LOG.info("Master will use handler for new table DDL"
397 + " and all unfinished table DDLs in procedure store will be disgarded.");
398 this.procedureConf = ProcedureConf.PROCEDURE_FULLY_DISABLED;
399 } else if (procedureConfString.equalsIgnoreCase("unused")) {
400 LOG.info("Master will use handler for new table DDL"
401 + " and all unfinished table DDLs in procedure store will continue to execute.");
402 this.procedureConf = ProcedureConf.HANDLER_USED;
403 } else {
404 this.procedureConf = ProcedureConf.PROCEDURE_ENABLED;
405 }
406
407 this.preLoadTableDescriptors = conf.getBoolean("hbase.master.preload.tabledescriptors", true);
408
409
410
411 boolean shouldPublish = conf.getBoolean(HConstants.STATUS_PUBLISHED,
412 HConstants.STATUS_PUBLISHED_DEFAULT);
413 Class<? extends ClusterStatusPublisher.Publisher> publisherClass =
414 conf.getClass(ClusterStatusPublisher.STATUS_PUBLISHER_CLASS,
415 ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS,
416 ClusterStatusPublisher.Publisher.class);
417
418 if (shouldPublish) {
419 if (publisherClass == null) {
420 LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " +
421 ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS +
422 " is not set - not publishing status");
423 } else {
424 clusterStatusPublisherChore = new ClusterStatusPublisher(this, conf, publisherClass);
425 getChoreService().scheduleChore(clusterStatusPublisherChore);
426 }
427 }
428
429
430 if (!conf.getBoolean("hbase.testing.nocluster", false)) {
431 activeMasterManager = new ActiveMasterManager(zooKeeper, this.serverName, this);
432 int infoPort = putUpJettyServer();
433 startActiveMasterManager(infoPort);
434 } else {
435 activeMasterManager = null;
436 }
437 }
438
439
440 private int putUpJettyServer() throws IOException {
441 if (!conf.getBoolean("hbase.master.infoserver.redirect", true)) {
442 return -1;
443 }
444 int infoPort = conf.getInt("hbase.master.info.port.orig",
445 HConstants.DEFAULT_MASTER_INFOPORT);
446
447 if (infoPort < 0 || infoServer == null) {
448 return -1;
449 }
450 String addr = conf.get("hbase.master.info.bindAddress", "0.0.0.0");
451 if (!Addressing.isLocalAddress(InetAddress.getByName(addr))) {
452 String msg =
453 "Failed to start redirecting jetty server. Address " + addr
454 + " does not belong to this host. Correct configuration parameter: "
455 + "hbase.master.info.bindAddress";
456 LOG.error(msg);
457 throw new IOException(msg);
458 }
459
460 RedirectServlet.regionServerInfoPort = infoServer.getPort();
461 if(RedirectServlet.regionServerInfoPort == infoPort) {
462 return infoPort;
463 }
464 masterJettyServer = new org.mortbay.jetty.Server();
465 Connector connector = new SelectChannelConnector();
466 connector.setHost(addr);
467 connector.setPort(infoPort);
468 masterJettyServer.addConnector(connector);
469 masterJettyServer.setStopAtShutdown(true);
470 Context context = new Context(masterJettyServer, "/", Context.NO_SESSIONS);
471 context.addServlet(RedirectServlet.class, "/*");
472 try {
473 masterJettyServer.start();
474 } catch (Exception e) {
475 throw new IOException("Failed to start redirecting jetty server", e);
476 }
477 return connector.getLocalPort();
478 }
479
480
481
482
483 @Override
484 protected void login(UserProvider user, String host) throws IOException {
485 try {
486 super.login(user, host);
487 } catch (IOException ie) {
488 user.login("hbase.master.keytab.file",
489 "hbase.master.kerberos.principal", host);
490 }
491 }
492
493
494
495
496
497
498 @Override
499 protected void waitForMasterActive(){
500 boolean tablesOnMaster = BaseLoadBalancer.tablesOnMaster(conf);
501 while (!(tablesOnMaster && isActiveMaster)
502 && !isStopped() && !isAborted()) {
503 sleeper.sleep();
504 }
505 }
506
507 @VisibleForTesting
508 public MasterRpcServices getMasterRpcServices() {
509 return (MasterRpcServices)rpcServices;
510 }
511
512 public boolean balanceSwitch(final boolean b) throws IOException {
513 return getMasterRpcServices().switchBalancer(b, BalanceSwitchMode.ASYNC);
514 }
515
516 @Override
517 protected String getProcessName() {
518 return MASTER;
519 }
520
521 @Override
522 protected boolean canCreateBaseZNode() {
523 return true;
524 }
525
526 @Override
527 protected boolean canUpdateTableDescriptor() {
528 return true;
529 }
530
531 @Override
532 protected RSRpcServices createRpcServices() throws IOException {
533 return new MasterRpcServices(this);
534 }
535
536 @Override
537 protected void configureInfoServer() {
538 infoServer.addServlet("master-status", "/master-status", MasterStatusServlet.class);
539 infoServer.setAttribute(MASTER, this);
540 if (BaseLoadBalancer.tablesOnMaster(conf)) {
541 super.configureInfoServer();
542 }
543 }
544
545 @Override
546 protected Class<? extends HttpServlet> getDumpServlet() {
547 return MasterDumpServlet.class;
548 }
549
550
551
552
553
554 @Override
555 protected void doMetrics() {
556 try {
557 if (assignmentManager != null) {
558 assignmentManager.updateRegionsInTransitionMetrics();
559 }
560 } catch (Throwable e) {
561 LOG.error("Couldn't update metrics: " + e.getMessage());
562 }
563 }
564
565 MetricsMaster getMasterMetrics() {
566 return metricsMaster;
567 }
568
569
570
571
572
573
574
575
576 void initializeZKBasedSystemTrackers() throws IOException,
577 InterruptedException, KeeperException, CoordinatedStateException {
578 this.balancer = LoadBalancerFactory.getLoadBalancer(conf);
579 this.loadBalancerTracker = new LoadBalancerTracker(zooKeeper, this);
580 this.loadBalancerTracker.start();
581 this.assignmentManager = new AssignmentManager(this, serverManager,
582 this.balancer, this.service, this.metricsMaster,
583 this.tableLockManager);
584 zooKeeper.registerListenerFirst(assignmentManager);
585
586 this.regionServerTracker = new RegionServerTracker(zooKeeper, this,
587 this.serverManager);
588 this.regionServerTracker.start();
589
590 this.drainingServerTracker = new DrainingServerTracker(zooKeeper, this,
591 this.serverManager);
592 this.drainingServerTracker.start();
593
594
595
596 boolean wasUp = this.clusterStatusTracker.isClusterUp();
597 if (!wasUp) this.clusterStatusTracker.setClusterUp();
598
599 LOG.info("Server active/primary master=" + this.serverName +
600 ", sessionid=0x" +
601 Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()) +
602 ", setting cluster-up flag (Was=" + wasUp + ")");
603
604
605 this.snapshotManager = new SnapshotManager();
606 this.mpmHost = new MasterProcedureManagerHost();
607 this.mpmHost.register(this.snapshotManager);
608 this.mpmHost.register(new MasterFlushTableProcedureManager());
609 this.mpmHost.loadProcedures(conf);
610 this.mpmHost.initialize(this, this.metricsMaster);
611 }
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633 private void finishActiveMasterInitialization(MonitoredTask status)
634 throws IOException, InterruptedException, KeeperException, CoordinatedStateException {
635
636 isActiveMaster = true;
637 Thread zombieDetector = new Thread(new InitializationMonitor(this));
638 zombieDetector.start();
639
640
641
642
643
644
645
646 status.setStatus("Initializing Master file system");
647
648 this.masterActiveTime = System.currentTimeMillis();
649
650 this.fileSystemManager = new MasterFileSystem(this, this);
651
652
653 this.tableDescriptors.setCacheOn();
654
655 this.tableDescriptors.get(TableName.META_TABLE_NAME).setRegionReplication(
656 conf.getInt(HConstants.META_REPLICAS_NUM, HConstants.DEFAULT_META_REPLICA_NUM));
657
658 if (preLoadTableDescriptors) {
659 status.setStatus("Pre-loading table descriptors");
660 this.tableDescriptors.getAll();
661 }
662
663
664 status.setStatus("Publishing Cluster ID in ZooKeeper");
665 ZKClusterId.setClusterId(this.zooKeeper, fileSystemManager.getClusterId());
666 this.serverManager = createServerManager(this, this);
667
668 setupClusterConnection();
669
670
671 this.tableLockManager.reapWriteLocks();
672
673 status.setStatus("Initializing ZK system trackers");
674 initializeZKBasedSystemTrackers();
675
676
677 status.setStatus("Initializing master coprocessors");
678 this.cpHost = new MasterCoprocessorHost(this, this.conf);
679
680
681 status.setStatus("Initializing master service threads");
682 startServiceThreads();
683
684
685 sleeper.skipSleepCycle();
686
687
688 this.serverManager.waitForRegionServers(status);
689
690 for (ServerName sn: this.regionServerTracker.getOnlineServers()) {
691
692 if (!this.serverManager.isServerOnline(sn)
693 && serverManager.checkAndRecordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD)) {
694 LOG.info("Registered server found up in zk but who has not yet reported in: " + sn);
695 }
696 }
697
698
699
700
701 Set<ServerName> previouslyFailedServers = this.fileSystemManager
702 .getFailedServersFromLogFolders();
703
704
705 this.fileSystemManager.removeStaleRecoveringRegionsFromZK(previouslyFailedServers);
706
707
708 ServerName oldMetaServerLocation = metaTableLocator.getMetaRegionLocation(this.getZooKeeper());
709 if (oldMetaServerLocation != null && previouslyFailedServers.contains(oldMetaServerLocation)) {
710 splitMetaLogBeforeAssignment(oldMetaServerLocation);
711
712
713 }
714 Set<ServerName> previouslyFailedMetaRSs = getPreviouselyFailedMetaServersFromZK();
715
716
717
718
719
720
721
722
723 previouslyFailedMetaRSs.addAll(previouslyFailedServers);
724
725 this.initializationBeforeMetaAssignment = true;
726
727
728 if (BaseLoadBalancer.tablesOnMaster(conf)) {
729 waitForServerOnline();
730 }
731
732
733 this.balancer.setClusterStatus(getClusterStatus());
734 this.balancer.setMasterServices(this);
735 this.balancer.initialize();
736
737
738
739 if(isStopped()) return;
740
741
742 status.setStatus("Assigning Meta Region");
743 assignMeta(status, previouslyFailedMetaRSs, HRegionInfo.DEFAULT_REPLICA_ID);
744
745
746 if(isStopped()) return;
747
748 status.setStatus("Submitting log splitting work for previously failed region servers");
749
750
751 for (ServerName tmpServer : previouslyFailedServers) {
752 this.serverManager.processDeadServer(tmpServer, true);
753 }
754
755
756
757
758 if (this.conf.getBoolean("hbase.MetaMigrationConvertingToPB", true)) {
759 MetaMigrationConvertingToPB.updateMetaIfNecessary(this);
760 }
761
762
763 status.setStatus("Starting assignment manager");
764 this.assignmentManager.joinCluster();
765
766
767 this.balancer.setClusterStatus(getClusterStatus());
768
769
770
771 status.setStatus("Starting balancer and catalog janitor");
772 this.clusterStatusChore = new ClusterStatusChore(this, balancer);
773 getChoreService().scheduleChore(clusterStatusChore);
774 this.balancerChore = new BalancerChore(this);
775 getChoreService().scheduleChore(balancerChore);
776 this.catalogJanitorChore = new CatalogJanitor(this, this);
777 getChoreService().scheduleChore(catalogJanitorChore);
778
779 status.setStatus("Starting namespace manager");
780 initNamespace();
781
782 if (this.cpHost != null) {
783 try {
784 this.cpHost.preMasterInitialization();
785 } catch (IOException e) {
786 LOG.error("Coprocessor preMasterInitialization() hook failed", e);
787 }
788 }
789
790 status.markComplete("Initialization successful");
791 LOG.info("Master has completed initialization");
792 configurationManager.registerObserver(this.balancer);
793 initialized = true;
794
795 status.setStatus("Starting quota manager");
796 initQuotaManager();
797
798
799 Set<ServerName> EMPTY_SET = new HashSet<ServerName>();
800 int numReplicas = conf.getInt(HConstants.META_REPLICAS_NUM,
801 HConstants.DEFAULT_META_REPLICA_NUM);
802 for (int i = 1; i < numReplicas; i++) {
803 assignMeta(status, EMPTY_SET, i);
804 }
805 unassignExcessMetaReplica(zooKeeper, numReplicas);
806
807
808
809
810 this.serverManager.clearDeadServersWithSameHostNameAndPortOfOnlineServer();
811
812
813 status.setStatus("Checking ZNode ACLs");
814 zooKeeper.checkAndSetZNodeAcls();
815
816 status.setStatus("Calling postStartMaster coprocessors");
817 if (this.cpHost != null) {
818
819 try {
820 this.cpHost.postStartMaster();
821 } catch (IOException ioe) {
822 LOG.error("Coprocessor postStartMaster() hook failed", ioe);
823 }
824 }
825
826 zombieDetector.interrupt();
827 }
828
829 private void initQuotaManager() throws IOException {
830 quotaManager = new MasterQuotaManager(this);
831 this.assignmentManager.setRegionStateListener((RegionStateListener) quotaManager);
832 quotaManager.start();
833 }
834
835
836
837
838
839
840
841
842
843 ServerManager createServerManager(final Server master,
844 final MasterServices services)
845 throws IOException {
846
847
848 return new ServerManager(master, services);
849 }
850
851 private void unassignExcessMetaReplica(ZooKeeperWatcher zkw, int numMetaReplicasConfigured) {
852
853
854 try {
855 List<String> metaReplicaZnodes = zooKeeper.getMetaReplicaNodes();
856 for (String metaReplicaZnode : metaReplicaZnodes) {
857 int replicaId = zooKeeper.getMetaReplicaIdFromZnode(metaReplicaZnode);
858 if (replicaId >= numMetaReplicasConfigured) {
859 RegionState r = MetaTableLocator.getMetaRegionState(zkw, replicaId);
860 LOG.info("Closing excess replica of meta region " + r.getRegion());
861
862 ServerManager.closeRegionSilentlyAndWait(getConnection(), r.getServerName(),
863 r.getRegion(), 30000);
864 ZKUtil.deleteNode(zkw, zkw.getZNodeForReplica(replicaId));
865 }
866 }
867 } catch (Exception ex) {
868
869
870 LOG.warn("Ignoring exception " + ex);
871 }
872 }
873
874
875
876
877
878
879
880
881
882
883 void assignMeta(MonitoredTask status, Set<ServerName> previouslyFailedMetaRSs, int replicaId)
884 throws InterruptedException, IOException, KeeperException {
885
886 int assigned = 0;
887 long timeout = this.conf.getLong("hbase.catalog.verification.timeout", 1000);
888 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) {
889 status.setStatus("Assigning hbase:meta region");
890 } else {
891 status.setStatus("Assigning hbase:meta region, replicaId " + replicaId);
892 }
893
894 RegionStates regionStates = assignmentManager.getRegionStates();
895 RegionState metaState = MetaTableLocator.getMetaRegionState(getZooKeeper(), replicaId);
896 HRegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(HRegionInfo.FIRST_META_REGIONINFO,
897 replicaId);
898 ServerName currentMetaServer = metaState.getServerName();
899 if (!ConfigUtil.useZKForAssignment(conf)) {
900 regionStates.createRegionState(hri, metaState.getState(),
901 currentMetaServer, null);
902 } else {
903 regionStates.createRegionState(hri);
904 }
905 boolean rit = this.assignmentManager.
906 processRegionInTransitionAndBlockUntilAssigned(hri);
907 boolean metaRegionLocation = metaTableLocator.verifyMetaRegionLocation(
908 this.getConnection(), this.getZooKeeper(), timeout, replicaId);
909 if (!metaRegionLocation || !metaState.isOpened()) {
910
911
912 assigned++;
913 if (!ConfigUtil.useZKForAssignment(conf)) {
914 assignMetaZkLess(regionStates, metaState, timeout, previouslyFailedMetaRSs);
915 } else if (!rit) {
916
917 if (currentMetaServer != null) {
918
919
920
921
922
923
924
925 if (serverManager.isServerOnline(currentMetaServer)) {
926 LOG.info("Forcing expire of " + currentMetaServer);
927 serverManager.expireServer(currentMetaServer);
928 }
929 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) {
930 splitMetaLogBeforeAssignment(currentMetaServer);
931 previouslyFailedMetaRSs.add(currentMetaServer);
932 }
933 }
934 assignmentManager.assignMeta(hri);
935 }
936 } else {
937
938 regionStates.updateRegionState(
939 HRegionInfo.FIRST_META_REGIONINFO, State.OPEN, currentMetaServer);
940 this.assignmentManager.regionOnline(
941 HRegionInfo.FIRST_META_REGIONINFO, currentMetaServer);
942 }
943
944 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) enableMeta(TableName.META_TABLE_NAME);
945
946 if ((RecoveryMode.LOG_REPLAY == this.getMasterFileSystem().getLogRecoveryMode())
947 && (!previouslyFailedMetaRSs.isEmpty())) {
948
949 status.setStatus("replaying log for Meta Region");
950 this.fileSystemManager.splitMetaLog(previouslyFailedMetaRSs);
951 }
952
953
954
955
956
957 if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) enableServerShutdownHandler(assigned != 0);
958 LOG.info("hbase:meta with replicaId " + replicaId + " assigned=" + assigned + ", rit=" + rit +
959 ", location=" + metaTableLocator.getMetaRegionLocation(this.getZooKeeper(), replicaId));
960 status.setStatus("META assigned.");
961 }
962
963 private void assignMetaZkLess(RegionStates regionStates, RegionState regionState, long timeout,
964 Set<ServerName> previouslyFailedRs) throws IOException, KeeperException {
965 ServerName currentServer = regionState.getServerName();
966 if (serverManager.isServerOnline(currentServer)) {
967 LOG.info("Meta was in transition on " + currentServer);
968 assignmentManager.processRegionInTransitionZkLess();
969 } else {
970 if (currentServer != null) {
971 if (regionState.getRegion().getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
972 splitMetaLogBeforeAssignment(currentServer);
973 regionStates.logSplit(HRegionInfo.FIRST_META_REGIONINFO);
974 previouslyFailedRs.add(currentServer);
975 }
976 }
977 LOG.info("Re-assigning hbase:meta, it was on " + currentServer);
978 regionStates.updateRegionState(regionState.getRegion(), State.OFFLINE);
979 assignmentManager.assignMeta(regionState.getRegion());
980 }
981 }
982
983 void initNamespace() throws IOException {
984
985 tableNamespaceManager = new TableNamespaceManager(this);
986 tableNamespaceManager.start();
987 }
988
989 boolean isCatalogJanitorEnabled() {
990 return catalogJanitorChore != null ?
991 catalogJanitorChore.getEnabled() : false;
992 }
993
994 private void splitMetaLogBeforeAssignment(ServerName currentMetaServer) throws IOException {
995 if (RecoveryMode.LOG_REPLAY == this.getMasterFileSystem().getLogRecoveryMode()) {
996
997 Set<HRegionInfo> regions = new HashSet<HRegionInfo>();
998 regions.add(HRegionInfo.FIRST_META_REGIONINFO);
999 this.fileSystemManager.prepareLogReplay(currentMetaServer, regions);
1000 } else {
1001
1002 this.fileSystemManager.splitMetaLog(currentMetaServer);
1003 }
1004 }
1005
1006 private void enableServerShutdownHandler(
1007 final boolean waitForMeta) throws IOException, InterruptedException {
1008
1009
1010
1011
1012
1013 if (!serverShutdownHandlerEnabled) {
1014 serverShutdownHandlerEnabled = true;
1015 this.serverManager.processQueuedDeadServers();
1016 }
1017
1018 if (waitForMeta) {
1019 metaTableLocator.waitMetaRegionLocation(this.getZooKeeper());
1020
1021
1022 this.assignmentManager.waitForAssignment(HRegionInfo.FIRST_META_REGIONINFO);
1023 }
1024 }
1025
1026 private void enableMeta(TableName metaTableName) {
1027 if (!this.assignmentManager.getTableStateManager().isTableState(metaTableName,
1028 ZooKeeperProtos.Table.State.ENABLED)) {
1029 this.assignmentManager.setEnabledTable(metaTableName);
1030 }
1031 }
1032
1033
1034
1035
1036
1037
1038 private Set<ServerName> getPreviouselyFailedMetaServersFromZK() throws KeeperException {
1039 Set<ServerName> result = new HashSet<ServerName>();
1040 String metaRecoveringZNode = ZKUtil.joinZNode(zooKeeper.recoveringRegionsZNode,
1041 HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
1042 List<String> regionFailedServers = ZKUtil.listChildrenNoWatch(zooKeeper, metaRecoveringZNode);
1043 if (regionFailedServers == null) return result;
1044
1045 for(String failedServer : regionFailedServers) {
1046 ServerName server = ServerName.parseServerName(failedServer);
1047 result.add(server);
1048 }
1049 return result;
1050 }
1051
1052 @Override
1053 public TableDescriptors getTableDescriptors() {
1054 return this.tableDescriptors;
1055 }
1056
1057 @Override
1058 public ServerManager getServerManager() {
1059 return this.serverManager;
1060 }
1061
1062 @Override
1063 public MasterFileSystem getMasterFileSystem() {
1064 return this.fileSystemManager;
1065 }
1066
1067
1068
1069
1070
1071
1072
1073
1074 private void startServiceThreads() throws IOException{
1075
1076 this.service.startExecutorService(ExecutorType.MASTER_OPEN_REGION,
1077 conf.getInt("hbase.master.executor.openregion.threads", 5));
1078 this.service.startExecutorService(ExecutorType.MASTER_CLOSE_REGION,
1079 conf.getInt("hbase.master.executor.closeregion.threads", 5));
1080 this.service.startExecutorService(ExecutorType.MASTER_SERVER_OPERATIONS,
1081 conf.getInt("hbase.master.executor.serverops.threads", 5));
1082 this.service.startExecutorService(ExecutorType.MASTER_META_SERVER_OPERATIONS,
1083 conf.getInt("hbase.master.executor.serverops.threads", 5));
1084 this.service.startExecutorService(ExecutorType.M_LOG_REPLAY_OPS,
1085 conf.getInt("hbase.master.executor.logreplayops.threads", 10));
1086
1087
1088
1089
1090
1091
1092 this.service.startExecutorService(ExecutorType.MASTER_TABLE_OPERATIONS, 1);
1093 startProcedureExecutor();
1094
1095
1096 int cleanerInterval = conf.getInt("hbase.master.cleaner.interval", 60 * 1000);
1097 this.logCleaner =
1098 new LogCleaner(cleanerInterval,
1099 this, conf, getMasterFileSystem().getFileSystem(),
1100 getMasterFileSystem().getOldLogDir());
1101 getChoreService().scheduleChore(logCleaner);
1102
1103
1104 Path archiveDir = HFileArchiveUtil.getArchivePath(conf);
1105 this.hfileCleaner = new HFileCleaner(cleanerInterval, this, conf, getMasterFileSystem()
1106 .getFileSystem(), archiveDir);
1107 getChoreService().scheduleChore(hfileCleaner);
1108 serviceStarted = true;
1109 if (LOG.isTraceEnabled()) {
1110 LOG.trace("Started service threads");
1111 }
1112 }
1113
1114 @Override
1115 protected void sendShutdownInterrupt() {
1116 super.sendShutdownInterrupt();
1117 stopProcedureExecutor();
1118 }
1119
1120 @Override
1121 protected void stopServiceThreads() {
1122 if (masterJettyServer != null) {
1123 LOG.info("Stopping master jetty server");
1124 try {
1125 masterJettyServer.stop();
1126 } catch (Exception e) {
1127 LOG.error("Failed to stop master jetty server", e);
1128 }
1129 }
1130 super.stopServiceThreads();
1131 stopChores();
1132
1133
1134
1135 if (!isAborted() && this.serverManager != null &&
1136 this.serverManager.isClusterShutdown()) {
1137 this.serverManager.letRegionServersShutdown();
1138 }
1139 if (LOG.isDebugEnabled()) {
1140 LOG.debug("Stopping service threads");
1141 }
1142
1143 if (this.logCleaner != null) this.logCleaner.cancel(true);
1144 if (this.hfileCleaner != null) this.hfileCleaner.cancel(true);
1145 if (this.quotaManager != null) this.quotaManager.stop();
1146 if (this.activeMasterManager != null) this.activeMasterManager.stop();
1147 if (this.serverManager != null) this.serverManager.stop();
1148 if (this.assignmentManager != null) this.assignmentManager.stop();
1149 if (this.fileSystemManager != null) this.fileSystemManager.stop();
1150 if (this.mpmHost != null) this.mpmHost.stop("server shutting down.");
1151 }
1152
1153
1154
1155
1156 @Override
1157 public boolean isMasterProcedureExecutorEnabled() {
1158 return (this.procedureConf == ProcedureConf.PROCEDURE_ENABLED);
1159 }
1160
1161 private void startProcedureExecutor() throws IOException {
1162 final MasterProcedureEnv procEnv = new MasterProcedureEnv(this);
1163 final Path logDir = new Path(fileSystemManager.getRootDir(),
1164 MasterProcedureConstants.MASTER_PROCEDURE_LOGDIR);
1165
1166 if (this.procedureConf == ProcedureConf.PROCEDURE_FULLY_DISABLED) {
1167
1168
1169
1170 try {
1171 fs.delete(logDir, true);
1172 LOG.warn("Procedure executor is disabled from configuartion. " +
1173 "All the state logs from procedure store were removed." +
1174 "You should check the cluster state using HBCK.");
1175 } catch (Exception e) {
1176
1177 LOG.error("Removing all the state logs from procedure store failed." +
1178 "You should check the cluster state using HBCK.");
1179 }
1180 return;
1181 }
1182
1183 procedureStore = new WALProcedureStore(conf, fileSystemManager.getFileSystem(), logDir,
1184 new MasterProcedureEnv.WALStoreLeaseRecovery(this));
1185
1186 procedureStore.registerListener(new MasterProcedureEnv.MasterProcedureStoreListener(this));
1187 procedureExecutor = new ProcedureExecutor(conf, procEnv, procedureStore,
1188 procEnv.getProcedureQueue());
1189
1190 final int numThreads = conf.getInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS,
1191 Math.max(Runtime.getRuntime().availableProcessors(),
1192 MasterProcedureConstants.DEFAULT_MIN_MASTER_PROCEDURE_THREADS));
1193 procedureStore.start(numThreads);
1194 procedureExecutor.start(numThreads);
1195 }
1196
1197 private void stopProcedureExecutor() {
1198 if (procedureExecutor != null) {
1199 procedureExecutor.stop();
1200 }
1201
1202 if (procedureStore != null) {
1203 procedureStore.stop(isAborted());
1204 }
1205 }
1206
1207 private void stopChores() {
1208 if (this.balancerChore != null) {
1209 this.balancerChore.cancel(true);
1210 }
1211 if (this.clusterStatusChore != null) {
1212 this.clusterStatusChore.cancel(true);
1213 }
1214 if (this.catalogJanitorChore != null) {
1215 this.catalogJanitorChore.cancel(true);
1216 }
1217 if (this.clusterStatusPublisherChore != null){
1218 clusterStatusPublisherChore.cancel(true);
1219 }
1220 }
1221
1222
1223
1224
1225
1226 InetAddress getRemoteInetAddress(final int port,
1227 final long serverStartCode) throws UnknownHostException {
1228
1229
1230 InetAddress ia = RpcServer.getRemoteIp();
1231
1232
1233
1234 if (ia == null && serverStartCode == startcode) {
1235 InetSocketAddress isa = rpcServices.getSocketAddress();
1236 if (isa != null && isa.getPort() == port) {
1237 ia = isa.getAddress();
1238 }
1239 }
1240 return ia;
1241 }
1242
1243
1244
1245
1246 private int getBalancerCutoffTime() {
1247 int balancerCutoffTime =
1248 getConfiguration().getInt("hbase.balancer.max.balancing", -1);
1249 if (balancerCutoffTime == -1) {
1250
1251 int balancerPeriod =
1252 getConfiguration().getInt("hbase.balancer.period", 300000);
1253 balancerCutoffTime = balancerPeriod;
1254
1255 if (balancerCutoffTime <= 0) balancerCutoffTime = balancerPeriod;
1256 }
1257 return balancerCutoffTime;
1258 }
1259
1260 public boolean balance() throws IOException {
1261
1262 if (!this.initialized) {
1263 LOG.debug("Master has not been initialized, don't run balancer.");
1264 return false;
1265 }
1266
1267 int maximumBalanceTime = getBalancerCutoffTime();
1268 synchronized (this.balancer) {
1269
1270 if (!this.loadBalancerTracker.isBalancerOn()) return false;
1271
1272 if (this.assignmentManager.getRegionStates().isRegionsInTransition()) {
1273 Map<String, RegionState> regionsInTransition =
1274 this.assignmentManager.getRegionStates().getRegionsInTransition();
1275 LOG.debug("Not running balancer because " + regionsInTransition.size() +
1276 " region(s) in transition: " + org.apache.commons.lang.StringUtils.
1277 abbreviate(regionsInTransition.toString(), 256));
1278 return false;
1279 }
1280 if (this.serverManager.areDeadServersInProgress()) {
1281 LOG.debug("Not running balancer because processing dead regionserver(s): " +
1282 this.serverManager.getDeadServers());
1283 return false;
1284 }
1285
1286 if (this.cpHost != null) {
1287 try {
1288 if (this.cpHost.preBalance()) {
1289 LOG.debug("Coprocessor bypassing balancer request");
1290 return false;
1291 }
1292 } catch (IOException ioe) {
1293 LOG.error("Error invoking master coprocessor preBalance()", ioe);
1294 return false;
1295 }
1296 }
1297
1298 Map<TableName, Map<ServerName, List<HRegionInfo>>> assignmentsByTable =
1299 this.assignmentManager.getRegionStates().getAssignmentsByTable();
1300
1301 List<RegionPlan> plans = new ArrayList<RegionPlan>();
1302
1303 this.balancer.setClusterStatus(getClusterStatus());
1304 for (Map<ServerName, List<HRegionInfo>> assignments : assignmentsByTable.values()) {
1305 List<RegionPlan> partialPlans = this.balancer.balanceCluster(assignments);
1306 if (partialPlans != null) plans.addAll(partialPlans);
1307 }
1308 long cutoffTime = System.currentTimeMillis() + maximumBalanceTime;
1309 int rpCount = 0;
1310 long totalRegPlanExecTime = 0;
1311 if (plans != null && !plans.isEmpty()) {
1312 for (RegionPlan plan: plans) {
1313 LOG.info("balance " + plan);
1314 long balStartTime = System.currentTimeMillis();
1315
1316 this.assignmentManager.balance(plan);
1317 totalRegPlanExecTime += System.currentTimeMillis()-balStartTime;
1318 rpCount++;
1319 if (rpCount < plans.size() &&
1320
1321 (System.currentTimeMillis() + (totalRegPlanExecTime / rpCount)) > cutoffTime) {
1322
1323 LOG.debug("No more balancing till next balance run; maximumBalanceTime=" +
1324 maximumBalanceTime);
1325 break;
1326 }
1327 }
1328 }
1329 if (this.cpHost != null) {
1330 try {
1331 this.cpHost.postBalance(rpCount < plans.size() ? plans.subList(0, rpCount) : plans);
1332 } catch (IOException ioe) {
1333
1334 LOG.error("Error invoking master coprocessor postBalance()", ioe);
1335 }
1336 }
1337 }
1338
1339
1340 return true;
1341 }
1342
1343
1344
1345
1346 String getClientIdAuditPrefix() {
1347 return "Client=" + RpcServer.getRequestUserName() + "/" + RpcServer.getRemoteAddress();
1348 }
1349
1350
1351
1352
1353
1354
1355
1356 public void setCatalogJanitorEnabled(final boolean b) {
1357 this.catalogJanitorChore.setEnabled(b);
1358 }
1359
1360 @Override
1361 public void dispatchMergingRegions(final HRegionInfo region_a,
1362 final HRegionInfo region_b, final boolean forcible) throws IOException {
1363 checkInitialized();
1364 this.service.submit(new DispatchMergingRegionHandler(this,
1365 this.catalogJanitorChore, region_a, region_b, forcible));
1366 }
1367
1368 void move(final byte[] encodedRegionName,
1369 final byte[] destServerName) throws HBaseIOException {
1370 RegionState regionState = assignmentManager.getRegionStates().
1371 getRegionState(Bytes.toString(encodedRegionName));
1372 if (regionState == null) {
1373 throw new UnknownRegionException(Bytes.toStringBinary(encodedRegionName));
1374 }
1375
1376 HRegionInfo hri = regionState.getRegion();
1377 ServerName dest;
1378 if (destServerName == null || destServerName.length == 0) {
1379 LOG.info("Passed destination servername is null/empty so " +
1380 "choosing a server at random");
1381 final List<ServerName> destServers = this.serverManager.createDestinationServersList(
1382 regionState.getServerName());
1383 dest = balancer.randomAssignment(hri, destServers);
1384 if (dest == null) {
1385 LOG.debug("Unable to determine a plan to assign " + hri);
1386 return;
1387 }
1388 } else {
1389 dest = ServerName.valueOf(Bytes.toString(destServerName));
1390 if (dest.equals(serverName) && balancer instanceof BaseLoadBalancer
1391 && !((BaseLoadBalancer)balancer).shouldBeOnMaster(hri)) {
1392
1393
1394
1395 LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
1396 + " to avoid unnecessary region moving later by load balancer,"
1397 + " because it should not be on master");
1398 return;
1399 }
1400 }
1401
1402 if (dest.equals(regionState.getServerName())) {
1403 LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
1404 + " because region already assigned to the same server " + dest + ".");
1405 return;
1406 }
1407
1408
1409 RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), dest);
1410
1411 try {
1412 checkInitialized();
1413 if (this.cpHost != null) {
1414 if (this.cpHost.preMove(hri, rp.getSource(), rp.getDestination())) {
1415 return;
1416 }
1417 }
1418
1419
1420
1421 serverManager.sendRegionWarmup(rp.getDestination(), hri);
1422
1423 LOG.info(getClientIdAuditPrefix() + " move " + rp + ", running balancer");
1424 this.assignmentManager.balance(rp);
1425 if (this.cpHost != null) {
1426 this.cpHost.postMove(hri, rp.getSource(), rp.getDestination());
1427 }
1428 } catch (IOException ioe) {
1429 if (ioe instanceof HBaseIOException) {
1430 throw (HBaseIOException)ioe;
1431 }
1432 throw new HBaseIOException(ioe);
1433 }
1434 }
1435
1436 @Override
1437 public long createTable(
1438 final HTableDescriptor hTableDescriptor,
1439 final byte [][] splitKeys,
1440 final long nonceGroup,
1441 final long nonce) throws IOException {
1442 if (isStopped()) {
1443 throw new MasterNotRunningException();
1444 }
1445
1446 TableName tableName = hTableDescriptor.getTableName();
1447 String namespace = tableName.getNamespaceAsString();
1448 ensureNamespaceExists(namespace);
1449
1450 HRegionInfo[] newRegions = ModifyRegionUtils.createHRegionInfos(hTableDescriptor, splitKeys);
1451 checkInitialized();
1452 sanityCheckTableDescriptor(hTableDescriptor);
1453 if (cpHost != null) {
1454 cpHost.preCreateTable(hTableDescriptor, newRegions);
1455 }
1456 LOG.info(getClientIdAuditPrefix() + " create " + hTableDescriptor);
1457
1458 long procId = -1;
1459 if (isMasterProcedureExecutorEnabled()) {
1460
1461
1462 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch();
1463 procId = this.procedureExecutor.submitProcedure(
1464 new CreateTableProcedure(
1465 procedureExecutor.getEnvironment(), hTableDescriptor, newRegions, latch),
1466 nonceGroup,
1467 nonce);
1468 latch.await();
1469 } else {
1470 try {
1471 this.quotaManager.checkNamespaceTableAndRegionQuota(tableName, newRegions.length);
1472 this.service.submit(new CreateTableHandler(this, this.fileSystemManager, hTableDescriptor,
1473 conf, newRegions, this).prepare());
1474 } catch (IOException e) {
1475 this.quotaManager.removeTableFromNamespaceQuota(tableName);
1476 LOG.error("Exception occurred while creating the table " + tableName.getNameAsString(), e);
1477 throw e;
1478 }
1479 }
1480
1481 if (cpHost != null) {
1482 cpHost.postCreateTable(hTableDescriptor, newRegions);
1483 }
1484
1485 return procId;
1486 }
1487
1488
1489
1490
1491
1492
1493 private void sanityCheckTableDescriptor(final HTableDescriptor htd) throws IOException {
1494 final String CONF_KEY = "hbase.table.sanity.checks";
1495 boolean logWarn = false;
1496 if (!conf.getBoolean(CONF_KEY, true)) {
1497 logWarn = true;
1498 }
1499 String tableVal = htd.getConfigurationValue(CONF_KEY);
1500 if (tableVal != null && !Boolean.valueOf(tableVal)) {
1501 logWarn = true;
1502 }
1503
1504
1505 long maxFileSizeLowerLimit = 2 * 1024 * 1024L;
1506 long maxFileSize = htd.getMaxFileSize();
1507 if (maxFileSize < 0) {
1508 maxFileSize = conf.getLong(HConstants.HREGION_MAX_FILESIZE, maxFileSizeLowerLimit);
1509 }
1510 if (maxFileSize < conf.getLong("hbase.hregion.max.filesize.limit", maxFileSizeLowerLimit)) {
1511 String message = "MAX_FILESIZE for table descriptor or "
1512 + "\"hbase.hregion.max.filesize\" (" + maxFileSize
1513 + ") is too small, which might cause over splitting into unmanageable "
1514 + "number of regions.";
1515 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1516 }
1517
1518
1519 long flushSizeLowerLimit = 1024 * 1024L;
1520 long flushSize = htd.getMemStoreFlushSize();
1521 if (flushSize < 0) {
1522 flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, flushSizeLowerLimit);
1523 }
1524 if (flushSize < conf.getLong("hbase.hregion.memstore.flush.size.limit", flushSizeLowerLimit)) {
1525 String message = "MEMSTORE_FLUSHSIZE for table descriptor or "
1526 + "\"hbase.hregion.memstore.flush.size\" ("+flushSize+") is too small, which might cause"
1527 + " very frequent flushing.";
1528 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1529 }
1530
1531
1532 try {
1533 checkClassLoading(conf, htd);
1534 } catch (Exception ex) {
1535 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, ex.getMessage(), null);
1536 }
1537
1538
1539 try {
1540 checkCompression(htd);
1541 } catch (IOException e) {
1542 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e);
1543 }
1544
1545
1546 try {
1547 checkEncryption(conf, htd);
1548 } catch (IOException e) {
1549 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e);
1550 }
1551
1552
1553 if (htd.getColumnFamilies().length == 0) {
1554 String message = "Table should have at least one column family.";
1555 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1556 }
1557
1558 for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1559 if (hcd.getTimeToLive() <= 0) {
1560 String message = "TTL for column family " + hcd.getNameAsString() + " must be positive.";
1561 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1562 }
1563
1564
1565 if (hcd.getBlocksize() < 1024 || hcd.getBlocksize() > 16 * 1024 * 1024) {
1566 String message = "Block size for column family " + hcd.getNameAsString()
1567 + " must be between 1K and 16MB.";
1568 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1569 }
1570
1571
1572 if (hcd.getMinVersions() < 0) {
1573 String message = "Min versions for column family " + hcd.getNameAsString()
1574 + " must be positive.";
1575 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1576 }
1577
1578
1579
1580
1581
1582 if (hcd.getMinVersions() > hcd.getMaxVersions()) {
1583 String message = "Min versions for column family " + hcd.getNameAsString()
1584 + " must be less than the Max versions.";
1585 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1586 }
1587
1588
1589 if (hcd.getScope() < 0) {
1590 String message = "Replication scope for column family "
1591 + hcd.getNameAsString() + " must be positive.";
1592 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1593 }
1594
1595
1596 }
1597 }
1598
1599
1600 private static void warnOrThrowExceptionForFailure(boolean logWarn, String confKey,
1601 String message, Exception cause) throws IOException {
1602 if (!logWarn) {
1603 throw new DoNotRetryIOException(message + " Set " + confKey +
1604 " to false at conf or table descriptor if you want to bypass sanity checks", cause);
1605 }
1606 LOG.warn(message);
1607 }
1608
1609 private void startActiveMasterManager(int infoPort) throws KeeperException {
1610 String backupZNode = ZKUtil.joinZNode(
1611 zooKeeper.backupMasterAddressesZNode, serverName.toString());
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622 LOG.info("Adding backup master ZNode " + backupZNode);
1623 if (!MasterAddressTracker.setMasterAddress(zooKeeper, backupZNode,
1624 serverName, infoPort)) {
1625 LOG.warn("Failed create of " + backupZNode + " by " + serverName);
1626 }
1627
1628 activeMasterManager.setInfoPort(infoPort);
1629
1630 Threads.setDaemonThreadRunning(new Thread(new Runnable() {
1631 @Override
1632 public void run() {
1633 int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT,
1634 HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
1635
1636 if (conf.getBoolean(HConstants.MASTER_TYPE_BACKUP,
1637 HConstants.DEFAULT_MASTER_TYPE_BACKUP)) {
1638 LOG.debug("HMaster started in backup mode. "
1639 + "Stalling until master znode is written.");
1640
1641
1642 while (!activeMasterManager.hasActiveMaster()) {
1643 LOG.debug("Waiting for master address ZNode to be written "
1644 + "(Also watching cluster state node)");
1645 Threads.sleep(timeout);
1646 }
1647 }
1648 MonitoredTask status = TaskMonitor.get().createStatus("Master startup");
1649 status.setDescription("Master startup");
1650 try {
1651 if (activeMasterManager.blockUntilBecomingActiveMaster(timeout, status)) {
1652 finishActiveMasterInitialization(status);
1653 }
1654 } catch (Throwable t) {
1655 status.setStatus("Failed to become active: " + t.getMessage());
1656 LOG.fatal("Failed to become active master", t);
1657
1658 if (t instanceof NoClassDefFoundError &&
1659 t.getMessage().contains("org/apache/hadoop/hdfs/protocol/FSConstants$SafeModeAction")) {
1660
1661 abort("HBase is having a problem with its Hadoop jars. You may need to "
1662 + "recompile HBase against Hadoop version "
1663 + org.apache.hadoop.util.VersionInfo.getVersion()
1664 + " or change your hadoop jars to start properly", t);
1665 } else {
1666 abort("Unhandled exception. Starting shutdown.", t);
1667 }
1668 } finally {
1669 status.cleanup();
1670 }
1671 }
1672 }, getServerName().toShortString() + ".activeMasterManager"));
1673 }
1674
1675 private void checkCompression(final HTableDescriptor htd)
1676 throws IOException {
1677 if (!this.masterCheckCompression) return;
1678 for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1679 checkCompression(hcd);
1680 }
1681 }
1682
1683 private void checkCompression(final HColumnDescriptor hcd)
1684 throws IOException {
1685 if (!this.masterCheckCompression) return;
1686 CompressionTest.testCompression(hcd.getCompression());
1687 CompressionTest.testCompression(hcd.getCompactionCompression());
1688 }
1689
1690 private void checkEncryption(final Configuration conf, final HTableDescriptor htd)
1691 throws IOException {
1692 if (!this.masterCheckEncryption) return;
1693 for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1694 checkEncryption(conf, hcd);
1695 }
1696 }
1697
1698 private void checkEncryption(final Configuration conf, final HColumnDescriptor hcd)
1699 throws IOException {
1700 if (!this.masterCheckEncryption) return;
1701 EncryptionTest.testEncryption(conf, hcd.getEncryptionType(), hcd.getEncryptionKey());
1702 }
1703
1704 private void checkClassLoading(final Configuration conf, final HTableDescriptor htd)
1705 throws IOException {
1706 RegionSplitPolicy.getSplitPolicyClass(htd, conf);
1707 RegionCoprocessorHost.testTableCoprocessorAttrs(conf, htd);
1708 }
1709
1710 private static boolean isCatalogTable(final TableName tableName) {
1711 return tableName.equals(TableName.META_TABLE_NAME);
1712 }
1713
1714 @Override
1715 public long deleteTable(
1716 final TableName tableName,
1717 final long nonceGroup,
1718 final long nonce) throws IOException {
1719 checkInitialized();
1720 if (cpHost != null) {
1721 cpHost.preDeleteTable(tableName);
1722 }
1723 LOG.info(getClientIdAuditPrefix() + " delete " + tableName);
1724
1725 long procId = -1;
1726 if (isMasterProcedureExecutorEnabled()) {
1727
1728 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch();
1729 procId = this.procedureExecutor.submitProcedure(
1730 new DeleteTableProcedure(procedureExecutor.getEnvironment(), tableName, latch),
1731 nonceGroup,
1732 nonce);
1733 latch.await();
1734 } else {
1735 this.service.submit(new DeleteTableHandler(tableName, this, this).prepare());
1736 }
1737
1738 if (cpHost != null) {
1739 cpHost.postDeleteTable(tableName);
1740 }
1741
1742 return procId;
1743 }
1744
1745 @Override
1746 public void truncateTable(
1747 final TableName tableName,
1748 final boolean preserveSplits,
1749 final long nonceGroup,
1750 final long nonce) throws IOException {
1751 checkInitialized();
1752 if (cpHost != null) {
1753 cpHost.preTruncateTable(tableName);
1754 }
1755 LOG.info(getClientIdAuditPrefix() + " truncate " + tableName);
1756
1757 if (isMasterProcedureExecutorEnabled()) {
1758 long procId = this.procedureExecutor.submitProcedure(
1759 new TruncateTableProcedure(procedureExecutor.getEnvironment(), tableName, preserveSplits),
1760 nonceGroup,
1761 nonce);
1762 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1763 } else {
1764 TruncateTableHandler handler =
1765 new TruncateTableHandler(tableName, this, this, preserveSplits);
1766 handler.prepare();
1767 handler.process();
1768 }
1769
1770 if (cpHost != null) {
1771 cpHost.postTruncateTable(tableName);
1772 }
1773 }
1774
1775 @Override
1776 public void addColumn(
1777 final TableName tableName,
1778 final HColumnDescriptor columnDescriptor,
1779 final long nonceGroup,
1780 final long nonce)
1781 throws IOException {
1782 checkInitialized();
1783 checkCompression(columnDescriptor);
1784 checkEncryption(conf, columnDescriptor);
1785 if (cpHost != null) {
1786 if (cpHost.preAddColumn(tableName, columnDescriptor)) {
1787 return;
1788 }
1789 }
1790
1791 LOG.info(getClientIdAuditPrefix() + " add " + columnDescriptor);
1792
1793 if (isMasterProcedureExecutorEnabled()) {
1794
1795 long procId = this.procedureExecutor.submitProcedure(
1796 new AddColumnFamilyProcedure(
1797 procedureExecutor.getEnvironment(), tableName, columnDescriptor),
1798 nonceGroup,
1799 nonce);
1800 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1801 } else {
1802 new TableAddFamilyHandler(tableName, columnDescriptor, this, this).prepare().process();
1803 }
1804 if (cpHost != null) {
1805 cpHost.postAddColumn(tableName, columnDescriptor);
1806 }
1807 }
1808
1809 @Override
1810 public void modifyColumn(
1811 final TableName tableName,
1812 final HColumnDescriptor descriptor,
1813 final long nonceGroup,
1814 final long nonce)
1815 throws IOException {
1816 checkInitialized();
1817 checkCompression(descriptor);
1818 checkEncryption(conf, descriptor);
1819 if (cpHost != null) {
1820 if (cpHost.preModifyColumn(tableName, descriptor)) {
1821 return;
1822 }
1823 }
1824 LOG.info(getClientIdAuditPrefix() + " modify " + descriptor);
1825
1826 if (isMasterProcedureExecutorEnabled()) {
1827
1828 long procId = this.procedureExecutor.submitProcedure(
1829 new ModifyColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, descriptor),
1830 nonceGroup,
1831 nonce);
1832 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1833 } else {
1834 new TableModifyFamilyHandler(tableName, descriptor, this, this).prepare().process();
1835 }
1836
1837 if (cpHost != null) {
1838 cpHost.postModifyColumn(tableName, descriptor);
1839 }
1840 }
1841
1842 @Override
1843 public void deleteColumn(
1844 final TableName tableName,
1845 final byte[] columnName,
1846 final long nonceGroup,
1847 final long nonce)
1848 throws IOException {
1849 checkInitialized();
1850 if (cpHost != null) {
1851 if (cpHost.preDeleteColumn(tableName, columnName)) {
1852 return;
1853 }
1854 }
1855 LOG.info(getClientIdAuditPrefix() + " delete " + Bytes.toString(columnName));
1856
1857 if (isMasterProcedureExecutorEnabled()) {
1858
1859 long procId = this.procedureExecutor.submitProcedure(
1860 new DeleteColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, columnName),
1861 nonceGroup,
1862 nonce);
1863 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1864 } else {
1865 new TableDeleteFamilyHandler(tableName, columnName, this, this).prepare().process();
1866 }
1867
1868 if (cpHost != null) {
1869 cpHost.postDeleteColumn(tableName, columnName);
1870 }
1871 }
1872
1873 @Override
1874 public long enableTable(
1875 final TableName tableName,
1876 final long nonceGroup,
1877 final long nonce) throws IOException {
1878 checkInitialized();
1879 if (cpHost != null) {
1880 cpHost.preEnableTable(tableName);
1881 }
1882 LOG.info(getClientIdAuditPrefix() + " enable " + tableName);
1883
1884 long procId = -1;
1885 if (isMasterProcedureExecutorEnabled()) {
1886
1887 final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
1888 procId = this.procedureExecutor.submitProcedure(
1889 new EnableTableProcedure(
1890 procedureExecutor.getEnvironment(), tableName, false, prepareLatch),
1891 nonceGroup,
1892 nonce);
1893 this.procedureExecutor.submitProcedure(new EnableTableProcedure(procedureExecutor
1894 .getEnvironment(), tableName, false, prepareLatch));
1895
1896
1897
1898
1899 prepareLatch.await();
1900 } else {
1901 this.service.submit(new EnableTableHandler(this, tableName,
1902 assignmentManager, tableLockManager, false).prepare());
1903 }
1904
1905 if (cpHost != null) {
1906 cpHost.postEnableTable(tableName);
1907 }
1908
1909 return procId;
1910 }
1911
1912 @Override
1913 public long disableTable(
1914 final TableName tableName,
1915 final long nonceGroup,
1916 final long nonce) throws IOException {
1917 checkInitialized();
1918 if (cpHost != null) {
1919 cpHost.preDisableTable(tableName);
1920 }
1921 LOG.info(getClientIdAuditPrefix() + " disable " + tableName);
1922
1923 long procId = -1;
1924 if (isMasterProcedureExecutorEnabled()) {
1925
1926 final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
1927 procId = this.procedureExecutor.submitProcedure(
1928 new DisableTableProcedure(
1929 procedureExecutor.getEnvironment(), tableName, false, prepareLatch),
1930 nonceGroup,
1931 nonce);
1932
1933
1934
1935
1936 prepareLatch.await();
1937 } else {
1938 this.service.submit(new DisableTableHandler(this, tableName,
1939 assignmentManager, tableLockManager, false).prepare());
1940 }
1941
1942 if (cpHost != null) {
1943 cpHost.postDisableTable(tableName);
1944 }
1945
1946 return procId;
1947 }
1948
1949
1950
1951
1952
1953
1954
1955 @VisibleForTesting
1956 Pair<HRegionInfo, ServerName> getTableRegionForRow(
1957 final TableName tableName, final byte [] rowKey)
1958 throws IOException {
1959 final AtomicReference<Pair<HRegionInfo, ServerName>> result =
1960 new AtomicReference<Pair<HRegionInfo, ServerName>>(null);
1961
1962 MetaScannerVisitor visitor =
1963 new MetaScannerVisitorBase() {
1964 @Override
1965 public boolean processRow(Result data) throws IOException {
1966 if (data == null || data.size() <= 0) {
1967 return true;
1968 }
1969 Pair<HRegionInfo, ServerName> pair = HRegionInfo.getHRegionInfoAndServerName(data);
1970 if (pair == null) {
1971 return false;
1972 }
1973 if (!pair.getFirst().getTable().equals(tableName)) {
1974 return false;
1975 }
1976 result.set(pair);
1977 return true;
1978 }
1979 };
1980
1981 MetaScanner.metaScan(clusterConnection, visitor, tableName, rowKey, 1);
1982 return result.get();
1983 }
1984
1985 @Override
1986 public void modifyTable(
1987 final TableName tableName,
1988 final HTableDescriptor descriptor,
1989 final long nonceGroup,
1990 final long nonce)
1991 throws IOException {
1992 checkInitialized();
1993 sanityCheckTableDescriptor(descriptor);
1994 if (cpHost != null) {
1995 cpHost.preModifyTable(tableName, descriptor);
1996 }
1997
1998 LOG.info(getClientIdAuditPrefix() + " modify " + tableName);
1999
2000 if (isMasterProcedureExecutorEnabled()) {
2001
2002 long procId = this.procedureExecutor.submitProcedure(
2003 new ModifyTableProcedure(procedureExecutor.getEnvironment(), descriptor),
2004 nonceGroup,
2005 nonce);
2006 ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
2007 } else {
2008 new ModifyTableHandler(tableName, descriptor, this, this).prepare().process();
2009 }
2010
2011 if (cpHost != null) {
2012 cpHost.postModifyTable(tableName, descriptor);
2013 }
2014 }
2015
2016 @Override
2017 public void checkTableModifiable(final TableName tableName)
2018 throws IOException, TableNotFoundException, TableNotDisabledException {
2019 if (isCatalogTable(tableName)) {
2020 throw new IOException("Can't modify catalog tables");
2021 }
2022 if (!MetaTableAccessor.tableExists(getConnection(), tableName)) {
2023 throw new TableNotFoundException(tableName);
2024 }
2025 if (!getAssignmentManager().getTableStateManager().
2026 isTableState(tableName, ZooKeeperProtos.Table.State.DISABLED)) {
2027 throw new TableNotDisabledException(tableName);
2028 }
2029 }
2030
2031
2032
2033
2034 public ClusterStatus getClusterStatus() throws InterruptedIOException {
2035
2036 List<String> backupMasterStrings;
2037 try {
2038 backupMasterStrings = ZKUtil.listChildrenNoWatch(this.zooKeeper,
2039 this.zooKeeper.backupMasterAddressesZNode);
2040 } catch (KeeperException e) {
2041 LOG.warn(this.zooKeeper.prefix("Unable to list backup servers"), e);
2042 backupMasterStrings = null;
2043 }
2044
2045 List<ServerName> backupMasters = null;
2046 if (backupMasterStrings != null && !backupMasterStrings.isEmpty()) {
2047 backupMasters = new ArrayList<ServerName>(backupMasterStrings.size());
2048 for (String s: backupMasterStrings) {
2049 try {
2050 byte [] bytes;
2051 try {
2052 bytes = ZKUtil.getData(this.zooKeeper, ZKUtil.joinZNode(
2053 this.zooKeeper.backupMasterAddressesZNode, s));
2054 } catch (InterruptedException e) {
2055 throw new InterruptedIOException();
2056 }
2057 if (bytes != null) {
2058 ServerName sn;
2059 try {
2060 sn = ServerName.parseFrom(bytes);
2061 } catch (DeserializationException e) {
2062 LOG.warn("Failed parse, skipping registering backup server", e);
2063 continue;
2064 }
2065 backupMasters.add(sn);
2066 }
2067 } catch (KeeperException e) {
2068 LOG.warn(this.zooKeeper.prefix("Unable to get information about " +
2069 "backup servers"), e);
2070 }
2071 }
2072 Collections.sort(backupMasters, new Comparator<ServerName>() {
2073 @Override
2074 public int compare(ServerName s1, ServerName s2) {
2075 return s1.getServerName().compareTo(s2.getServerName());
2076 }});
2077 }
2078
2079 String clusterId = fileSystemManager != null ?
2080 fileSystemManager.getClusterId().toString() : null;
2081 Map<String, RegionState> regionsInTransition = assignmentManager != null ?
2082 assignmentManager.getRegionStates().getRegionsInTransition() : null;
2083 String[] coprocessors = cpHost != null ? getMasterCoprocessors() : null;
2084 boolean balancerOn = loadBalancerTracker != null ?
2085 loadBalancerTracker.isBalancerOn() : false;
2086 Map<ServerName, ServerLoad> onlineServers = null;
2087 Set<ServerName> deadServers = null;
2088 if (serverManager != null) {
2089 deadServers = serverManager.getDeadServers().copyServerNames();
2090 onlineServers = serverManager.getOnlineServers();
2091 }
2092 return new ClusterStatus(VersionInfo.getVersion(), clusterId,
2093 onlineServers, deadServers, serverName, backupMasters,
2094 regionsInTransition, coprocessors, balancerOn);
2095 }
2096
2097
2098
2099
2100
2101
2102
2103
2104 public static String getLoadedCoprocessors() {
2105 return CoprocessorHost.getLoadedCoprocessors().toString();
2106 }
2107
2108
2109
2110
2111 public long getMasterStartTime() {
2112 return startcode;
2113 }
2114
2115
2116
2117
2118 public long getMasterActiveTime() {
2119 return masterActiveTime;
2120 }
2121
2122 public int getRegionServerInfoPort(final ServerName sn) {
2123 RegionServerInfo info = this.regionServerTracker.getRegionServerInfo(sn);
2124 if (info == null || info.getInfoPort() == 0) {
2125 return conf.getInt(HConstants.REGIONSERVER_INFO_PORT,
2126 HConstants.DEFAULT_REGIONSERVER_INFOPORT);
2127 }
2128 return info.getInfoPort();
2129 }
2130
2131
2132
2133
2134 public String[] getMasterCoprocessors() {
2135 Set<String> masterCoprocessors = getMasterCoprocessorHost().getCoprocessors();
2136 return masterCoprocessors.toArray(new String[masterCoprocessors.size()]);
2137 }
2138
2139 @Override
2140 public void abort(final String msg, final Throwable t) {
2141 if (isAborted() || isStopped()) {
2142 return;
2143 }
2144 if (cpHost != null) {
2145
2146 LOG.fatal("Master server abort: loaded coprocessors are: " +
2147 getLoadedCoprocessors());
2148 }
2149 if (t != null) LOG.fatal(msg, t);
2150 stop(msg);
2151 }
2152
2153 @Override
2154 public ZooKeeperWatcher getZooKeeper() {
2155 return zooKeeper;
2156 }
2157
2158 @Override
2159 public MasterCoprocessorHost getMasterCoprocessorHost() {
2160 return cpHost;
2161 }
2162
2163 @Override
2164 public MasterQuotaManager getMasterQuotaManager() {
2165 return quotaManager;
2166 }
2167
2168 @Override
2169 public ProcedureExecutor<MasterProcedureEnv> getMasterProcedureExecutor() {
2170 return procedureExecutor;
2171 }
2172
2173 @Override
2174 public ServerName getServerName() {
2175 return this.serverName;
2176 }
2177
2178 @Override
2179 public AssignmentManager getAssignmentManager() {
2180 return this.assignmentManager;
2181 }
2182
2183 public MemoryBoundedLogMessageBuffer getRegionServerFatalLogBuffer() {
2184 return rsFatals;
2185 }
2186
2187 public void shutdown() {
2188 if (cpHost != null) {
2189 try {
2190 cpHost.preShutdown();
2191 } catch (IOException ioe) {
2192 LOG.error("Error call master coprocessor preShutdown()", ioe);
2193 }
2194 }
2195
2196 if (this.serverManager != null) {
2197 this.serverManager.shutdownCluster();
2198 }
2199 if (this.clusterStatusTracker != null){
2200 try {
2201 this.clusterStatusTracker.setClusterDown();
2202 } catch (KeeperException e) {
2203 LOG.error("ZooKeeper exception trying to set cluster as down in ZK", e);
2204 }
2205 }
2206 }
2207
2208 public void stopMaster() {
2209 if (cpHost != null) {
2210 try {
2211 cpHost.preStopMaster();
2212 } catch (IOException ioe) {
2213 LOG.error("Error call master coprocessor preStopMaster()", ioe);
2214 }
2215 }
2216 stop("Stopped by " + Thread.currentThread().getName());
2217 }
2218
2219 void checkServiceStarted() throws ServerNotRunningYetException {
2220 if (!serviceStarted) {
2221 throw new ServerNotRunningYetException("Server is not running yet");
2222 }
2223 }
2224
2225 void checkInitialized() throws PleaseHoldException, ServerNotRunningYetException {
2226 checkServiceStarted();
2227 if (!this.initialized) {
2228 throw new PleaseHoldException("Master is initializing");
2229 }
2230 }
2231
2232 void checkNamespaceManagerReady() throws IOException {
2233 checkInitialized();
2234 if (tableNamespaceManager == null ||
2235 !tableNamespaceManager.isTableAvailableAndInitialized()) {
2236 throw new IOException("Table Namespace Manager not ready yet, try again later");
2237 }
2238 }
2239
2240
2241
2242
2243
2244
2245
2246
2247 public boolean isActiveMaster() {
2248 return isActiveMaster;
2249 }
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260 @Override
2261 public boolean isInitialized() {
2262 return initialized;
2263 }
2264
2265
2266
2267
2268
2269
2270 @Override
2271 public boolean isServerShutdownHandlerEnabled() {
2272 return this.serverShutdownHandlerEnabled;
2273 }
2274
2275
2276
2277
2278
2279 public boolean isInitializationStartsMetaRegionAssignment() {
2280 return this.initializationBeforeMetaAssignment;
2281 }
2282
2283 public void assignRegion(HRegionInfo hri) {
2284 assignmentManager.assign(hri, true);
2285 }
2286
2287
2288
2289
2290
2291
2292
2293 public double getAverageLoad() {
2294 if (this.assignmentManager == null) {
2295 return 0;
2296 }
2297
2298 RegionStates regionStates = this.assignmentManager.getRegionStates();
2299 if (regionStates == null) {
2300 return 0;
2301 }
2302 return regionStates.getAverageLoad();
2303 }
2304
2305 @Override
2306 public boolean registerService(Service instance) {
2307
2308
2309
2310 Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType();
2311 if (coprocessorServiceHandlers.containsKey(serviceDesc.getFullName())) {
2312 LOG.error("Coprocessor service "+serviceDesc.getFullName()+
2313 " already registered, rejecting request from "+instance
2314 );
2315 return false;
2316 }
2317
2318 coprocessorServiceHandlers.put(serviceDesc.getFullName(), instance);
2319 if (LOG.isDebugEnabled()) {
2320 LOG.debug("Registered master coprocessor service: service="+serviceDesc.getFullName());
2321 }
2322 return true;
2323 }
2324
2325
2326
2327
2328
2329
2330
2331 public static HMaster constructMaster(Class<? extends HMaster> masterClass,
2332 final Configuration conf, final CoordinatedStateManager cp) {
2333 try {
2334 Constructor<? extends HMaster> c =
2335 masterClass.getConstructor(Configuration.class, CoordinatedStateManager.class);
2336 return c.newInstance(conf, cp);
2337 } catch (InvocationTargetException ite) {
2338 Throwable target = ite.getTargetException() != null?
2339 ite.getTargetException(): ite;
2340 if (target.getCause() != null) target = target.getCause();
2341 throw new RuntimeException("Failed construction of Master: " +
2342 masterClass.toString(), target);
2343 } catch (Exception e) {
2344 throw new RuntimeException("Failed construction of Master: " +
2345 masterClass.toString() + ((e.getCause() != null)?
2346 e.getCause().getMessage(): ""), e);
2347 }
2348 }
2349
2350
2351
2352
2353 public static void main(String [] args) {
2354 VersionInfo.logVersion();
2355 new HMasterCommandLine(HMaster.class).doMain(args);
2356 }
2357
2358 public HFileCleaner getHFileCleaner() {
2359 return this.hfileCleaner;
2360 }
2361
2362
2363
2364
2365
2366 public SnapshotManager getSnapshotManagerForTesting() {
2367 return this.snapshotManager;
2368 }
2369
2370 @Override
2371 public void createNamespace(NamespaceDescriptor descriptor) throws IOException {
2372 TableName.isLegalNamespaceName(Bytes.toBytes(descriptor.getName()));
2373 checkNamespaceManagerReady();
2374 if (cpHost != null) {
2375 if (cpHost.preCreateNamespace(descriptor)) {
2376 return;
2377 }
2378 }
2379 LOG.info(getClientIdAuditPrefix() + " creating " + descriptor);
2380 tableNamespaceManager.create(descriptor);
2381 if (cpHost != null) {
2382 cpHost.postCreateNamespace(descriptor);
2383 }
2384 }
2385
2386 @Override
2387 public void modifyNamespace(NamespaceDescriptor descriptor) throws IOException {
2388 TableName.isLegalNamespaceName(Bytes.toBytes(descriptor.getName()));
2389 checkNamespaceManagerReady();
2390 if (cpHost != null) {
2391 if (cpHost.preModifyNamespace(descriptor)) {
2392 return;
2393 }
2394 }
2395 LOG.info(getClientIdAuditPrefix() + " modify " + descriptor);
2396 tableNamespaceManager.update(descriptor);
2397 if (cpHost != null) {
2398 cpHost.postModifyNamespace(descriptor);
2399 }
2400 }
2401
2402 @Override
2403 public void deleteNamespace(String name) throws IOException {
2404 checkNamespaceManagerReady();
2405 if (cpHost != null) {
2406 if (cpHost.preDeleteNamespace(name)) {
2407 return;
2408 }
2409 }
2410 LOG.info(getClientIdAuditPrefix() + " delete " + name);
2411 tableNamespaceManager.remove(name);
2412 if (cpHost != null) {
2413 cpHost.postDeleteNamespace(name);
2414 }
2415 }
2416
2417
2418
2419
2420
2421
2422
2423
2424 private void ensureNamespaceExists(final String name)
2425 throws IOException, NamespaceNotFoundException {
2426 checkNamespaceManagerReady();
2427 NamespaceDescriptor nsd = tableNamespaceManager.get(name);
2428 if (nsd == null) {
2429 throw new NamespaceNotFoundException(name);
2430 }
2431 }
2432
2433 @Override
2434 public NamespaceDescriptor getNamespaceDescriptor(String name) throws IOException {
2435 checkNamespaceManagerReady();
2436
2437 if (cpHost != null) {
2438 cpHost.preGetNamespaceDescriptor(name);
2439 }
2440
2441 NamespaceDescriptor nsd = tableNamespaceManager.get(name);
2442 if (nsd == null) {
2443 throw new NamespaceNotFoundException(name);
2444 }
2445
2446 if (cpHost != null) {
2447 cpHost.postGetNamespaceDescriptor(nsd);
2448 }
2449
2450 return nsd;
2451 }
2452
2453 @Override
2454 public List<NamespaceDescriptor> listNamespaceDescriptors() throws IOException {
2455 checkNamespaceManagerReady();
2456
2457 final List<NamespaceDescriptor> descriptors = new ArrayList<NamespaceDescriptor>();
2458 boolean bypass = false;
2459 if (cpHost != null) {
2460 bypass = cpHost.preListNamespaceDescriptors(descriptors);
2461 }
2462
2463 if (!bypass) {
2464 descriptors.addAll(tableNamespaceManager.list());
2465
2466 if (cpHost != null) {
2467 cpHost.postListNamespaceDescriptors(descriptors);
2468 }
2469 }
2470 return descriptors;
2471 }
2472
2473 @Override
2474 public boolean abortProcedure(final long procId, final boolean mayInterruptIfRunning)
2475 throws IOException {
2476 if (cpHost != null) {
2477 cpHost.preAbortProcedure(this.procedureExecutor, procId);
2478 }
2479
2480 final boolean result = this.procedureExecutor.abort(procId, mayInterruptIfRunning);
2481
2482 if (cpHost != null) {
2483 cpHost.postAbortProcedure();
2484 }
2485
2486 return result;
2487 }
2488
2489 @Override
2490 public List<ProcedureInfo> listProcedures() throws IOException {
2491 if (cpHost != null) {
2492 cpHost.preListProcedures();
2493 }
2494
2495 final List<ProcedureInfo> procInfoList = this.procedureExecutor.listProcedures();
2496
2497 if (cpHost != null) {
2498 cpHost.postListProcedures(procInfoList);
2499 }
2500
2501 return procInfoList;
2502 }
2503
2504 @Override
2505 public List<HTableDescriptor> listTableDescriptorsByNamespace(String name) throws IOException {
2506 ensureNamespaceExists(name);
2507 return listTableDescriptors(name, null, null, true);
2508 }
2509
2510 @Override
2511 public List<TableName> listTableNamesByNamespace(String name) throws IOException {
2512 ensureNamespaceExists(name);
2513 return listTableNames(name, null, true);
2514 }
2515
2516
2517
2518
2519
2520
2521
2522
2523
2524
2525 public List<HTableDescriptor> listTableDescriptors(final String namespace, final String regex,
2526 final List<TableName> tableNameList, final boolean includeSysTables)
2527 throws IOException {
2528 final List<HTableDescriptor> descriptors = new ArrayList<HTableDescriptor>();
2529
2530 boolean bypass = false;
2531 if (cpHost != null) {
2532 bypass = cpHost.preGetTableDescriptors(tableNameList, descriptors);
2533
2534 bypass |= cpHost.preGetTableDescriptors(tableNameList, descriptors, regex);
2535 }
2536
2537 if (!bypass) {
2538 if (tableNameList == null || tableNameList.size() == 0) {
2539
2540 Collection<HTableDescriptor> htds;
2541 if (namespace != null && namespace.length() > 0) {
2542 htds = tableDescriptors.getByNamespace(namespace).values();
2543 } else {
2544 htds = tableDescriptors.getAll().values();
2545 }
2546
2547 for (HTableDescriptor desc: htds) {
2548 if (includeSysTables || !desc.getTableName().isSystemTable()) {
2549 descriptors.add(desc);
2550 }
2551 }
2552 } else {
2553 for (TableName s: tableNameList) {
2554 HTableDescriptor desc = tableDescriptors.get(s);
2555 if (desc != null) {
2556 descriptors.add(desc);
2557 }
2558 }
2559 }
2560
2561
2562 if (regex != null) {
2563 filterTablesByRegex(descriptors, Pattern.compile(regex));
2564 }
2565
2566 if (cpHost != null) {
2567 cpHost.postGetTableDescriptors(descriptors);
2568
2569 cpHost.postGetTableDescriptors(tableNameList, descriptors, regex);
2570 }
2571 }
2572 return descriptors;
2573 }
2574
2575
2576
2577
2578
2579
2580
2581
2582 public List<TableName> listTableNames(final String namespace, final String regex,
2583 final boolean includeSysTables) throws IOException {
2584 final List<HTableDescriptor> descriptors = new ArrayList<HTableDescriptor>();
2585
2586 boolean bypass = false;
2587 if (cpHost != null) {
2588 bypass = cpHost.preGetTableNames(descriptors, regex);
2589 }
2590
2591 if (!bypass) {
2592
2593 Collection<HTableDescriptor> htds;
2594 if (namespace != null && namespace.length() > 0) {
2595 htds = tableDescriptors.getByNamespace(namespace).values();
2596 } else {
2597 htds = tableDescriptors.getAll().values();
2598 }
2599
2600 for (HTableDescriptor htd: htds) {
2601 if (includeSysTables || !htd.getTableName().isSystemTable()) {
2602 descriptors.add(htd);
2603 }
2604 }
2605
2606
2607 if (regex != null) {
2608 filterTablesByRegex(descriptors, Pattern.compile(regex));
2609 }
2610
2611 if (cpHost != null) {
2612 cpHost.postGetTableNames(descriptors, regex);
2613 }
2614 }
2615
2616 List<TableName> result = new ArrayList<TableName>(descriptors.size());
2617 for (HTableDescriptor htd: descriptors) {
2618 result.add(htd.getTableName());
2619 }
2620 return result;
2621 }
2622
2623
2624
2625
2626
2627
2628
2629 private static void filterTablesByRegex(final Collection<HTableDescriptor> descriptors,
2630 final Pattern pattern) {
2631 final String defaultNS = NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR;
2632 Iterator<HTableDescriptor> itr = descriptors.iterator();
2633 while (itr.hasNext()) {
2634 HTableDescriptor htd = itr.next();
2635 String tableName = htd.getTableName().getNameAsString();
2636 boolean matched = pattern.matcher(tableName).matches();
2637 if (!matched && htd.getTableName().getNamespaceAsString().equals(defaultNS)) {
2638 matched = pattern.matcher(defaultNS + TableName.NAMESPACE_DELIM + tableName).matches();
2639 }
2640 if (!matched) {
2641 itr.remove();
2642 }
2643 }
2644 }
2645
2646 @Override
2647 public long getLastMajorCompactionTimestamp(TableName table) throws IOException {
2648 return getClusterStatus().getLastMajorCompactionTsForTable(table);
2649 }
2650
2651 @Override
2652 public long getLastMajorCompactionTimestampForRegion(byte[] regionName) throws IOException {
2653 return getClusterStatus().getLastMajorCompactionTsForRegion(regionName);
2654 }
2655
2656
2657
2658
2659
2660
2661
2662 public boolean isBalancerOn() {
2663 if (null == loadBalancerTracker) return false;
2664 return loadBalancerTracker.isBalancerOn();
2665 }
2666
2667
2668
2669
2670
2671
2672 public String getLoadBalancerClassName() {
2673 return conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, LoadBalancerFactory
2674 .getDefaultLoadBalancerClass().getName());
2675 }
2676 }