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