001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.master; 019 020import static org.apache.hadoop.hbase.HConstants.HBASE_MASTER_LOGCLEANER_PLUGINS; 021 022import com.google.protobuf.Descriptors; 023import com.google.protobuf.Service; 024import java.io.IOException; 025import java.io.InterruptedIOException; 026import java.lang.reflect.Constructor; 027import java.lang.reflect.InvocationTargetException; 028import java.net.InetAddress; 029import java.net.InetSocketAddress; 030import java.net.UnknownHostException; 031import java.util.ArrayList; 032import java.util.Arrays; 033import java.util.Collection; 034import java.util.Collections; 035import java.util.Comparator; 036import java.util.EnumSet; 037import java.util.HashMap; 038import java.util.Iterator; 039import java.util.List; 040import java.util.Map; 041import java.util.Map.Entry; 042import java.util.Objects; 043import java.util.Optional; 044import java.util.Set; 045import java.util.concurrent.ExecutionException; 046import java.util.concurrent.Future; 047import java.util.concurrent.TimeUnit; 048import java.util.concurrent.TimeoutException; 049import java.util.concurrent.atomic.AtomicInteger; 050import java.util.concurrent.atomic.AtomicReference; 051import java.util.function.Function; 052import java.util.regex.Pattern; 053import java.util.stream.Collectors; 054import javax.servlet.ServletException; 055import javax.servlet.http.HttpServlet; 056import javax.servlet.http.HttpServletRequest; 057import javax.servlet.http.HttpServletResponse; 058import org.apache.commons.lang3.StringUtils; 059import org.apache.hadoop.conf.Configuration; 060import org.apache.hadoop.fs.Path; 061import org.apache.hadoop.hbase.ClusterId; 062import org.apache.hadoop.hbase.ClusterMetrics; 063import org.apache.hadoop.hbase.ClusterMetrics.Option; 064import org.apache.hadoop.hbase.ClusterMetricsBuilder; 065import org.apache.hadoop.hbase.DoNotRetryIOException; 066import org.apache.hadoop.hbase.HBaseIOException; 067import org.apache.hadoop.hbase.HBaseInterfaceAudience; 068import org.apache.hadoop.hbase.HConstants; 069import org.apache.hadoop.hbase.InvalidFamilyOperationException; 070import org.apache.hadoop.hbase.MasterNotRunningException; 071import org.apache.hadoop.hbase.MetaTableAccessor; 072import org.apache.hadoop.hbase.NamespaceDescriptor; 073import org.apache.hadoop.hbase.PleaseHoldException; 074import org.apache.hadoop.hbase.ServerName; 075import org.apache.hadoop.hbase.TableDescriptors; 076import org.apache.hadoop.hbase.TableName; 077import org.apache.hadoop.hbase.TableNotDisabledException; 078import org.apache.hadoop.hbase.TableNotFoundException; 079import org.apache.hadoop.hbase.UnknownRegionException; 080import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; 081import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; 082import org.apache.hadoop.hbase.client.MasterSwitchType; 083import org.apache.hadoop.hbase.client.RegionInfo; 084import org.apache.hadoop.hbase.client.RegionInfoBuilder; 085import org.apache.hadoop.hbase.client.Result; 086import org.apache.hadoop.hbase.client.TableDescriptor; 087import org.apache.hadoop.hbase.client.TableDescriptorBuilder; 088import org.apache.hadoop.hbase.client.TableState; 089import org.apache.hadoop.hbase.client.VersionInfoUtil; 090import org.apache.hadoop.hbase.coprocessor.CoprocessorHost; 091import org.apache.hadoop.hbase.exceptions.DeserializationException; 092import org.apache.hadoop.hbase.exceptions.MergeRegionException; 093import org.apache.hadoop.hbase.executor.ExecutorType; 094import org.apache.hadoop.hbase.favored.FavoredNodesManager; 095import org.apache.hadoop.hbase.favored.FavoredNodesPromoter; 096import org.apache.hadoop.hbase.http.InfoServer; 097import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils; 098import org.apache.hadoop.hbase.ipc.RpcServer; 099import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException; 100import org.apache.hadoop.hbase.log.HBaseMarkers; 101import org.apache.hadoop.hbase.master.MasterRpcServices.BalanceSwitchMode; 102import org.apache.hadoop.hbase.master.assignment.AssignmentManager; 103import org.apache.hadoop.hbase.master.assignment.MergeTableRegionsProcedure; 104import org.apache.hadoop.hbase.master.assignment.RegionStates; 105import org.apache.hadoop.hbase.master.assignment.RegionStates.RegionStateNode; 106import org.apache.hadoop.hbase.master.balancer.BalancerChore; 107import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer; 108import org.apache.hadoop.hbase.master.balancer.ClusterStatusChore; 109import org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory; 110import org.apache.hadoop.hbase.master.cleaner.HFileCleaner; 111import org.apache.hadoop.hbase.master.cleaner.LogCleaner; 112import org.apache.hadoop.hbase.master.cleaner.ReplicationZKNodeCleaner; 113import org.apache.hadoop.hbase.master.cleaner.ReplicationZKNodeCleanerChore; 114import org.apache.hadoop.hbase.master.locking.LockManager; 115import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan; 116import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan.PlanType; 117import org.apache.hadoop.hbase.master.normalizer.RegionNormalizer; 118import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerChore; 119import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerFactory; 120import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure; 121import org.apache.hadoop.hbase.master.procedure.DeleteNamespaceProcedure; 122import org.apache.hadoop.hbase.master.procedure.DeleteTableProcedure; 123import org.apache.hadoop.hbase.master.procedure.DisableTableProcedure; 124import org.apache.hadoop.hbase.master.procedure.EnableTableProcedure; 125import org.apache.hadoop.hbase.master.procedure.InitMetaProcedure; 126import org.apache.hadoop.hbase.master.procedure.MasterProcedureConstants; 127import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv; 128import org.apache.hadoop.hbase.master.procedure.MasterProcedureScheduler; 129import org.apache.hadoop.hbase.master.procedure.MasterProcedureUtil; 130import org.apache.hadoop.hbase.master.procedure.ModifyTableProcedure; 131import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch; 132import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure; 133import org.apache.hadoop.hbase.master.procedure.TruncateTableProcedure; 134import org.apache.hadoop.hbase.master.replication.ReplicationManager; 135import org.apache.hadoop.hbase.master.snapshot.SnapshotManager; 136import org.apache.hadoop.hbase.mob.MobConstants; 137import org.apache.hadoop.hbase.monitoring.MemoryBoundedLogMessageBuffer; 138import org.apache.hadoop.hbase.monitoring.MonitoredTask; 139import org.apache.hadoop.hbase.monitoring.TaskMonitor; 140import org.apache.hadoop.hbase.procedure.MasterProcedureManagerHost; 141import org.apache.hadoop.hbase.procedure.flush.MasterFlushTableProcedureManager; 142import org.apache.hadoop.hbase.procedure2.LockedResource; 143import org.apache.hadoop.hbase.procedure2.Procedure; 144import org.apache.hadoop.hbase.procedure2.ProcedureEvent; 145import org.apache.hadoop.hbase.procedure2.ProcedureExecutor; 146import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore; 147import org.apache.hadoop.hbase.quotas.MasterQuotaManager; 148import org.apache.hadoop.hbase.quotas.MasterQuotasObserver; 149import org.apache.hadoop.hbase.quotas.QuotaObserverChore; 150import org.apache.hadoop.hbase.quotas.QuotaUtil; 151import org.apache.hadoop.hbase.quotas.SnapshotQuotaObserverChore; 152import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshotNotifier; 153import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshotNotifierFactory; 154import org.apache.hadoop.hbase.regionserver.DefaultStoreEngine; 155import org.apache.hadoop.hbase.regionserver.HRegionServer; 156import org.apache.hadoop.hbase.regionserver.HStore; 157import org.apache.hadoop.hbase.regionserver.RSRpcServices; 158import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost; 159import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy; 160import org.apache.hadoop.hbase.regionserver.compactions.ExploringCompactionPolicy; 161import org.apache.hadoop.hbase.regionserver.compactions.FIFOCompactionPolicy; 162import org.apache.hadoop.hbase.replication.ReplicationException; 163import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; 164import org.apache.hadoop.hbase.replication.ReplicationPeerDescription; 165import org.apache.hadoop.hbase.replication.ReplicationUtils; 166import org.apache.hadoop.hbase.replication.master.ReplicationHFileCleaner; 167import org.apache.hadoop.hbase.replication.master.ReplicationLogCleaner; 168import org.apache.hadoop.hbase.replication.master.ReplicationPeerConfigUpgrader; 169import org.apache.hadoop.hbase.security.AccessDeniedException; 170import org.apache.hadoop.hbase.security.UserProvider; 171import org.apache.hadoop.hbase.trace.TraceUtil; 172import org.apache.hadoop.hbase.util.Addressing; 173import org.apache.hadoop.hbase.util.Bytes; 174import org.apache.hadoop.hbase.util.CompressionTest; 175import org.apache.hadoop.hbase.util.EncryptionTest; 176import org.apache.hadoop.hbase.util.HBaseFsck; 177import org.apache.hadoop.hbase.util.HFileArchiveUtil; 178import org.apache.hadoop.hbase.util.HasThread; 179import org.apache.hadoop.hbase.util.IdLock; 180import org.apache.hadoop.hbase.util.ModifyRegionUtils; 181import org.apache.hadoop.hbase.util.Pair; 182import org.apache.hadoop.hbase.util.Threads; 183import org.apache.hadoop.hbase.util.VersionInfo; 184import org.apache.hadoop.hbase.zookeeper.LoadBalancerTracker; 185import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker; 186import org.apache.hadoop.hbase.zookeeper.MasterMaintenanceModeTracker; 187import org.apache.hadoop.hbase.zookeeper.RegionNormalizerTracker; 188import org.apache.hadoop.hbase.zookeeper.ZKClusterId; 189import org.apache.hadoop.hbase.zookeeper.ZKUtil; 190import org.apache.hadoop.hbase.zookeeper.ZKWatcher; 191import org.apache.hadoop.hbase.zookeeper.ZNodePaths; 192import org.apache.yetus.audience.InterfaceAudience; 193import org.apache.zookeeper.KeeperException; 194import org.eclipse.jetty.server.Server; 195import org.eclipse.jetty.server.ServerConnector; 196import org.eclipse.jetty.servlet.ServletHolder; 197import org.eclipse.jetty.webapp.WebAppContext; 198import org.slf4j.Logger; 199import org.slf4j.LoggerFactory; 200 201import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 202import org.apache.hbase.thirdparty.com.google.common.collect.Lists; 203import org.apache.hbase.thirdparty.com.google.common.collect.Maps; 204 205import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 206import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState; 207import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.Quotas; 208import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.SpaceViolationPolicy; 209import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos.SnapshotDescription; 210import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos; 211 212/** 213 * HMaster is the "master server" for HBase. An HBase cluster has one active 214 * master. If many masters are started, all compete. Whichever wins goes on to 215 * run the cluster. All others park themselves in their constructor until 216 * master or cluster shutdown or until the active master loses its lease in 217 * zookeeper. Thereafter, all running master jostle to take over master role. 218 * 219 * <p>The Master can be asked shutdown the cluster. See {@link #shutdown()}. In 220 * this case it will tell all regionservers to go down and then wait on them 221 * all reporting in that they are down. This master will then shut itself down. 222 * 223 * <p>You can also shutdown just this master. Call {@link #stopMaster()}. 224 * 225 * @see org.apache.zookeeper.Watcher 226 */ 227@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS) 228@SuppressWarnings("deprecation") 229public class HMaster extends HRegionServer implements MasterServices { 230 private static Logger LOG = LoggerFactory.getLogger(HMaster.class); 231 232 /** 233 * Protection against zombie master. Started once Master accepts active responsibility and 234 * starts taking over responsibilities. Allows a finite time window before giving up ownership. 235 */ 236 private static class InitializationMonitor extends HasThread { 237 /** The amount of time in milliseconds to sleep before checking initialization status. */ 238 public static final String TIMEOUT_KEY = "hbase.master.initializationmonitor.timeout"; 239 public static final long TIMEOUT_DEFAULT = TimeUnit.MILLISECONDS.convert(15, TimeUnit.MINUTES); 240 241 /** 242 * When timeout expired and initialization has not complete, call {@link System#exit(int)} when 243 * true, do nothing otherwise. 244 */ 245 public static final String HALT_KEY = "hbase.master.initializationmonitor.haltontimeout"; 246 public static final boolean HALT_DEFAULT = false; 247 248 private final HMaster master; 249 private final long timeout; 250 private final boolean haltOnTimeout; 251 252 /** Creates a Thread that monitors the {@link #isInitialized()} state. */ 253 InitializationMonitor(HMaster master) { 254 super("MasterInitializationMonitor"); 255 this.master = master; 256 this.timeout = master.getConfiguration().getLong(TIMEOUT_KEY, TIMEOUT_DEFAULT); 257 this.haltOnTimeout = master.getConfiguration().getBoolean(HALT_KEY, HALT_DEFAULT); 258 this.setDaemon(true); 259 } 260 261 @Override 262 public void run() { 263 try { 264 while (!master.isStopped() && master.isActiveMaster()) { 265 Thread.sleep(timeout); 266 if (master.isInitialized()) { 267 LOG.debug("Initialization completed within allotted tolerance. Monitor exiting."); 268 } else { 269 LOG.error("Master failed to complete initialization after " + timeout + "ms. Please" 270 + " consider submitting a bug report including a thread dump of this process."); 271 if (haltOnTimeout) { 272 LOG.error("Zombie Master exiting. Thread dump to stdout"); 273 Threads.printThreadInfo(System.out, "Zombie HMaster"); 274 System.exit(-1); 275 } 276 } 277 } 278 } catch (InterruptedException ie) { 279 LOG.trace("InitMonitor thread interrupted. Existing."); 280 } 281 } 282 } 283 284 // MASTER is name of the webapp and the attribute name used stuffing this 285 //instance into web context. 286 public static final String MASTER = "master"; 287 288 // Manager and zk listener for master election 289 private final ActiveMasterManager activeMasterManager; 290 // Region server tracker 291 private RegionServerTracker regionServerTracker; 292 // Draining region server tracker 293 private DrainingServerTracker drainingServerTracker; 294 // Tracker for load balancer state 295 LoadBalancerTracker loadBalancerTracker; 296 297 // Tracker for split and merge state 298 private SplitOrMergeTracker splitOrMergeTracker; 299 300 // Tracker for region normalizer state 301 private RegionNormalizerTracker regionNormalizerTracker; 302 303 //Tracker for master maintenance mode setting 304 private MasterMaintenanceModeTracker maintenanceModeTracker; 305 306 private ClusterSchemaService clusterSchemaService; 307 308 public static final String HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS = 309 "hbase.master.wait.on.service.seconds"; 310 public static final int DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS = 5 * 60; 311 312 // Metrics for the HMaster 313 final MetricsMaster metricsMaster; 314 // file system manager for the master FS operations 315 private MasterFileSystem fileSystemManager; 316 private MasterWalManager walManager; 317 318 // server manager to deal with region server info 319 private volatile ServerManager serverManager; 320 321 // manager of assignment nodes in zookeeper 322 private AssignmentManager assignmentManager; 323 324 // manager of replication 325 private ReplicationManager replicationManager; 326 327 // buffer for "fatal error" notices from region servers 328 // in the cluster. This is only used for assisting 329 // operations/debugging. 330 MemoryBoundedLogMessageBuffer rsFatals; 331 332 // flag set after we become the active master (used for testing) 333 private volatile boolean activeMaster = false; 334 335 // flag set after we complete initialization once active, 336 // it is not private since it's used in unit tests 337 private final ProcedureEvent initialized = new ProcedureEvent("master initialized"); 338 339 // flag set after master services are started, 340 // initialization may have not completed yet. 341 volatile boolean serviceStarted = false; 342 343 // Maximum time we should run balancer for 344 private final int maxBlancingTime; 345 // Maximum percent of regions in transition when balancing 346 private final double maxRitPercent; 347 348 private final LockManager lockManager = new LockManager(this); 349 350 private LoadBalancer balancer; 351 private RegionNormalizer normalizer; 352 private BalancerChore balancerChore; 353 private RegionNormalizerChore normalizerChore; 354 private ClusterStatusChore clusterStatusChore; 355 private ClusterStatusPublisher clusterStatusPublisherChore = null; 356 357 CatalogJanitor catalogJanitorChore; 358 359 private ReplicationZKNodeCleanerChore replicationZKNodeCleanerChore; 360 private LogCleaner logCleaner; 361 private HFileCleaner hfileCleaner; 362 private ExpiredMobFileCleanerChore expiredMobFileCleanerChore; 363 private MobCompactionChore mobCompactChore; 364 private MasterMobCompactionThread mobCompactThread; 365 // used to synchronize the mobCompactionStates 366 private final IdLock mobCompactionLock = new IdLock(); 367 // save the information of mob compactions in tables. 368 // the key is table name, the value is the number of compactions in that table. 369 private Map<TableName, AtomicInteger> mobCompactionStates = Maps.newConcurrentMap(); 370 371 MasterCoprocessorHost cpHost; 372 373 private final boolean preLoadTableDescriptors; 374 375 // Time stamps for when a hmaster became active 376 private long masterActiveTime; 377 378 // Time stamp for when HMaster finishes becoming Active Master 379 private long masterFinishedInitializationTime; 380 381 //should we check the compression codec type at master side, default true, HBASE-6370 382 private final boolean masterCheckCompression; 383 384 //should we check encryption settings at master side, default true 385 private final boolean masterCheckEncryption; 386 387 Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap(); 388 389 // monitor for snapshot of hbase tables 390 SnapshotManager snapshotManager; 391 // monitor for distributed procedures 392 private MasterProcedureManagerHost mpmHost; 393 394 // it is assigned after 'initialized' guard set to true, so should be volatile 395 private volatile MasterQuotaManager quotaManager; 396 private SpaceQuotaSnapshotNotifier spaceQuotaSnapshotNotifier; 397 private QuotaObserverChore quotaObserverChore; 398 private SnapshotQuotaObserverChore snapshotQuotaChore; 399 400 private ProcedureExecutor<MasterProcedureEnv> procedureExecutor; 401 private WALProcedureStore procedureStore; 402 403 // handle table states 404 private TableStateManager tableStateManager; 405 406 private long splitPlanCount; 407 private long mergePlanCount; 408 409 /* Handle favored nodes information */ 410 private FavoredNodesManager favoredNodesManager; 411 412 /** jetty server for master to redirect requests to regionserver infoServer */ 413 private Server masterJettyServer; 414 415 public static class RedirectServlet extends HttpServlet { 416 private static final long serialVersionUID = 2894774810058302473L; 417 private final int regionServerInfoPort; 418 private final String regionServerHostname; 419 420 /** 421 * @param infoServer that we're trying to send all requests to 422 * @param hostname may be null. if given, will be used for redirects instead of host from client. 423 */ 424 public RedirectServlet(InfoServer infoServer, String hostname) { 425 regionServerInfoPort = infoServer.getPort(); 426 regionServerHostname = hostname; 427 } 428 429 @Override 430 public void doGet(HttpServletRequest request, 431 HttpServletResponse response) throws ServletException, IOException { 432 String redirectHost = regionServerHostname; 433 if(redirectHost == null) { 434 redirectHost = request.getServerName(); 435 if(!Addressing.isLocalAddress(InetAddress.getByName(redirectHost))) { 436 LOG.warn("Couldn't resolve '" + redirectHost + "' as an address local to this node and '" + 437 MASTER_HOSTNAME_KEY + "' is not set; client will get a HTTP 400 response. If " + 438 "your HBase deployment relies on client accessible names that the region server process " + 439 "can't resolve locally, then you should set the previously mentioned configuration variable " + 440 "to an appropriate hostname."); 441 // no sending client provided input back to the client, so the goal host is just in the logs. 442 response.sendError(400, "Request was to a host that I can't resolve for any of the network interfaces on " + 443 "this node. If this is due to an intermediary such as an HTTP load balancer or other proxy, your HBase " + 444 "administrator can set '" + MASTER_HOSTNAME_KEY + "' to point to the correct hostname."); 445 return; 446 } 447 } 448 // TODO this scheme should come from looking at the scheme registered in the infoserver's http server for the 449 // host and port we're using, but it's buried way too deep to do that ATM. 450 String redirectUrl = request.getScheme() + "://" 451 + redirectHost + ":" + regionServerInfoPort 452 + request.getRequestURI(); 453 response.sendRedirect(redirectUrl); 454 } 455 } 456 457 /** 458 * Initializes the HMaster. The steps are as follows: 459 * <p> 460 * <ol> 461 * <li>Initialize the local HRegionServer 462 * <li>Start the ActiveMasterManager. 463 * </ol> 464 * <p> 465 * Remaining steps of initialization occur in 466 * #finishActiveMasterInitialization(MonitoredTask) after 467 * the master becomes the active one. 468 */ 469 public HMaster(final Configuration conf) 470 throws IOException, KeeperException { 471 super(conf); 472 TraceUtil.initTracer(conf); 473 try { 474 this.rsFatals = new MemoryBoundedLogMessageBuffer( 475 conf.getLong("hbase.master.buffer.for.rs.fatals", 1 * 1024 * 1024)); 476 LOG.info("hbase.rootdir=" + getRootDir() + 477 ", hbase.cluster.distributed=" + this.conf.getBoolean(HConstants.CLUSTER_DISTRIBUTED, false)); 478 479 // Disable usage of meta replicas in the master 480 this.conf.setBoolean(HConstants.USE_META_REPLICAS, false); 481 482 decorateMasterConfiguration(this.conf); 483 484 // Hack! Maps DFSClient => Master for logs. HDFS made this 485 // config param for task trackers, but we can piggyback off of it. 486 if (this.conf.get("mapreduce.task.attempt.id") == null) { 487 this.conf.set("mapreduce.task.attempt.id", "hb_m_" + this.serverName.toString()); 488 } 489 490 // should we check the compression codec type at master side, default true, HBASE-6370 491 this.masterCheckCompression = conf.getBoolean("hbase.master.check.compression", true); 492 493 // should we check encryption settings at master side, default true 494 this.masterCheckEncryption = conf.getBoolean("hbase.master.check.encryption", true); 495 496 this.metricsMaster = new MetricsMaster(new MetricsMasterWrapperImpl(this)); 497 498 // preload table descriptor at startup 499 this.preLoadTableDescriptors = conf.getBoolean("hbase.master.preload.tabledescriptors", true); 500 501 this.maxBlancingTime = getMaxBalancingTime(); 502 this.maxRitPercent = conf.getDouble(HConstants.HBASE_MASTER_BALANCER_MAX_RIT_PERCENT, 503 HConstants.DEFAULT_HBASE_MASTER_BALANCER_MAX_RIT_PERCENT); 504 505 // Do we publish the status? 506 507 boolean shouldPublish = conf.getBoolean(HConstants.STATUS_PUBLISHED, 508 HConstants.STATUS_PUBLISHED_DEFAULT); 509 Class<? extends ClusterStatusPublisher.Publisher> publisherClass = 510 conf.getClass(ClusterStatusPublisher.STATUS_PUBLISHER_CLASS, 511 ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS, 512 ClusterStatusPublisher.Publisher.class); 513 514 if (shouldPublish) { 515 if (publisherClass == null) { 516 LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " + 517 ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS + 518 " is not set - not publishing status"); 519 } else { 520 clusterStatusPublisherChore = new ClusterStatusPublisher(this, conf, publisherClass); 521 getChoreService().scheduleChore(clusterStatusPublisherChore); 522 } 523 } 524 525 // Some unit tests don't need a cluster, so no zookeeper at all 526 if (!conf.getBoolean("hbase.testing.nocluster", false)) { 527 this.activeMasterManager = new ActiveMasterManager(zooKeeper, this.serverName, this); 528 } else { 529 this.activeMasterManager = null; 530 } 531 } catch (Throwable t) { 532 // Make sure we log the exception. HMaster is often started via reflection and the 533 // cause of failed startup is lost. 534 LOG.error("Failed construction of Master", t); 535 throw t; 536 } 537 } 538 539 @Override 540 protected String getUseThisHostnameInstead(Configuration conf) { 541 return conf.get(MASTER_HOSTNAME_KEY); 542 } 543 544 // Main run loop. Calls through to the regionserver run loop AFTER becoming active Master; will 545 // block in here until then. 546 @Override 547 public void run() { 548 try { 549 if (!conf.getBoolean("hbase.testing.nocluster", false)) { 550 Threads.setDaemonThreadRunning(new Thread(() -> { 551 try { 552 int infoPort = putUpJettyServer(); 553 startActiveMasterManager(infoPort); 554 } catch (Throwable t) { 555 // Make sure we log the exception. 556 String error = "Failed to become Active Master"; 557 LOG.error(error, t); 558 // Abort should have been called already. 559 if (!isAborted()) { 560 abort(error, t); 561 } 562 } 563 })); 564 } 565 // Fall in here even if we have been aborted. Need to run the shutdown services and 566 // the super run call will do this for us. 567 super.run(); 568 } finally { 569 if (this.clusterSchemaService != null) { 570 // If on way out, then we are no longer active master. 571 this.clusterSchemaService.stopAsync(); 572 try { 573 this.clusterSchemaService.awaitTerminated( 574 getConfiguration().getInt(HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS, 575 DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS), TimeUnit.SECONDS); 576 } catch (TimeoutException te) { 577 LOG.warn("Failed shutdown of clusterSchemaService", te); 578 } 579 } 580 this.activeMaster = false; 581 } 582 } 583 584 // return the actual infoPort, -1 means disable info server. 585 private int putUpJettyServer() throws IOException { 586 if (!conf.getBoolean("hbase.master.infoserver.redirect", true)) { 587 return -1; 588 } 589 final int infoPort = conf.getInt("hbase.master.info.port.orig", 590 HConstants.DEFAULT_MASTER_INFOPORT); 591 // -1 is for disabling info server, so no redirecting 592 if (infoPort < 0 || infoServer == null) { 593 return -1; 594 } 595 if(infoPort == infoServer.getPort()) { 596 return infoPort; 597 } 598 final String addr = conf.get("hbase.master.info.bindAddress", "0.0.0.0"); 599 if (!Addressing.isLocalAddress(InetAddress.getByName(addr))) { 600 String msg = 601 "Failed to start redirecting jetty server. Address " + addr 602 + " does not belong to this host. Correct configuration parameter: " 603 + "hbase.master.info.bindAddress"; 604 LOG.error(msg); 605 throw new IOException(msg); 606 } 607 608 // TODO I'm pretty sure we could just add another binding to the InfoServer run by 609 // the RegionServer and have it run the RedirectServlet instead of standing up 610 // a second entire stack here. 611 masterJettyServer = new Server(); 612 final ServerConnector connector = new ServerConnector(masterJettyServer); 613 connector.setHost(addr); 614 connector.setPort(infoPort); 615 masterJettyServer.addConnector(connector); 616 masterJettyServer.setStopAtShutdown(true); 617 618 final String redirectHostname = 619 StringUtils.isBlank(useThisHostnameInstead) ? null : useThisHostnameInstead; 620 621 final RedirectServlet redirect = new RedirectServlet(infoServer, redirectHostname); 622 final WebAppContext context = new WebAppContext(null, "/", null, null, null, null, WebAppContext.NO_SESSIONS); 623 context.addServlet(new ServletHolder(redirect), "/*"); 624 context.setServer(masterJettyServer); 625 626 try { 627 masterJettyServer.start(); 628 } catch (Exception e) { 629 throw new IOException("Failed to start redirecting jetty server", e); 630 } 631 return connector.getLocalPort(); 632 } 633 634 @Override 635 protected Function<TableDescriptorBuilder, TableDescriptorBuilder> getMetaTableObserver() { 636 return builder -> builder.setRegionReplication(conf.getInt(HConstants.META_REPLICAS_NUM, HConstants.DEFAULT_META_REPLICA_NUM)); 637 } 638 /** 639 * For compatibility, if failed with regionserver credentials, try the master one 640 */ 641 @Override 642 protected void login(UserProvider user, String host) throws IOException { 643 try { 644 super.login(user, host); 645 } catch (IOException ie) { 646 user.login("hbase.master.keytab.file", 647 "hbase.master.kerberos.principal", host); 648 } 649 } 650 651 /** 652 * If configured to put regions on active master, 653 * wait till a backup master becomes active. 654 * Otherwise, loop till the server is stopped or aborted. 655 */ 656 @Override 657 protected void waitForMasterActive(){ 658 boolean tablesOnMaster = LoadBalancer.isTablesOnMaster(conf); 659 while (!(tablesOnMaster && activeMaster) && !isStopped() && !isAborted()) { 660 sleeper.sleep(); 661 } 662 } 663 664 @VisibleForTesting 665 public MasterRpcServices getMasterRpcServices() { 666 return (MasterRpcServices)rpcServices; 667 } 668 669 public boolean balanceSwitch(final boolean b) throws IOException { 670 return getMasterRpcServices().switchBalancer(b, BalanceSwitchMode.ASYNC); 671 } 672 673 @Override 674 protected String getProcessName() { 675 return MASTER; 676 } 677 678 @Override 679 protected boolean canCreateBaseZNode() { 680 return true; 681 } 682 683 @Override 684 protected boolean canUpdateTableDescriptor() { 685 return true; 686 } 687 688 @Override 689 protected RSRpcServices createRpcServices() throws IOException { 690 return new MasterRpcServices(this); 691 } 692 693 @Override 694 protected void configureInfoServer() { 695 infoServer.addServlet("master-status", "/master-status", MasterStatusServlet.class); 696 infoServer.setAttribute(MASTER, this); 697 if (LoadBalancer.isTablesOnMaster(conf)) { 698 super.configureInfoServer(); 699 } 700 } 701 702 @Override 703 protected Class<? extends HttpServlet> getDumpServlet() { 704 return MasterDumpServlet.class; 705 } 706 707 @Override 708 public MetricsMaster getMasterMetrics() { 709 return metricsMaster; 710 } 711 712 /** 713 * Initialize all ZK based system trackers. But do not include {@link RegionServerTracker}, it 714 * should have already been initialized along with {@link ServerManager}. 715 * Will be overridden in tests. 716 */ 717 @VisibleForTesting 718 protected void initializeZKBasedSystemTrackers() 719 throws IOException, InterruptedException, KeeperException { 720 this.balancer = LoadBalancerFactory.getLoadBalancer(conf); 721 this.normalizer = RegionNormalizerFactory.getRegionNormalizer(conf); 722 this.normalizer.setMasterServices(this); 723 this.normalizer.setMasterRpcServices((MasterRpcServices)rpcServices); 724 this.loadBalancerTracker = new LoadBalancerTracker(zooKeeper, this); 725 this.loadBalancerTracker.start(); 726 727 this.regionNormalizerTracker = new RegionNormalizerTracker(zooKeeper, this); 728 this.regionNormalizerTracker.start(); 729 730 this.splitOrMergeTracker = new SplitOrMergeTracker(zooKeeper, conf, this); 731 this.splitOrMergeTracker.start(); 732 733 this.replicationManager = new ReplicationManager(conf, zooKeeper, this); 734 735 this.drainingServerTracker = new DrainingServerTracker(zooKeeper, this, this.serverManager); 736 this.drainingServerTracker.start(); 737 738 this.maintenanceModeTracker = new MasterMaintenanceModeTracker(zooKeeper); 739 this.maintenanceModeTracker.start(); 740 741 // Set the cluster as up. If new RSs, they'll be waiting on this before 742 // going ahead with their startup. 743 boolean wasUp = this.clusterStatusTracker.isClusterUp(); 744 if (!wasUp) this.clusterStatusTracker.setClusterUp(); 745 746 LOG.info("Active/primary master=" + this.serverName + 747 ", sessionid=0x" + 748 Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()) + 749 ", setting cluster-up flag (Was=" + wasUp + ")"); 750 751 // create/initialize the snapshot manager and other procedure managers 752 this.snapshotManager = new SnapshotManager(); 753 this.mpmHost = new MasterProcedureManagerHost(); 754 this.mpmHost.register(this.snapshotManager); 755 this.mpmHost.register(new MasterFlushTableProcedureManager()); 756 this.mpmHost.loadProcedures(conf); 757 this.mpmHost.initialize(this, this.metricsMaster); 758 } 759 760 /** 761 * Finish initialization of HMaster after becoming the primary master. 762 * 763 * The startup order is a bit complicated but very important, do not change it unless you know 764 * what you are doing. 765 * <li>Initialize file system based components - file system manager, wal manager, table 766 * descriptors, etc</li> 767 * <li>Publish cluster id</li> 768 * <li>Here comes the most complicated part - initialize server manager, assignment manager and 769 * region server tracker 770 * <ol type='i'> 771 * <li>Create server manager</li> 772 * <li>Create procedure executor, load the procedures, but do not start workers. We will start it 773 * later after we finish scheduling SCPs to avoid scheduling duplicated SCPs for the same 774 * server</li> 775 * <li>Create assignment manager and start it, load the meta region state, but do not load data 776 * from meta region</li> 777 * <li>Start region server tracker, construct the online servers set and find out dead servers and 778 * schedule SCP for them. The online servers will be constructed by scanning zk, and we will also 779 * scan the wal directory to find out possible live region servers, and the differences between 780 * these two sets are the dead servers</li> 781 * </ol> 782 * </li> 783 * <li>If this is a new deploy, schedule a InitMetaProcedure to initialize meta</li> 784 * <li>Start necessary service threads - balancer, catalog janior, executor services, and also the 785 * procedure executor, etc. Notice that the balancer must be created first as assignment manager 786 * may use it when assigning regions.</li> 787 * <li>Wait for meta to be initialized if necesssary, start table state manager.</li> 788 * <li>Wait for enough region servers to check-in</li> 789 * <li>Let assignment manager load data from meta and construct region states</li> 790 * <li>Start all other things such as chore services, etc</li> 791 * <p/> 792 * Notice that now we will not schedule a special procedure to make meta online(unless the first 793 * time where meta has not been created yet), we will rely on SCP to bring meta online. 794 */ 795 private void finishActiveMasterInitialization(MonitoredTask status) 796 throws IOException, InterruptedException, KeeperException { 797 798 Thread zombieDetector = new Thread(new InitializationMonitor(this), 799 "ActiveMasterInitializationMonitor-" + System.currentTimeMillis()); 800 zombieDetector.setDaemon(true); 801 zombieDetector.start(); 802 803 /* 804 * We are active master now... go initialize components we need to run. 805 */ 806 status.setStatus("Initializing Master file system"); 807 808 this.masterActiveTime = System.currentTimeMillis(); 809 // TODO: Do this using Dependency Injection, using PicoContainer, Guice or Spring. 810 // Initialize the chunkCreator 811 initializeMemStoreChunkCreator(); 812 this.fileSystemManager = new MasterFileSystem(conf); 813 this.walManager = new MasterWalManager(this); 814 815 // enable table descriptors cache 816 this.tableDescriptors.setCacheOn(); 817 818 // warm-up HTDs cache on master initialization 819 if (preLoadTableDescriptors) { 820 status.setStatus("Pre-loading table descriptors"); 821 this.tableDescriptors.getAll(); 822 } 823 824 // Publish cluster ID; set it in Master too. The superclass RegionServer does this later but 825 // only after it has checked in with the Master. At least a few tests ask Master for clusterId 826 // before it has called its run method and before RegionServer has done the reportForDuty. 827 ClusterId clusterId = fileSystemManager.getClusterId(); 828 status.setStatus("Publishing Cluster ID " + clusterId + " in ZooKeeper"); 829 ZKClusterId.setClusterId(this.zooKeeper, fileSystemManager.getClusterId()); 830 this.clusterId = clusterId.toString(); 831 832 // Precaution. Put in place the old hbck1 lock file to fence out old hbase1s running their 833 // hbck1s against an hbase2 cluster; it could do damage. To skip this behavior, set 834 // hbase.write.hbck1.lock.file to false. 835 if (this.conf.getBoolean("hbase.write.hbck1.lock.file", true)) { 836 HBaseFsck.checkAndMarkRunningHbck(this.conf, 837 HBaseFsck.createLockRetryCounterFactory(this.conf).create()); 838 } 839 840 status.setStatus("Initialze ServerManager and schedule SCP for crash servers"); 841 this.serverManager = createServerManager(this); 842 createProcedureExecutor(); 843 // Create Assignment Manager 844 this.assignmentManager = new AssignmentManager(this); 845 this.assignmentManager.start(); 846 this.regionServerTracker = new RegionServerTracker(zooKeeper, this, this.serverManager); 847 this.regionServerTracker.start( 848 procedureExecutor.getProcedures().stream().filter(p -> p instanceof ServerCrashProcedure) 849 .map(p -> ((ServerCrashProcedure) p).getServerName()).collect(Collectors.toSet()), 850 walManager.getLiveServersFromWALDir()); 851 // This manager will be started AFTER hbase:meta is confirmed on line. 852 // hbase.mirror.table.state.to.zookeeper is so hbase1 clients can connect. They read table 853 // state from zookeeper while hbase2 reads it from hbase:meta. Disable if no hbase1 clients. 854 this.tableStateManager = 855 this.conf.getBoolean(MirroringTableStateManager.MIRROR_TABLE_STATE_TO_ZK_KEY, true)? 856 new MirroringTableStateManager(this): 857 new TableStateManager(this); 858 859 status.setStatus("Initializing ZK system trackers"); 860 initializeZKBasedSystemTrackers(); 861 // Set ourselves as active Master now our claim has succeeded up in zk. 862 this.activeMaster = true; 863 864 // This is for backwards compatibility 865 // See HBASE-11393 866 status.setStatus("Update TableCFs node in ZNode"); 867 ReplicationPeerConfigUpgrader tableCFsUpdater = new ReplicationPeerConfigUpgrader(zooKeeper, 868 conf, this.clusterConnection); 869 tableCFsUpdater.copyTableCFs(); 870 871 // Add the Observer to delete quotas on table deletion before starting all CPs by 872 // default with quota support, avoiding if user specifically asks to not load this Observer. 873 if (QuotaUtil.isQuotaEnabled(conf)) { 874 updateConfigurationForQuotasObserver(conf); 875 } 876 // initialize master side coprocessors before we start handling requests 877 status.setStatus("Initializing master coprocessors"); 878 this.cpHost = new MasterCoprocessorHost(this, this.conf); 879 880 status.setStatus("Initializing meta table if this is a new deploy"); 881 InitMetaProcedure initMetaProc = null; 882 if (assignmentManager.getRegionStates().getRegionState(RegionInfoBuilder.FIRST_META_REGIONINFO) 883 .isOffline()) { 884 Optional<Procedure<MasterProcedureEnv>> optProc = procedureExecutor.getProcedures().stream() 885 .filter(p -> p instanceof InitMetaProcedure).findAny(); 886 if (optProc.isPresent()) { 887 initMetaProc = (InitMetaProcedure) optProc.get(); 888 } else { 889 // schedule an init meta procedure if meta has not been deployed yet 890 initMetaProc = new InitMetaProcedure(); 891 procedureExecutor.submitProcedure(initMetaProc); 892 } 893 } 894 if (this.balancer instanceof FavoredNodesPromoter) { 895 favoredNodesManager = new FavoredNodesManager(this); 896 } 897 898 // initialize load balancer 899 this.balancer.setMasterServices(this); 900 this.balancer.setClusterMetrics(getClusterMetricsWithoutCoprocessor()); 901 this.balancer.initialize(); 902 903 // start up all service threads. 904 status.setStatus("Initializing master service threads"); 905 startServiceThreads(); 906 // wait meta to be initialized after we start procedure executor 907 if (initMetaProc != null) { 908 initMetaProc.await(); 909 } 910 tableStateManager.start(); 911 // Wake up this server to check in 912 sleeper.skipSleepCycle(); 913 914 // Wait for region servers to report in. 915 // With this as part of master initialization, it precludes our being able to start a single 916 // server that is both Master and RegionServer. Needs more thought. TODO. 917 String statusStr = "Wait for region servers to report in"; 918 status.setStatus(statusStr); 919 LOG.info(Objects.toString(status)); 920 waitForRegionServers(status); 921 922 // Check if master is shutting down because issue initializing regionservers or balancer. 923 if (isStopped()) { 924 return; 925 } 926 927 //Initialize after meta as it scans meta 928 if (favoredNodesManager != null) { 929 SnapshotOfRegionAssignmentFromMeta snapshotOfRegionAssignment = 930 new SnapshotOfRegionAssignmentFromMeta(getConnection()); 931 snapshotOfRegionAssignment.initialize(); 932 favoredNodesManager.initialize(snapshotOfRegionAssignment); 933 } 934 935 // Fix up assignment manager status 936 status.setStatus("Starting assignment manager"); 937 this.assignmentManager.joinCluster(); 938 939 // set cluster status again after user regions are assigned 940 this.balancer.setClusterMetrics(getClusterMetricsWithoutCoprocessor()); 941 942 // Start balancer and meta catalog janitor after meta and regions have been assigned. 943 status.setStatus("Starting balancer and catalog janitor"); 944 this.clusterStatusChore = new ClusterStatusChore(this, balancer); 945 getChoreService().scheduleChore(clusterStatusChore); 946 this.balancerChore = new BalancerChore(this); 947 getChoreService().scheduleChore(balancerChore); 948 this.normalizerChore = new RegionNormalizerChore(this); 949 getChoreService().scheduleChore(normalizerChore); 950 this.catalogJanitorChore = new CatalogJanitor(this); 951 getChoreService().scheduleChore(catalogJanitorChore); 952 953 status.setStatus("Starting cluster schema service"); 954 initClusterSchemaService(); 955 956 if (this.cpHost != null) { 957 try { 958 this.cpHost.preMasterInitialization(); 959 } catch (IOException e) { 960 LOG.error("Coprocessor preMasterInitialization() hook failed", e); 961 } 962 } 963 964 status.markComplete("Initialization successful"); 965 LOG.info(String.format("Master has completed initialization %.3fsec", 966 (System.currentTimeMillis() - masterActiveTime) / 1000.0f)); 967 this.masterFinishedInitializationTime = System.currentTimeMillis(); 968 configurationManager.registerObserver(this.balancer); 969 configurationManager.registerObserver(this.hfileCleaner); 970 configurationManager.registerObserver(this.logCleaner); 971 // Set master as 'initialized'. 972 setInitialized(true); 973 assignmentManager.checkIfShouldMoveSystemRegionAsync(); 974 status.setStatus("Assign meta replicas"); 975 MasterMetaBootstrap metaBootstrap = createMetaBootstrap(); 976 metaBootstrap.assignMetaReplicas(); 977 status.setStatus("Starting quota manager"); 978 initQuotaManager(); 979 if (QuotaUtil.isQuotaEnabled(conf)) { 980 // Create the quota snapshot notifier 981 spaceQuotaSnapshotNotifier = createQuotaSnapshotNotifier(); 982 spaceQuotaSnapshotNotifier.initialize(getClusterConnection()); 983 this.quotaObserverChore = new QuotaObserverChore(this, getMasterMetrics()); 984 // Start the chore to read the region FS space reports and act on them 985 getChoreService().scheduleChore(quotaObserverChore); 986 987 this.snapshotQuotaChore = new SnapshotQuotaObserverChore(this, getMasterMetrics()); 988 // Start the chore to read snapshots and add their usage to table/NS quotas 989 getChoreService().scheduleChore(snapshotQuotaChore); 990 } 991 992 // clear the dead servers with same host name and port of online server because we are not 993 // removing dead server with same hostname and port of rs which is trying to check in before 994 // master initialization. See HBASE-5916. 995 this.serverManager.clearDeadServersWithSameHostNameAndPortOfOnlineServer(); 996 997 // Check and set the znode ACLs if needed in case we are overtaking a non-secure configuration 998 status.setStatus("Checking ZNode ACLs"); 999 zooKeeper.checkAndSetZNodeAcls(); 1000 1001 status.setStatus("Initializing MOB Cleaner"); 1002 initMobCleaner(); 1003 1004 status.setStatus("Calling postStartMaster coprocessors"); 1005 if (this.cpHost != null) { 1006 // don't let cp initialization errors kill the master 1007 try { 1008 this.cpHost.postStartMaster(); 1009 } catch (IOException ioe) { 1010 LOG.error("Coprocessor postStartMaster() hook failed", ioe); 1011 } 1012 } 1013 1014 zombieDetector.interrupt(); 1015 1016 /* 1017 * After master has started up, lets do balancer post startup initialization. Since this runs 1018 * in activeMasterManager thread, it should be fine. 1019 */ 1020 long start = System.currentTimeMillis(); 1021 this.balancer.postMasterStartupInitialize(); 1022 if (LOG.isDebugEnabled()) { 1023 LOG.debug("Balancer post startup initialization complete, took " + ( 1024 (System.currentTimeMillis() - start) / 1000) + " seconds"); 1025 } 1026 } 1027 1028 /** 1029 * Adds the {@code MasterQuotasObserver} to the list of configured Master observers to 1030 * automatically remove quotas for a table when that table is deleted. 1031 */ 1032 @VisibleForTesting 1033 public void updateConfigurationForQuotasObserver(Configuration conf) { 1034 // We're configured to not delete quotas on table deletion, so we don't need to add the obs. 1035 if (!conf.getBoolean( 1036 MasterQuotasObserver.REMOVE_QUOTA_ON_TABLE_DELETE, 1037 MasterQuotasObserver.REMOVE_QUOTA_ON_TABLE_DELETE_DEFAULT)) { 1038 return; 1039 } 1040 String[] masterCoprocs = conf.getStrings(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY); 1041 final int length = null == masterCoprocs ? 0 : masterCoprocs.length; 1042 String[] updatedCoprocs = new String[length + 1]; 1043 if (length > 0) { 1044 System.arraycopy(masterCoprocs, 0, updatedCoprocs, 0, masterCoprocs.length); 1045 } 1046 updatedCoprocs[length] = MasterQuotasObserver.class.getName(); 1047 conf.setStrings(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY, updatedCoprocs); 1048 } 1049 1050 private void initMobCleaner() { 1051 this.expiredMobFileCleanerChore = new ExpiredMobFileCleanerChore(this); 1052 getChoreService().scheduleChore(expiredMobFileCleanerChore); 1053 1054 int mobCompactionPeriod = conf.getInt(MobConstants.MOB_COMPACTION_CHORE_PERIOD, 1055 MobConstants.DEFAULT_MOB_COMPACTION_CHORE_PERIOD); 1056 if (mobCompactionPeriod > 0) { 1057 this.mobCompactChore = new MobCompactionChore(this, mobCompactionPeriod); 1058 getChoreService().scheduleChore(mobCompactChore); 1059 } else { 1060 LOG 1061 .info("The period is " + mobCompactionPeriod + " seconds, MobCompactionChore is disabled"); 1062 } 1063 this.mobCompactThread = new MasterMobCompactionThread(this); 1064 } 1065 1066 /** 1067 * <p> 1068 * Create a {@link MasterMetaBootstrap} instance. 1069 * </p> 1070 * <p> 1071 * Will be overridden in tests. 1072 * </p> 1073 */ 1074 @VisibleForTesting 1075 protected MasterMetaBootstrap createMetaBootstrap() { 1076 // We put this out here in a method so can do a Mockito.spy and stub it out 1077 // w/ a mocked up MasterMetaBootstrap. 1078 return new MasterMetaBootstrap(this); 1079 } 1080 1081 /** 1082 * <p> 1083 * Create a {@link ServerManager} instance. 1084 * </p> 1085 * <p> 1086 * Will be overridden in tests. 1087 * </p> 1088 */ 1089 @VisibleForTesting 1090 protected ServerManager createServerManager(final MasterServices master) throws IOException { 1091 // We put this out here in a method so can do a Mockito.spy and stub it out 1092 // w/ a mocked up ServerManager. 1093 setupClusterConnection(); 1094 return new ServerManager(master); 1095 } 1096 1097 private void waitForRegionServers(final MonitoredTask status) 1098 throws IOException, InterruptedException { 1099 this.serverManager.waitForRegionServers(status); 1100 } 1101 1102 // Will be overridden in tests 1103 @VisibleForTesting 1104 protected void initClusterSchemaService() throws IOException, InterruptedException { 1105 this.clusterSchemaService = new ClusterSchemaServiceImpl(this); 1106 this.clusterSchemaService.startAsync(); 1107 try { 1108 this.clusterSchemaService.awaitRunning(getConfiguration().getInt( 1109 HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS, 1110 DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS), TimeUnit.SECONDS); 1111 } catch (TimeoutException toe) { 1112 throw new IOException("Timedout starting ClusterSchemaService", toe); 1113 } 1114 } 1115 1116 private void initQuotaManager() throws IOException { 1117 MasterQuotaManager quotaManager = new MasterQuotaManager(this); 1118 quotaManager.start(); 1119 this.quotaManager = quotaManager; 1120 } 1121 1122 private SpaceQuotaSnapshotNotifier createQuotaSnapshotNotifier() { 1123 SpaceQuotaSnapshotNotifier notifier = 1124 SpaceQuotaSnapshotNotifierFactory.getInstance().create(getConfiguration()); 1125 return notifier; 1126 } 1127 1128 boolean isCatalogJanitorEnabled() { 1129 return catalogJanitorChore != null ? 1130 catalogJanitorChore.getEnabled() : false; 1131 } 1132 1133 boolean isCleanerChoreEnabled() { 1134 boolean hfileCleanerFlag = true, logCleanerFlag = true; 1135 1136 if (hfileCleaner != null) { 1137 hfileCleanerFlag = hfileCleaner.getEnabled(); 1138 } 1139 1140 if (logCleaner != null) { 1141 logCleanerFlag = logCleaner.getEnabled(); 1142 } 1143 1144 return (hfileCleanerFlag && logCleanerFlag); 1145 } 1146 1147 @Override 1148 public TableDescriptors getTableDescriptors() { 1149 return this.tableDescriptors; 1150 } 1151 1152 @Override 1153 public ServerManager getServerManager() { 1154 return this.serverManager; 1155 } 1156 1157 @Override 1158 public MasterFileSystem getMasterFileSystem() { 1159 return this.fileSystemManager; 1160 } 1161 1162 @Override 1163 public MasterWalManager getMasterWalManager() { 1164 return this.walManager; 1165 } 1166 1167 @Override 1168 public TableStateManager getTableStateManager() { 1169 return tableStateManager; 1170 } 1171 1172 /* 1173 * Start up all services. If any of these threads gets an unhandled exception 1174 * then they just die with a logged message. This should be fine because 1175 * in general, we do not expect the master to get such unhandled exceptions 1176 * as OOMEs; it should be lightly loaded. See what HRegionServer does if 1177 * need to install an unexpected exception handler. 1178 */ 1179 private void startServiceThreads() throws IOException{ 1180 // Start the executor service pools 1181 this.executorService.startExecutorService(ExecutorType.MASTER_OPEN_REGION, 1182 conf.getInt("hbase.master.executor.openregion.threads", 5)); 1183 this.executorService.startExecutorService(ExecutorType.MASTER_CLOSE_REGION, 1184 conf.getInt("hbase.master.executor.closeregion.threads", 5)); 1185 this.executorService.startExecutorService(ExecutorType.MASTER_SERVER_OPERATIONS, 1186 conf.getInt("hbase.master.executor.serverops.threads", 5)); 1187 this.executorService.startExecutorService(ExecutorType.MASTER_META_SERVER_OPERATIONS, 1188 conf.getInt("hbase.master.executor.meta.serverops.threads", 5)); 1189 this.executorService.startExecutorService(ExecutorType.M_LOG_REPLAY_OPS, 1190 conf.getInt("hbase.master.executor.logreplayops.threads", 10)); 1191 1192 // We depend on there being only one instance of this executor running 1193 // at a time. To do concurrency, would need fencing of enable/disable of 1194 // tables. 1195 // Any time changing this maxThreads to > 1, pls see the comment at 1196 // AccessController#postCompletedCreateTableAction 1197 this.executorService.startExecutorService(ExecutorType.MASTER_TABLE_OPERATIONS, 1); 1198 startProcedureExecutor(); 1199 1200 // Start log cleaner thread 1201 int cleanerInterval = conf.getInt("hbase.master.cleaner.interval", 600 * 1000); 1202 this.logCleaner = 1203 new LogCleaner(cleanerInterval, 1204 this, conf, getMasterWalManager().getFileSystem(), 1205 getMasterWalManager().getOldLogDir()); 1206 getChoreService().scheduleChore(logCleaner); 1207 1208 //start the hfile archive cleaner thread 1209 Path archiveDir = HFileArchiveUtil.getArchivePath(conf); 1210 Map<String, Object> params = new HashMap<>(); 1211 params.put(MASTER, this); 1212 this.hfileCleaner = new HFileCleaner(cleanerInterval, this, conf, getMasterFileSystem() 1213 .getFileSystem(), archiveDir, params); 1214 getChoreService().scheduleChore(hfileCleaner); 1215 serviceStarted = true; 1216 if (LOG.isTraceEnabled()) { 1217 LOG.trace("Started service threads"); 1218 } 1219 // Start replication zk node cleaner 1220 try { 1221 replicationZKNodeCleanerChore = new ReplicationZKNodeCleanerChore(this, cleanerInterval, 1222 new ReplicationZKNodeCleaner(this.conf, this.getZooKeeper(), this)); 1223 getChoreService().scheduleChore(replicationZKNodeCleanerChore); 1224 } catch (Exception e) { 1225 LOG.error("start replicationZKNodeCleanerChore failed", e); 1226 } 1227 } 1228 1229 @Override 1230 protected void stopServiceThreads() { 1231 if (masterJettyServer != null) { 1232 LOG.info("Stopping master jetty server"); 1233 try { 1234 masterJettyServer.stop(); 1235 } catch (Exception e) { 1236 LOG.error("Failed to stop master jetty server", e); 1237 } 1238 } 1239 super.stopServiceThreads(); 1240 stopChores(); 1241 1242 if (LOG.isDebugEnabled()) { 1243 LOG.debug("Stopping service threads"); 1244 } 1245 1246 // Clean up and close up shop 1247 if (this.logCleaner != null) this.logCleaner.cancel(true); 1248 if (this.hfileCleaner != null) this.hfileCleaner.cancel(true); 1249 if (this.replicationZKNodeCleanerChore != null) this.replicationZKNodeCleanerChore.cancel(true); 1250 if (this.quotaManager != null) this.quotaManager.stop(); 1251 1252 if (this.activeMasterManager != null) this.activeMasterManager.stop(); 1253 if (this.serverManager != null) this.serverManager.stop(); 1254 if (this.assignmentManager != null) this.assignmentManager.stop(); 1255 1256 stopProcedureExecutor(); 1257 1258 if (this.walManager != null) this.walManager.stop(); 1259 if (this.fileSystemManager != null) this.fileSystemManager.stop(); 1260 if (this.mpmHost != null) this.mpmHost.stop("server shutting down."); 1261 } 1262 1263 private void createProcedureExecutor() throws IOException { 1264 MasterProcedureEnv procEnv = new MasterProcedureEnv(this); 1265 procedureStore = 1266 new WALProcedureStore(conf, new MasterProcedureEnv.WALStoreLeaseRecovery(this)); 1267 procedureStore.registerListener(new MasterProcedureEnv.MasterProcedureStoreListener(this)); 1268 MasterProcedureScheduler procedureScheduler = procEnv.getProcedureScheduler(); 1269 procedureExecutor = new ProcedureExecutor<>(conf, procEnv, procedureStore, procedureScheduler); 1270 configurationManager.registerObserver(procEnv); 1271 1272 int cpus = Runtime.getRuntime().availableProcessors(); 1273 final int numThreads = conf.getInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, Math.max( 1274 (cpus > 0 ? cpus / 4 : 0), MasterProcedureConstants.DEFAULT_MIN_MASTER_PROCEDURE_THREADS)); 1275 final boolean abortOnCorruption = 1276 conf.getBoolean(MasterProcedureConstants.EXECUTOR_ABORT_ON_CORRUPTION, 1277 MasterProcedureConstants.DEFAULT_EXECUTOR_ABORT_ON_CORRUPTION); 1278 procedureStore.start(numThreads); 1279 // Just initialize it but do not start the workers, we will start the workers later by calling 1280 // startProcedureExecutor. See the javadoc for finishActiveMasterInitialization for more 1281 // details. 1282 procedureExecutor.init(numThreads, abortOnCorruption); 1283 procEnv.getRemoteDispatcher().start(); 1284 } 1285 1286 private void startProcedureExecutor() throws IOException { 1287 procedureExecutor.startWorkers(); 1288 } 1289 1290 private void stopProcedureExecutor() { 1291 if (procedureExecutor != null) { 1292 configurationManager.deregisterObserver(procedureExecutor.getEnvironment()); 1293 procedureExecutor.getEnvironment().getRemoteDispatcher().stop(); 1294 procedureExecutor.stop(); 1295 procedureExecutor.join(); 1296 procedureExecutor = null; 1297 } 1298 1299 if (procedureStore != null) { 1300 procedureStore.stop(isAborted()); 1301 procedureStore = null; 1302 } 1303 if (this.regionServerTracker != null) { 1304 this.regionServerTracker.stop(); 1305 } 1306 } 1307 1308 private void stopChores() { 1309 if (this.expiredMobFileCleanerChore != null) { 1310 this.expiredMobFileCleanerChore.cancel(true); 1311 } 1312 if (this.mobCompactChore != null) { 1313 this.mobCompactChore.cancel(true); 1314 } 1315 if (this.balancerChore != null) { 1316 this.balancerChore.cancel(true); 1317 } 1318 if (this.normalizerChore != null) { 1319 this.normalizerChore.cancel(true); 1320 } 1321 if (this.clusterStatusChore != null) { 1322 this.clusterStatusChore.cancel(true); 1323 } 1324 if (this.catalogJanitorChore != null) { 1325 this.catalogJanitorChore.cancel(true); 1326 } 1327 if (this.clusterStatusPublisherChore != null){ 1328 clusterStatusPublisherChore.cancel(true); 1329 } 1330 if (this.mobCompactThread != null) { 1331 this.mobCompactThread.close(); 1332 } 1333 1334 if (this.quotaObserverChore != null) { 1335 quotaObserverChore.cancel(); 1336 } 1337 if (this.snapshotQuotaChore != null) { 1338 snapshotQuotaChore.cancel(); 1339 } 1340 } 1341 1342 /** 1343 * @return Get remote side's InetAddress 1344 */ 1345 InetAddress getRemoteInetAddress(final int port, 1346 final long serverStartCode) throws UnknownHostException { 1347 // Do it out here in its own little method so can fake an address when 1348 // mocking up in tests. 1349 InetAddress ia = RpcServer.getRemoteIp(); 1350 1351 // The call could be from the local regionserver, 1352 // in which case, there is no remote address. 1353 if (ia == null && serverStartCode == startcode) { 1354 InetSocketAddress isa = rpcServices.getSocketAddress(); 1355 if (isa != null && isa.getPort() == port) { 1356 ia = isa.getAddress(); 1357 } 1358 } 1359 return ia; 1360 } 1361 1362 /** 1363 * @return Maximum time we should run balancer for 1364 */ 1365 private int getMaxBalancingTime() { 1366 int maxBalancingTime = getConfiguration().getInt(HConstants.HBASE_BALANCER_MAX_BALANCING, -1); 1367 if (maxBalancingTime == -1) { 1368 // if max balancing time isn't set, defaulting it to period time 1369 maxBalancingTime = getConfiguration().getInt(HConstants.HBASE_BALANCER_PERIOD, 1370 HConstants.DEFAULT_HBASE_BALANCER_PERIOD); 1371 } 1372 return maxBalancingTime; 1373 } 1374 1375 /** 1376 * @return Maximum number of regions in transition 1377 */ 1378 private int getMaxRegionsInTransition() { 1379 int numRegions = this.assignmentManager.getRegionStates().getRegionAssignments().size(); 1380 return Math.max((int) Math.floor(numRegions * this.maxRitPercent), 1); 1381 } 1382 1383 /** 1384 * It first sleep to the next balance plan start time. Meanwhile, throttling by the max 1385 * number regions in transition to protect availability. 1386 * @param nextBalanceStartTime The next balance plan start time 1387 * @param maxRegionsInTransition max number of regions in transition 1388 * @param cutoffTime when to exit balancer 1389 */ 1390 private void balanceThrottling(long nextBalanceStartTime, int maxRegionsInTransition, 1391 long cutoffTime) { 1392 boolean interrupted = false; 1393 1394 // Sleep to next balance plan start time 1395 // But if there are zero regions in transition, it can skip sleep to speed up. 1396 while (!interrupted && System.currentTimeMillis() < nextBalanceStartTime 1397 && this.assignmentManager.getRegionStates().hasRegionsInTransition()) { 1398 try { 1399 Thread.sleep(100); 1400 } catch (InterruptedException ie) { 1401 interrupted = true; 1402 } 1403 } 1404 1405 // Throttling by max number regions in transition 1406 while (!interrupted 1407 && maxRegionsInTransition > 0 1408 && this.assignmentManager.getRegionStates().getRegionsInTransitionCount() 1409 >= maxRegionsInTransition && System.currentTimeMillis() <= cutoffTime) { 1410 try { 1411 // sleep if the number of regions in transition exceeds the limit 1412 Thread.sleep(100); 1413 } catch (InterruptedException ie) { 1414 interrupted = true; 1415 } 1416 } 1417 1418 if (interrupted) Thread.currentThread().interrupt(); 1419 } 1420 1421 public boolean balance() throws IOException { 1422 return balance(false); 1423 } 1424 1425 public boolean balance(boolean force) throws IOException { 1426 // if master not initialized, don't run balancer. 1427 if (!isInitialized()) { 1428 LOG.debug("Master has not been initialized, don't run balancer."); 1429 return false; 1430 } 1431 1432 if (isInMaintenanceMode()) { 1433 LOG.info("Master is in maintenanceMode mode, don't run balancer."); 1434 return false; 1435 } 1436 1437 int maxRegionsInTransition = getMaxRegionsInTransition(); 1438 synchronized (this.balancer) { 1439 // If balance not true, don't run balancer. 1440 if (!this.loadBalancerTracker.isBalancerOn()) return false; 1441 // Only allow one balance run at at time. 1442 if (this.assignmentManager.hasRegionsInTransition()) { 1443 List<RegionStateNode> regionsInTransition = assignmentManager.getRegionsInTransition(); 1444 // if hbase:meta region is in transition, result of assignment cannot be recorded 1445 // ignore the force flag in that case 1446 boolean metaInTransition = assignmentManager.isMetaRegionInTransition(); 1447 String prefix = force && !metaInTransition ? "R" : "Not r"; 1448 List<RegionStateNode> toPrint = regionsInTransition; 1449 int max = 5; 1450 boolean truncated = false; 1451 if (regionsInTransition.size() > max) { 1452 toPrint = regionsInTransition.subList(0, max); 1453 truncated = true; 1454 } 1455 LOG.info(prefix + "unning balancer because " + regionsInTransition.size() + 1456 " region(s) in transition: " + toPrint + (truncated? "(truncated list)": "")); 1457 if (!force || metaInTransition) return false; 1458 } 1459 if (this.serverManager.areDeadServersInProgress()) { 1460 LOG.info("Not running balancer because processing dead regionserver(s): " + 1461 this.serverManager.getDeadServers()); 1462 return false; 1463 } 1464 1465 if (this.cpHost != null) { 1466 try { 1467 if (this.cpHost.preBalance()) { 1468 LOG.debug("Coprocessor bypassing balancer request"); 1469 return false; 1470 } 1471 } catch (IOException ioe) { 1472 LOG.error("Error invoking master coprocessor preBalance()", ioe); 1473 return false; 1474 } 1475 } 1476 1477 boolean isByTable = getConfiguration().getBoolean("hbase.master.loadbalance.bytable", false); 1478 Map<TableName, Map<ServerName, List<RegionInfo>>> assignmentsByTable = 1479 this.assignmentManager.getRegionStates().getAssignmentsByTable(!isByTable); 1480 1481 List<RegionPlan> plans = new ArrayList<>(); 1482 1483 //Give the balancer the current cluster state. 1484 this.balancer.setClusterMetrics(getClusterMetricsWithoutCoprocessor()); 1485 this.balancer.setClusterLoad(assignmentsByTable); 1486 1487 for (Map<ServerName, List<RegionInfo>> serverMap : assignmentsByTable.values()) { 1488 serverMap.keySet().removeAll(this.serverManager.getDrainingServersList()); 1489 } 1490 for (Entry<TableName, Map<ServerName, List<RegionInfo>>> e : assignmentsByTable.entrySet()) { 1491 List<RegionPlan> partialPlans = this.balancer.balanceCluster(e.getKey(), e.getValue()); 1492 if (partialPlans != null) plans.addAll(partialPlans); 1493 } 1494 1495 long balanceStartTime = System.currentTimeMillis(); 1496 long cutoffTime = balanceStartTime + this.maxBlancingTime; 1497 int rpCount = 0; // number of RegionPlans balanced so far 1498 if (plans != null && !plans.isEmpty()) { 1499 int balanceInterval = this.maxBlancingTime / plans.size(); 1500 LOG.info("Balancer plans size is " + plans.size() + ", the balance interval is " 1501 + balanceInterval + " ms, and the max number regions in transition is " 1502 + maxRegionsInTransition); 1503 1504 for (RegionPlan plan: plans) { 1505 LOG.info("balance " + plan); 1506 //TODO: bulk assign 1507 this.assignmentManager.moveAsync(plan); 1508 rpCount++; 1509 1510 balanceThrottling(balanceStartTime + rpCount * balanceInterval, maxRegionsInTransition, 1511 cutoffTime); 1512 1513 // if performing next balance exceeds cutoff time, exit the loop 1514 if (rpCount < plans.size() && System.currentTimeMillis() > cutoffTime) { 1515 // TODO: After balance, there should not be a cutoff time (keeping it as 1516 // a security net for now) 1517 LOG.debug("No more balancing till next balance run; maxBalanceTime=" 1518 + this.maxBlancingTime); 1519 break; 1520 } 1521 } 1522 } 1523 1524 if (this.cpHost != null) { 1525 try { 1526 this.cpHost.postBalance(rpCount < plans.size() ? plans.subList(0, rpCount) : plans); 1527 } catch (IOException ioe) { 1528 // balancing already succeeded so don't change the result 1529 LOG.error("Error invoking master coprocessor postBalance()", ioe); 1530 } 1531 } 1532 } 1533 // If LoadBalancer did not generate any plans, it means the cluster is already balanced. 1534 // Return true indicating a success. 1535 return true; 1536 } 1537 1538 @Override 1539 @VisibleForTesting 1540 public RegionNormalizer getRegionNormalizer() { 1541 return this.normalizer; 1542 } 1543 1544 /** 1545 * Perform normalization of cluster (invoked by {@link RegionNormalizerChore}). 1546 * 1547 * @return true if normalization step was performed successfully, false otherwise 1548 * (specifically, if HMaster hasn't been initialized properly or normalization 1549 * is globally disabled) 1550 */ 1551 public boolean normalizeRegions() throws IOException { 1552 if (!isInitialized()) { 1553 LOG.debug("Master has not been initialized, don't run region normalizer."); 1554 return false; 1555 } 1556 1557 if (isInMaintenanceMode()) { 1558 LOG.info("Master is in maintenance mode, don't run region normalizer."); 1559 return false; 1560 } 1561 1562 if (!this.regionNormalizerTracker.isNormalizerOn()) { 1563 LOG.debug("Region normalization is disabled, don't run region normalizer."); 1564 return false; 1565 } 1566 1567 synchronized (this.normalizer) { 1568 // Don't run the normalizer concurrently 1569 List<TableName> allEnabledTables = new ArrayList<>( 1570 this.tableStateManager.getTablesInStates(TableState.State.ENABLED)); 1571 1572 Collections.shuffle(allEnabledTables); 1573 1574 for (TableName table : allEnabledTables) { 1575 if (isInMaintenanceMode()) { 1576 LOG.debug("Master is in maintenance mode, stop running region normalizer."); 1577 return false; 1578 } 1579 1580 TableDescriptor tblDesc = getTableDescriptors().get(table); 1581 if (table.isSystemTable() || (tblDesc != null && 1582 !tblDesc.isNormalizationEnabled())) { 1583 LOG.trace("Skipping normalization for {}, as it's either system" 1584 + " table or doesn't have auto normalization turned on", table); 1585 continue; 1586 } 1587 List<NormalizationPlan> plans = this.normalizer.computePlanForTable(table); 1588 if (plans != null) { 1589 for (NormalizationPlan plan : plans) { 1590 plan.execute(clusterConnection.getAdmin()); 1591 if (plan.getType() == PlanType.SPLIT) { 1592 splitPlanCount++; 1593 } else if (plan.getType() == PlanType.MERGE) { 1594 mergePlanCount++; 1595 } 1596 } 1597 } 1598 } 1599 } 1600 // If Region did not generate any plans, it means the cluster is already balanced. 1601 // Return true indicating a success. 1602 return true; 1603 } 1604 1605 /** 1606 * @return Client info for use as prefix on an audit log string; who did an action 1607 */ 1608 @Override 1609 public String getClientIdAuditPrefix() { 1610 return "Client=" + RpcServer.getRequestUserName().orElse(null) 1611 + "/" + RpcServer.getRemoteAddress().orElse(null); 1612 } 1613 1614 /** 1615 * Switch for the background CatalogJanitor thread. 1616 * Used for testing. The thread will continue to run. It will just be a noop 1617 * if disabled. 1618 * @param b If false, the catalog janitor won't do anything. 1619 */ 1620 public void setCatalogJanitorEnabled(final boolean b) { 1621 this.catalogJanitorChore.setEnabled(b); 1622 } 1623 1624 @Override 1625 public long mergeRegions( 1626 final RegionInfo[] regionsToMerge, 1627 final boolean forcible, 1628 final long nonceGroup, 1629 final long nonce) throws IOException { 1630 checkInitialized(); 1631 1632 assert(regionsToMerge.length == 2); 1633 1634 TableName tableName = regionsToMerge[0].getTable(); 1635 if (tableName == null || regionsToMerge[1].getTable() == null) { 1636 throw new UnknownRegionException ("Can't merge regions without table associated"); 1637 } 1638 1639 if (!tableName.equals(regionsToMerge[1].getTable())) { 1640 throw new IOException ( 1641 "Cannot merge regions from two different tables " + regionsToMerge[0].getTable() 1642 + " and " + regionsToMerge[1].getTable()); 1643 } 1644 1645 if (RegionInfo.COMPARATOR.compare(regionsToMerge[0], regionsToMerge[1]) == 0) { 1646 throw new MergeRegionException( 1647 "Cannot merge a region to itself " + regionsToMerge[0] + ", " + regionsToMerge[1]); 1648 } 1649 1650 return MasterProcedureUtil.submitProcedure( 1651 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 1652 @Override 1653 protected void run() throws IOException { 1654 getMaster().getMasterCoprocessorHost().preMergeRegions(regionsToMerge); 1655 1656 LOG.info(getClientIdAuditPrefix() + " Merge regions " + 1657 regionsToMerge[0].getEncodedName() + " and " + regionsToMerge[1].getEncodedName()); 1658 1659 submitProcedure(new MergeTableRegionsProcedure(procedureExecutor.getEnvironment(), 1660 regionsToMerge, forcible)); 1661 1662 getMaster().getMasterCoprocessorHost().postMergeRegions(regionsToMerge); 1663 } 1664 1665 @Override 1666 protected String getDescription() { 1667 return "MergeTableProcedure"; 1668 } 1669 }); 1670 } 1671 1672 @Override 1673 public long splitRegion(final RegionInfo regionInfo, final byte[] splitRow, 1674 final long nonceGroup, final long nonce) 1675 throws IOException { 1676 checkInitialized(); 1677 return MasterProcedureUtil.submitProcedure( 1678 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 1679 @Override 1680 protected void run() throws IOException { 1681 getMaster().getMasterCoprocessorHost().preSplitRegion(regionInfo.getTable(), splitRow); 1682 LOG.info(getClientIdAuditPrefix() + " split " + regionInfo.getRegionNameAsString()); 1683 1684 // Execute the operation asynchronously 1685 submitProcedure(getAssignmentManager().createSplitProcedure(regionInfo, splitRow)); 1686 } 1687 1688 @Override 1689 protected String getDescription() { 1690 return "SplitTableProcedure"; 1691 } 1692 }); 1693 } 1694 1695 // Public so can be accessed by tests. Blocks until move is done. 1696 // Replace with an async implementation from which you can get 1697 // a success/failure result. 1698 @VisibleForTesting 1699 public void move(final byte[] encodedRegionName, byte[] destServerName) throws HBaseIOException { 1700 RegionState regionState = assignmentManager.getRegionStates(). 1701 getRegionState(Bytes.toString(encodedRegionName)); 1702 1703 RegionInfo hri; 1704 if (regionState != null) { 1705 hri = regionState.getRegion(); 1706 } else { 1707 throw new UnknownRegionException(Bytes.toStringBinary(encodedRegionName)); 1708 } 1709 1710 ServerName dest; 1711 List<ServerName> exclude = hri.getTable().isSystemTable() ? assignmentManager.getExcludedServersForSystemTable() 1712 : new ArrayList<>(1); 1713 if (destServerName != null && exclude.contains(ServerName.valueOf(Bytes.toString(destServerName)))) { 1714 LOG.info( 1715 Bytes.toString(encodedRegionName) + " can not move to " + Bytes.toString(destServerName) 1716 + " because the server is in exclude list"); 1717 destServerName = null; 1718 } 1719 if (destServerName == null || destServerName.length == 0) { 1720 LOG.info("Passed destination servername is null/empty so " + 1721 "choosing a server at random"); 1722 exclude.add(regionState.getServerName()); 1723 final List<ServerName> destServers = this.serverManager.createDestinationServersList(exclude); 1724 dest = balancer.randomAssignment(hri, destServers); 1725 if (dest == null) { 1726 LOG.debug("Unable to determine a plan to assign " + hri); 1727 return; 1728 } 1729 } else { 1730 ServerName candidate = ServerName.valueOf(Bytes.toString(destServerName)); 1731 dest = balancer.randomAssignment(hri, Lists.newArrayList(candidate)); 1732 if (dest == null) { 1733 LOG.debug("Unable to determine a plan to assign " + hri); 1734 return; 1735 } 1736 // TODO: What is this? I don't get it. 1737 if (dest.equals(serverName) && balancer instanceof BaseLoadBalancer 1738 && !((BaseLoadBalancer)balancer).shouldBeOnMaster(hri)) { 1739 // To avoid unnecessary region moving later by balancer. Don't put user 1740 // regions on master. 1741 LOG.debug("Skipping move of region " + hri.getRegionNameAsString() 1742 + " to avoid unnecessary region moving later by load balancer," 1743 + " because it should not be on master"); 1744 return; 1745 } 1746 } 1747 1748 if (dest.equals(regionState.getServerName())) { 1749 LOG.debug("Skipping move of region " + hri.getRegionNameAsString() 1750 + " because region already assigned to the same server " + dest + "."); 1751 return; 1752 } 1753 1754 // Now we can do the move 1755 RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), dest); 1756 assert rp.getDestination() != null: rp.toString() + " " + dest; 1757 1758 try { 1759 checkInitialized(); 1760 if (this.cpHost != null) { 1761 this.cpHost.preMove(hri, rp.getSource(), rp.getDestination()); 1762 } 1763 // Warmup the region on the destination before initiating the move. this call 1764 // is synchronous and takes some time. doing it before the source region gets 1765 // closed 1766 serverManager.sendRegionWarmup(rp.getDestination(), hri); 1767 1768 LOG.info(getClientIdAuditPrefix() + " move " + rp + ", running balancer"); 1769 Future<byte []> future = this.assignmentManager.moveAsync(rp); 1770 try { 1771 // Is this going to work? Will we throw exception on error? 1772 // TODO: CompletableFuture rather than this stunted Future. 1773 future.get(); 1774 } catch (InterruptedException | ExecutionException e) { 1775 throw new HBaseIOException(e); 1776 } 1777 if (this.cpHost != null) { 1778 this.cpHost.postMove(hri, rp.getSource(), rp.getDestination()); 1779 } 1780 } catch (IOException ioe) { 1781 if (ioe instanceof HBaseIOException) { 1782 throw (HBaseIOException)ioe; 1783 } 1784 throw new HBaseIOException(ioe); 1785 } 1786 } 1787 1788 @Override 1789 public long createTable( 1790 final TableDescriptor tableDescriptor, 1791 final byte [][] splitKeys, 1792 final long nonceGroup, 1793 final long nonce) throws IOException { 1794 checkInitialized(); 1795 1796 String namespace = tableDescriptor.getTableName().getNamespaceAsString(); 1797 this.clusterSchemaService.getNamespace(namespace); 1798 1799 RegionInfo[] newRegions = ModifyRegionUtils.createRegionInfos(tableDescriptor, splitKeys); 1800 sanityCheckTableDescriptor(tableDescriptor); 1801 1802 return MasterProcedureUtil.submitProcedure( 1803 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 1804 @Override 1805 protected void run() throws IOException { 1806 getMaster().getMasterCoprocessorHost().preCreateTable(tableDescriptor, newRegions); 1807 1808 LOG.info(getClientIdAuditPrefix() + " create " + tableDescriptor); 1809 1810 // TODO: We can handle/merge duplicate requests, and differentiate the case of 1811 // TableExistsException by saying if the schema is the same or not. 1812 // 1813 // We need to wait for the procedure to potentially fail due to "prepare" sanity 1814 // checks. This will block only the beginning of the procedure. See HBASE-19953. 1815 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch(); 1816 submitProcedure(new CreateTableProcedure( 1817 procedureExecutor.getEnvironment(), tableDescriptor, newRegions, latch)); 1818 latch.await(); 1819 1820 getMaster().getMasterCoprocessorHost().postCreateTable(tableDescriptor, newRegions); 1821 } 1822 1823 @Override 1824 protected String getDescription() { 1825 return "CreateTableProcedure"; 1826 } 1827 }); 1828 } 1829 1830 @Override 1831 public long createSystemTable(final TableDescriptor tableDescriptor) throws IOException { 1832 if (isStopped()) { 1833 throw new MasterNotRunningException(); 1834 } 1835 1836 TableName tableName = tableDescriptor.getTableName(); 1837 if (!(tableName.isSystemTable())) { 1838 throw new IllegalArgumentException( 1839 "Only system table creation can use this createSystemTable API"); 1840 } 1841 1842 RegionInfo[] newRegions = ModifyRegionUtils.createRegionInfos(tableDescriptor, null); 1843 1844 LOG.info(getClientIdAuditPrefix() + " create " + tableDescriptor); 1845 1846 // This special create table is called locally to master. Therefore, no RPC means no need 1847 // to use nonce to detect duplicated RPC call. 1848 long procId = this.procedureExecutor.submitProcedure( 1849 new CreateTableProcedure(procedureExecutor.getEnvironment(), tableDescriptor, newRegions)); 1850 1851 return procId; 1852 } 1853 1854 /** 1855 * Checks whether the table conforms to some sane limits, and configured 1856 * values (compression, etc) work. Throws an exception if something is wrong. 1857 * @throws IOException 1858 */ 1859 private void sanityCheckTableDescriptor(final TableDescriptor htd) throws IOException { 1860 final String CONF_KEY = "hbase.table.sanity.checks"; 1861 boolean logWarn = false; 1862 if (!conf.getBoolean(CONF_KEY, true)) { 1863 logWarn = true; 1864 } 1865 String tableVal = htd.getValue(CONF_KEY); 1866 if (tableVal != null && !Boolean.valueOf(tableVal)) { 1867 logWarn = true; 1868 } 1869 1870 // check max file size 1871 long maxFileSizeLowerLimit = 2 * 1024 * 1024L; // 2M is the default lower limit 1872 long maxFileSize = htd.getMaxFileSize(); 1873 if (maxFileSize < 0) { 1874 maxFileSize = conf.getLong(HConstants.HREGION_MAX_FILESIZE, maxFileSizeLowerLimit); 1875 } 1876 if (maxFileSize < conf.getLong("hbase.hregion.max.filesize.limit", maxFileSizeLowerLimit)) { 1877 String message = "MAX_FILESIZE for table descriptor or " 1878 + "\"hbase.hregion.max.filesize\" (" + maxFileSize 1879 + ") is too small, which might cause over splitting into unmanageable " 1880 + "number of regions."; 1881 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1882 } 1883 1884 // check flush size 1885 long flushSizeLowerLimit = 1024 * 1024L; // 1M is the default lower limit 1886 long flushSize = htd.getMemStoreFlushSize(); 1887 if (flushSize < 0) { 1888 flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, flushSizeLowerLimit); 1889 } 1890 if (flushSize < conf.getLong("hbase.hregion.memstore.flush.size.limit", flushSizeLowerLimit)) { 1891 String message = "MEMSTORE_FLUSHSIZE for table descriptor or " 1892 + "\"hbase.hregion.memstore.flush.size\" ("+flushSize+") is too small, which might cause" 1893 + " very frequent flushing."; 1894 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1895 } 1896 1897 // check that coprocessors and other specified plugin classes can be loaded 1898 try { 1899 checkClassLoading(conf, htd); 1900 } catch (Exception ex) { 1901 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, ex.getMessage(), null); 1902 } 1903 1904 // check compression can be loaded 1905 try { 1906 checkCompression(htd); 1907 } catch (IOException e) { 1908 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e); 1909 } 1910 1911 // check encryption can be loaded 1912 try { 1913 checkEncryption(conf, htd); 1914 } catch (IOException e) { 1915 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e); 1916 } 1917 // Verify compaction policy 1918 try{ 1919 checkCompactionPolicy(conf, htd); 1920 } catch(IOException e){ 1921 warnOrThrowExceptionForFailure(false, CONF_KEY, e.getMessage(), e); 1922 } 1923 // check that we have at least 1 CF 1924 if (htd.getColumnFamilyCount() == 0) { 1925 String message = "Table should have at least one column family."; 1926 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1927 } 1928 1929 for (ColumnFamilyDescriptor hcd : htd.getColumnFamilies()) { 1930 if (hcd.getTimeToLive() <= 0) { 1931 String message = "TTL for column family " + hcd.getNameAsString() + " must be positive."; 1932 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1933 } 1934 1935 // check blockSize 1936 if (hcd.getBlocksize() < 1024 || hcd.getBlocksize() > 16 * 1024 * 1024) { 1937 String message = "Block size for column family " + hcd.getNameAsString() 1938 + " must be between 1K and 16MB."; 1939 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1940 } 1941 1942 // check versions 1943 if (hcd.getMinVersions() < 0) { 1944 String message = "Min versions for column family " + hcd.getNameAsString() 1945 + " must be positive."; 1946 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1947 } 1948 // max versions already being checked 1949 1950 // HBASE-13776 Setting illegal versions for ColumnFamilyDescriptor 1951 // does not throw IllegalArgumentException 1952 // check minVersions <= maxVerions 1953 if (hcd.getMinVersions() > hcd.getMaxVersions()) { 1954 String message = "Min versions for column family " + hcd.getNameAsString() 1955 + " must be less than the Max versions."; 1956 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1957 } 1958 1959 // check replication scope 1960 checkReplicationScope(hcd); 1961 1962 // check data replication factor, it can be 0(default value) when user has not explicitly 1963 // set the value, in this case we use default replication factor set in the file system. 1964 if (hcd.getDFSReplication() < 0) { 1965 String message = "HFile Replication for column family " + hcd.getNameAsString() 1966 + " must be greater than zero."; 1967 warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null); 1968 } 1969 1970 // TODO: should we check coprocessors and encryption ? 1971 } 1972 } 1973 1974 private void checkReplicationScope(ColumnFamilyDescriptor hcd) throws IOException{ 1975 // check replication scope 1976 WALProtos.ScopeType scop = WALProtos.ScopeType.valueOf(hcd.getScope()); 1977 if (scop == null) { 1978 String message = "Replication scope for column family " 1979 + hcd.getNameAsString() + " is " + hcd.getScope() + " which is invalid."; 1980 1981 LOG.error(message); 1982 throw new DoNotRetryIOException(message); 1983 } 1984 } 1985 1986 private void checkCompactionPolicy(Configuration conf, TableDescriptor htd) 1987 throws IOException { 1988 // FIFO compaction has some requirements 1989 // Actually FCP ignores periodic major compactions 1990 String className = htd.getValue(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY); 1991 if (className == null) { 1992 className = 1993 conf.get(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY, 1994 ExploringCompactionPolicy.class.getName()); 1995 } 1996 1997 int blockingFileCount = HStore.DEFAULT_BLOCKING_STOREFILE_COUNT; 1998 String sv = htd.getValue(HStore.BLOCKING_STOREFILES_KEY); 1999 if (sv != null) { 2000 blockingFileCount = Integer.parseInt(sv); 2001 } else { 2002 blockingFileCount = conf.getInt(HStore.BLOCKING_STOREFILES_KEY, blockingFileCount); 2003 } 2004 2005 for (ColumnFamilyDescriptor hcd : htd.getColumnFamilies()) { 2006 String compactionPolicy = 2007 hcd.getConfigurationValue(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY); 2008 if (compactionPolicy == null) { 2009 compactionPolicy = className; 2010 } 2011 if (!compactionPolicy.equals(FIFOCompactionPolicy.class.getName())) { 2012 continue; 2013 } 2014 // FIFOCompaction 2015 String message = null; 2016 2017 // 1. Check TTL 2018 if (hcd.getTimeToLive() == ColumnFamilyDescriptorBuilder.DEFAULT_TTL) { 2019 message = "Default TTL is not supported for FIFO compaction"; 2020 throw new IOException(message); 2021 } 2022 2023 // 2. Check min versions 2024 if (hcd.getMinVersions() > 0) { 2025 message = "MIN_VERSION > 0 is not supported for FIFO compaction"; 2026 throw new IOException(message); 2027 } 2028 2029 // 3. blocking file count 2030 sv = hcd.getConfigurationValue(HStore.BLOCKING_STOREFILES_KEY); 2031 if (sv != null) { 2032 blockingFileCount = Integer.parseInt(sv); 2033 } 2034 if (blockingFileCount < 1000) { 2035 message = 2036 "Blocking file count '" + HStore.BLOCKING_STOREFILES_KEY + "' " + blockingFileCount 2037 + " is below recommended minimum of 1000 for column family "+ hcd.getNameAsString(); 2038 throw new IOException(message); 2039 } 2040 } 2041 } 2042 2043 // HBASE-13350 - Helper method to log warning on sanity check failures if checks disabled. 2044 private static void warnOrThrowExceptionForFailure(boolean logWarn, String confKey, 2045 String message, Exception cause) throws IOException { 2046 if (!logWarn) { 2047 throw new DoNotRetryIOException(message + " Set " + confKey + 2048 " to false at conf or table descriptor if you want to bypass sanity checks", cause); 2049 } 2050 LOG.warn(message); 2051 } 2052 2053 private void startActiveMasterManager(int infoPort) throws KeeperException { 2054 String backupZNode = ZNodePaths.joinZNode( 2055 zooKeeper.znodePaths.backupMasterAddressesZNode, serverName.toString()); 2056 /* 2057 * Add a ZNode for ourselves in the backup master directory since we 2058 * may not become the active master. If so, we want the actual active 2059 * master to know we are backup masters, so that it won't assign 2060 * regions to us if so configured. 2061 * 2062 * If we become the active master later, ActiveMasterManager will delete 2063 * this node explicitly. If we crash before then, ZooKeeper will delete 2064 * this node for us since it is ephemeral. 2065 */ 2066 LOG.info("Adding backup master ZNode " + backupZNode); 2067 if (!MasterAddressTracker.setMasterAddress(zooKeeper, backupZNode, serverName, infoPort)) { 2068 LOG.warn("Failed create of " + backupZNode + " by " + serverName); 2069 } 2070 this.activeMasterManager.setInfoPort(infoPort); 2071 int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT, HConstants.DEFAULT_ZK_SESSION_TIMEOUT); 2072 // If we're a backup master, stall until a primary to write this address 2073 if (conf.getBoolean(HConstants.MASTER_TYPE_BACKUP, HConstants.DEFAULT_MASTER_TYPE_BACKUP)) { 2074 LOG.debug("HMaster started in backup mode. Stalling until master znode is written."); 2075 // This will only be a minute or so while the cluster starts up, 2076 // so don't worry about setting watches on the parent znode 2077 while (!activeMasterManager.hasActiveMaster()) { 2078 LOG.debug("Waiting for master address and cluster state znode to be written."); 2079 Threads.sleep(timeout); 2080 } 2081 } 2082 MonitoredTask status = TaskMonitor.get().createStatus("Master startup"); 2083 status.setDescription("Master startup"); 2084 try { 2085 if (activeMasterManager.blockUntilBecomingActiveMaster(timeout, status)) { 2086 finishActiveMasterInitialization(status); 2087 } 2088 } catch (Throwable t) { 2089 status.setStatus("Failed to become active: " + t.getMessage()); 2090 LOG.error(HBaseMarkers.FATAL, "Failed to become active master", t); 2091 // HBASE-5680: Likely hadoop23 vs hadoop 20.x/1.x incompatibility 2092 if (t instanceof NoClassDefFoundError && t.getMessage(). 2093 contains("org/apache/hadoop/hdfs/protocol/HdfsConstants$SafeModeAction")) { 2094 // improved error message for this special case 2095 abort("HBase is having a problem with its Hadoop jars. You may need to recompile " + 2096 "HBase against Hadoop version " + org.apache.hadoop.util.VersionInfo.getVersion() + 2097 " or change your hadoop jars to start properly", t); 2098 } else { 2099 abort("Unhandled exception. Starting shutdown.", t); 2100 } 2101 } finally { 2102 status.cleanup(); 2103 } 2104 } 2105 2106 private void checkCompression(final TableDescriptor htd) 2107 throws IOException { 2108 if (!this.masterCheckCompression) return; 2109 for (ColumnFamilyDescriptor hcd : htd.getColumnFamilies()) { 2110 checkCompression(hcd); 2111 } 2112 } 2113 2114 private void checkCompression(final ColumnFamilyDescriptor hcd) 2115 throws IOException { 2116 if (!this.masterCheckCompression) return; 2117 CompressionTest.testCompression(hcd.getCompressionType()); 2118 CompressionTest.testCompression(hcd.getCompactionCompressionType()); 2119 } 2120 2121 private void checkEncryption(final Configuration conf, final TableDescriptor htd) 2122 throws IOException { 2123 if (!this.masterCheckEncryption) return; 2124 for (ColumnFamilyDescriptor hcd : htd.getColumnFamilies()) { 2125 checkEncryption(conf, hcd); 2126 } 2127 } 2128 2129 private void checkEncryption(final Configuration conf, final ColumnFamilyDescriptor hcd) 2130 throws IOException { 2131 if (!this.masterCheckEncryption) return; 2132 EncryptionTest.testEncryption(conf, hcd.getEncryptionType(), hcd.getEncryptionKey()); 2133 } 2134 2135 private void checkClassLoading(final Configuration conf, final TableDescriptor htd) 2136 throws IOException { 2137 RegionSplitPolicy.getSplitPolicyClass(htd, conf); 2138 RegionCoprocessorHost.testTableCoprocessorAttrs(conf, htd); 2139 } 2140 2141 private static boolean isCatalogTable(final TableName tableName) { 2142 return tableName.equals(TableName.META_TABLE_NAME); 2143 } 2144 2145 @Override 2146 public long deleteTable( 2147 final TableName tableName, 2148 final long nonceGroup, 2149 final long nonce) throws IOException { 2150 checkInitialized(); 2151 2152 return MasterProcedureUtil.submitProcedure( 2153 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 2154 @Override 2155 protected void run() throws IOException { 2156 getMaster().getMasterCoprocessorHost().preDeleteTable(tableName); 2157 2158 LOG.info(getClientIdAuditPrefix() + " delete " + tableName); 2159 2160 // TODO: We can handle/merge duplicate request 2161 // 2162 // We need to wait for the procedure to potentially fail due to "prepare" sanity 2163 // checks. This will block only the beginning of the procedure. See HBASE-19953. 2164 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch(); 2165 submitProcedure(new DeleteTableProcedure(procedureExecutor.getEnvironment(), 2166 tableName, latch)); 2167 latch.await(); 2168 2169 getMaster().getMasterCoprocessorHost().postDeleteTable(tableName); 2170 } 2171 2172 @Override 2173 protected String getDescription() { 2174 return "DeleteTableProcedure"; 2175 } 2176 }); 2177 } 2178 2179 @Override 2180 public long truncateTable( 2181 final TableName tableName, 2182 final boolean preserveSplits, 2183 final long nonceGroup, 2184 final long nonce) throws IOException { 2185 checkInitialized(); 2186 2187 return MasterProcedureUtil.submitProcedure( 2188 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 2189 @Override 2190 protected void run() throws IOException { 2191 getMaster().getMasterCoprocessorHost().preTruncateTable(tableName); 2192 2193 LOG.info(getClientIdAuditPrefix() + " truncate " + tableName); 2194 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch(2, 0); 2195 submitProcedure(new TruncateTableProcedure(procedureExecutor.getEnvironment(), 2196 tableName, preserveSplits, latch)); 2197 latch.await(); 2198 2199 getMaster().getMasterCoprocessorHost().postTruncateTable(tableName); 2200 } 2201 2202 @Override 2203 protected String getDescription() { 2204 return "TruncateTableProcedure"; 2205 } 2206 }); 2207 } 2208 2209 @Override 2210 public long addColumn(final TableName tableName, final ColumnFamilyDescriptor column, 2211 final long nonceGroup, final long nonce) throws IOException { 2212 checkInitialized(); 2213 checkTableExists(tableName); 2214 2215 return modifyTable(tableName, new TableDescriptorGetter() { 2216 2217 @Override 2218 public TableDescriptor get() throws IOException { 2219 TableDescriptor old = getTableDescriptors().get(tableName); 2220 if (old.hasColumnFamily(column.getName())) { 2221 throw new InvalidFamilyOperationException("Column family '" + column.getNameAsString() 2222 + "' in table '" + tableName + "' already exists so cannot be added"); 2223 } 2224 2225 return TableDescriptorBuilder.newBuilder(old).setColumnFamily(column).build(); 2226 } 2227 }, nonceGroup, nonce); 2228 } 2229 2230 /** 2231 * Implement to return TableDescriptor after pre-checks 2232 */ 2233 protected interface TableDescriptorGetter { 2234 TableDescriptor get() throws IOException; 2235 } 2236 2237 @Override 2238 public long modifyColumn(final TableName tableName, final ColumnFamilyDescriptor descriptor, 2239 final long nonceGroup, final long nonce) throws IOException { 2240 checkInitialized(); 2241 checkTableExists(tableName); 2242 return modifyTable(tableName, new TableDescriptorGetter() { 2243 2244 @Override 2245 public TableDescriptor get() throws IOException { 2246 TableDescriptor old = getTableDescriptors().get(tableName); 2247 if (!old.hasColumnFamily(descriptor.getName())) { 2248 throw new InvalidFamilyOperationException("Family '" + descriptor.getNameAsString() 2249 + "' does not exist, so it cannot be modified"); 2250 } 2251 2252 return TableDescriptorBuilder.newBuilder(old).modifyColumnFamily(descriptor).build(); 2253 } 2254 }, nonceGroup, nonce); 2255 } 2256 2257 @Override 2258 public long deleteColumn(final TableName tableName, final byte[] columnName, 2259 final long nonceGroup, final long nonce) throws IOException { 2260 checkInitialized(); 2261 checkTableExists(tableName); 2262 2263 return modifyTable(tableName, new TableDescriptorGetter() { 2264 2265 @Override 2266 public TableDescriptor get() throws IOException { 2267 TableDescriptor old = getTableDescriptors().get(tableName); 2268 2269 if (!old.hasColumnFamily(columnName)) { 2270 throw new InvalidFamilyOperationException("Family '" + Bytes.toString(columnName) 2271 + "' does not exist, so it cannot be deleted"); 2272 } 2273 if (old.getColumnFamilyCount() == 1) { 2274 throw new InvalidFamilyOperationException("Family '" + Bytes.toString(columnName) 2275 + "' is the only column family in the table, so it cannot be deleted"); 2276 } 2277 return TableDescriptorBuilder.newBuilder(old).removeColumnFamily(columnName).build(); 2278 } 2279 }, nonceGroup, nonce); 2280 } 2281 2282 @Override 2283 public long enableTable(final TableName tableName, final long nonceGroup, final long nonce) 2284 throws IOException { 2285 checkInitialized(); 2286 2287 return MasterProcedureUtil.submitProcedure( 2288 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 2289 @Override 2290 protected void run() throws IOException { 2291 getMaster().getMasterCoprocessorHost().preEnableTable(tableName); 2292 2293 // Normally, it would make sense for this authorization check to exist inside 2294 // AccessController, but because the authorization check is done based on internal state 2295 // (rather than explicit permissions) we'll do the check here instead of in the 2296 // coprocessor. 2297 MasterQuotaManager quotaManager = getMasterQuotaManager(); 2298 if (quotaManager != null) { 2299 if (quotaManager.isQuotaInitialized()) { 2300 Quotas quotaForTable = QuotaUtil.getTableQuota(getConnection(), tableName); 2301 if (quotaForTable != null && quotaForTable.hasSpace()) { 2302 SpaceViolationPolicy policy = quotaForTable.getSpace().getViolationPolicy(); 2303 if (SpaceViolationPolicy.DISABLE == policy) { 2304 throw new AccessDeniedException("Enabling the table '" + tableName 2305 + "' is disallowed due to a violated space quota."); 2306 } 2307 } 2308 } else if (LOG.isTraceEnabled()) { 2309 LOG.trace("Unable to check for space quotas as the MasterQuotaManager is not enabled"); 2310 } 2311 } 2312 2313 LOG.info(getClientIdAuditPrefix() + " enable " + tableName); 2314 2315 // Execute the operation asynchronously - client will check the progress of the operation 2316 // In case the request is from a <1.1 client before returning, 2317 // we want to make sure that the table is prepared to be 2318 // enabled (the table is locked and the table state is set). 2319 // Note: if the procedure throws exception, we will catch it and rethrow. 2320 final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch(); 2321 submitProcedure(new EnableTableProcedure(procedureExecutor.getEnvironment(), 2322 tableName, false, prepareLatch)); 2323 prepareLatch.await(); 2324 2325 getMaster().getMasterCoprocessorHost().postEnableTable(tableName); 2326 } 2327 2328 @Override 2329 protected String getDescription() { 2330 return "EnableTableProcedure"; 2331 } 2332 }); 2333 } 2334 2335 @Override 2336 public long disableTable(final TableName tableName, final long nonceGroup, final long nonce) 2337 throws IOException { 2338 checkInitialized(); 2339 2340 return MasterProcedureUtil.submitProcedure( 2341 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 2342 @Override 2343 protected void run() throws IOException { 2344 getMaster().getMasterCoprocessorHost().preDisableTable(tableName); 2345 2346 LOG.info(getClientIdAuditPrefix() + " disable " + tableName); 2347 2348 // Execute the operation asynchronously - client will check the progress of the operation 2349 // In case the request is from a <1.1 client before returning, 2350 // we want to make sure that the table is prepared to be 2351 // enabled (the table is locked and the table state is set). 2352 // Note: if the procedure throws exception, we will catch it and rethrow. 2353 // 2354 // We need to wait for the procedure to potentially fail due to "prepare" sanity 2355 // checks. This will block only the beginning of the procedure. See HBASE-19953. 2356 final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createBlockingLatch(); 2357 submitProcedure(new DisableTableProcedure(procedureExecutor.getEnvironment(), 2358 tableName, false, prepareLatch)); 2359 prepareLatch.await(); 2360 2361 getMaster().getMasterCoprocessorHost().postDisableTable(tableName); 2362 } 2363 2364 @Override 2365 protected String getDescription() { 2366 return "DisableTableProcedure"; 2367 } 2368 }); 2369 } 2370 2371 /** 2372 * Return the region and current deployment for the region containing 2373 * the given row. If the region cannot be found, returns null. If it 2374 * is found, but not currently deployed, the second element of the pair 2375 * may be null. 2376 */ 2377 @VisibleForTesting // Used by TestMaster. 2378 Pair<RegionInfo, ServerName> getTableRegionForRow( 2379 final TableName tableName, final byte [] rowKey) 2380 throws IOException { 2381 final AtomicReference<Pair<RegionInfo, ServerName>> result = new AtomicReference<>(null); 2382 2383 MetaTableAccessor.Visitor visitor = new MetaTableAccessor.Visitor() { 2384 @Override 2385 public boolean visit(Result data) throws IOException { 2386 if (data == null || data.size() <= 0) { 2387 return true; 2388 } 2389 Pair<RegionInfo, ServerName> pair = 2390 new Pair(MetaTableAccessor.getRegionInfo(data), 2391 MetaTableAccessor.getServerName(data,0)); 2392 if (pair == null) { 2393 return false; 2394 } 2395 if (!pair.getFirst().getTable().equals(tableName)) { 2396 return false; 2397 } 2398 result.set(pair); 2399 return true; 2400 } 2401 }; 2402 2403 MetaTableAccessor.scanMeta(clusterConnection, visitor, tableName, rowKey, 1); 2404 return result.get(); 2405 } 2406 2407 private long modifyTable(final TableName tableName, 2408 final TableDescriptorGetter newDescriptorGetter, final long nonceGroup, final long nonce) 2409 throws IOException { 2410 return MasterProcedureUtil 2411 .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 2412 @Override 2413 protected void run() throws IOException { 2414 TableDescriptor newDescriptor = newDescriptorGetter.get(); 2415 sanityCheckTableDescriptor(newDescriptor); 2416 TableDescriptor oldDescriptor = getMaster().getTableDescriptors().get(tableName); 2417 getMaster().getMasterCoprocessorHost().preModifyTable(tableName, oldDescriptor, 2418 newDescriptor); 2419 2420 LOG.info(getClientIdAuditPrefix() + " modify " + tableName); 2421 2422 // Execute the operation synchronously - wait for the operation completes before 2423 // continuing. 2424 // 2425 // We need to wait for the procedure to potentially fail due to "prepare" sanity 2426 // checks. This will block only the beginning of the procedure. See HBASE-19953. 2427 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch(); 2428 submitProcedure( 2429 new ModifyTableProcedure(procedureExecutor.getEnvironment(), newDescriptor, latch)); 2430 latch.await(); 2431 2432 getMaster().getMasterCoprocessorHost().postModifyTable(tableName, oldDescriptor, 2433 newDescriptor); 2434 } 2435 2436 @Override 2437 protected String getDescription() { 2438 return "ModifyTableProcedure"; 2439 } 2440 }); 2441 2442 } 2443 2444 @Override 2445 public long modifyTable(final TableName tableName, final TableDescriptor newDescriptor, 2446 final long nonceGroup, final long nonce) throws IOException { 2447 checkInitialized(); 2448 return modifyTable(tableName, new TableDescriptorGetter() { 2449 @Override 2450 public TableDescriptor get() throws IOException { 2451 return newDescriptor; 2452 } 2453 }, nonceGroup, nonce); 2454 2455 } 2456 2457 public long restoreSnapshot(final SnapshotDescription snapshotDesc, 2458 final long nonceGroup, final long nonce, final boolean restoreAcl) throws IOException { 2459 checkInitialized(); 2460 getSnapshotManager().checkSnapshotSupport(); 2461 2462 // Ensure namespace exists. Will throw exception if non-known NS. 2463 final TableName dstTable = TableName.valueOf(snapshotDesc.getTable()); 2464 getClusterSchema().getNamespace(dstTable.getNamespaceAsString()); 2465 2466 return MasterProcedureUtil.submitProcedure( 2467 new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) { 2468 @Override 2469 protected void run() throws IOException { 2470 setProcId( 2471 getSnapshotManager().restoreOrCloneSnapshot(snapshotDesc, getNonceKey(), restoreAcl)); 2472 } 2473 2474 @Override 2475 protected String getDescription() { 2476 return "RestoreSnapshotProcedure"; 2477 } 2478 }); 2479 } 2480 2481 private void checkTableExists(final TableName tableName) 2482 throws IOException, TableNotFoundException { 2483 if (!MetaTableAccessor.tableExists(getConnection(), tableName)) { 2484 throw new TableNotFoundException(tableName); 2485 } 2486 } 2487 2488 @Override 2489 public void checkTableModifiable(final TableName tableName) 2490 throws IOException, TableNotFoundException, TableNotDisabledException { 2491 if (isCatalogTable(tableName)) { 2492 throw new IOException("Can't modify catalog tables"); 2493 } 2494 checkTableExists(tableName); 2495 TableState ts = getTableStateManager().getTableState(tableName); 2496 if (!ts.isDisabled()) { 2497 throw new TableNotDisabledException("Not DISABLED; " + ts); 2498 } 2499 } 2500 2501 public ClusterMetrics getClusterMetricsWithoutCoprocessor() throws InterruptedIOException { 2502 return getClusterMetricsWithoutCoprocessor(EnumSet.allOf(Option.class)); 2503 } 2504 2505 public ClusterMetrics getClusterMetricsWithoutCoprocessor(EnumSet<Option> options) 2506 throws InterruptedIOException { 2507 ClusterMetricsBuilder builder = ClusterMetricsBuilder.newBuilder(); 2508 // given that hbase1 can't submit the request with Option, 2509 // we return all information to client if the list of Option is empty. 2510 if (options.isEmpty()) { 2511 options = EnumSet.allOf(Option.class); 2512 } 2513 2514 for (Option opt : options) { 2515 switch (opt) { 2516 case HBASE_VERSION: builder.setHBaseVersion(VersionInfo.getVersion()); break; 2517 case CLUSTER_ID: builder.setClusterId(getClusterId()); break; 2518 case MASTER: builder.setMasterName(getServerName()); break; 2519 case BACKUP_MASTERS: builder.setBackerMasterNames(getBackupMasters()); break; 2520 case LIVE_SERVERS: { 2521 if (serverManager != null) { 2522 builder.setLiveServerMetrics(serverManager.getOnlineServers().entrySet().stream() 2523 .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue()))); 2524 } 2525 break; 2526 } 2527 case DEAD_SERVERS: { 2528 if (serverManager != null) { 2529 builder.setDeadServerNames(new ArrayList<>( 2530 serverManager.getDeadServers().copyServerNames())); 2531 } 2532 break; 2533 } 2534 case MASTER_COPROCESSORS: { 2535 if (cpHost != null) { 2536 builder.setMasterCoprocessorNames(Arrays.asList(getMasterCoprocessors())); 2537 } 2538 break; 2539 } 2540 case REGIONS_IN_TRANSITION: { 2541 if (assignmentManager != null) { 2542 builder.setRegionsInTransition(assignmentManager.getRegionStates() 2543 .getRegionsStateInTransition()); 2544 } 2545 break; 2546 } 2547 case BALANCER_ON: { 2548 if (loadBalancerTracker != null) { 2549 builder.setBalancerOn(loadBalancerTracker.isBalancerOn()); 2550 } 2551 break; 2552 } 2553 case MASTER_INFO_PORT: { 2554 if (infoServer != null) { 2555 builder.setMasterInfoPort(infoServer.getPort()); 2556 } 2557 break; 2558 } 2559 } 2560 } 2561 return builder.build(); 2562 } 2563 2564 /** 2565 * @return cluster status 2566 */ 2567 public ClusterMetrics getClusterMetrics() throws IOException { 2568 return getClusterMetrics(EnumSet.allOf(Option.class)); 2569 } 2570 2571 public ClusterMetrics getClusterMetrics(EnumSet<Option> options) throws IOException { 2572 if (cpHost != null) { 2573 cpHost.preGetClusterMetrics(); 2574 } 2575 ClusterMetrics status = getClusterMetricsWithoutCoprocessor(options); 2576 if (cpHost != null) { 2577 cpHost.postGetClusterMetrics(status); 2578 } 2579 return status; 2580 } 2581 2582 private List<ServerName> getBackupMasters() throws InterruptedIOException { 2583 // Build Set of backup masters from ZK nodes 2584 List<String> backupMasterStrings; 2585 try { 2586 backupMasterStrings = ZKUtil.listChildrenNoWatch(this.zooKeeper, 2587 this.zooKeeper.znodePaths.backupMasterAddressesZNode); 2588 } catch (KeeperException e) { 2589 LOG.warn(this.zooKeeper.prefix("Unable to list backup servers"), e); 2590 backupMasterStrings = null; 2591 } 2592 2593 List<ServerName> backupMasters = Collections.emptyList(); 2594 if (backupMasterStrings != null && !backupMasterStrings.isEmpty()) { 2595 backupMasters = new ArrayList<>(backupMasterStrings.size()); 2596 for (String s: backupMasterStrings) { 2597 try { 2598 byte [] bytes; 2599 try { 2600 bytes = ZKUtil.getData(this.zooKeeper, ZNodePaths.joinZNode( 2601 this.zooKeeper.znodePaths.backupMasterAddressesZNode, s)); 2602 } catch (InterruptedException e) { 2603 throw new InterruptedIOException(); 2604 } 2605 if (bytes != null) { 2606 ServerName sn; 2607 try { 2608 sn = ProtobufUtil.parseServerNameFrom(bytes); 2609 } catch (DeserializationException e) { 2610 LOG.warn("Failed parse, skipping registering backup server", e); 2611 continue; 2612 } 2613 backupMasters.add(sn); 2614 } 2615 } catch (KeeperException e) { 2616 LOG.warn(this.zooKeeper.prefix("Unable to get information about " + 2617 "backup servers"), e); 2618 } 2619 } 2620 Collections.sort(backupMasters, new Comparator<ServerName>() { 2621 @Override 2622 public int compare(ServerName s1, ServerName s2) { 2623 return s1.getServerName().compareTo(s2.getServerName()); 2624 }}); 2625 } 2626 return backupMasters; 2627 } 2628 2629 /** 2630 * The set of loaded coprocessors is stored in a static set. Since it's 2631 * statically allocated, it does not require that HMaster's cpHost be 2632 * initialized prior to accessing it. 2633 * @return a String representation of the set of names of the loaded coprocessors. 2634 */ 2635 public static String getLoadedCoprocessors() { 2636 return CoprocessorHost.getLoadedCoprocessors().toString(); 2637 } 2638 2639 /** 2640 * @return timestamp in millis when HMaster was started. 2641 */ 2642 public long getMasterStartTime() { 2643 return startcode; 2644 } 2645 2646 /** 2647 * @return timestamp in millis when HMaster became the active master. 2648 */ 2649 public long getMasterActiveTime() { 2650 return masterActiveTime; 2651 } 2652 2653 /** 2654 * @return timestamp in millis when HMaster finished becoming the active master 2655 */ 2656 public long getMasterFinishedInitializationTime() { 2657 return masterFinishedInitializationTime; 2658 } 2659 2660 public int getNumWALFiles() { 2661 return procedureStore != null ? procedureStore.getActiveLogs().size() : 0; 2662 } 2663 2664 public WALProcedureStore getWalProcedureStore() { 2665 return procedureStore; 2666 } 2667 2668 public int getRegionServerInfoPort(final ServerName sn) { 2669 int port = this.serverManager.getInfoPort(sn); 2670 return port == 0 ? conf.getInt(HConstants.REGIONSERVER_INFO_PORT, 2671 HConstants.DEFAULT_REGIONSERVER_INFOPORT) : port; 2672 } 2673 2674 @Override 2675 public String getRegionServerVersion(final ServerName sn) { 2676 // Will return 0 if the server is not online to prevent move system region to unknown version 2677 // RS. 2678 int versionNumber = this.serverManager.getServerVersion(sn); 2679 return VersionInfoUtil.versionNumberToString(versionNumber); 2680 } 2681 2682 @Override 2683 public void checkIfShouldMoveSystemRegionAsync() { 2684 assignmentManager.checkIfShouldMoveSystemRegionAsync(); 2685 } 2686 2687 /** 2688 * @return array of coprocessor SimpleNames. 2689 */ 2690 public String[] getMasterCoprocessors() { 2691 Set<String> masterCoprocessors = getMasterCoprocessorHost().getCoprocessors(); 2692 return masterCoprocessors.toArray(new String[masterCoprocessors.size()]); 2693 } 2694 2695 @Override 2696 public void abort(String reason, Throwable cause) { 2697 if (isAborted() || isStopped()) { 2698 return; 2699 } 2700 if (cpHost != null) { 2701 // HBASE-4014: dump a list of loaded coprocessors. 2702 LOG.error(HBaseMarkers.FATAL, "Master server abort: loaded coprocessors are: " + 2703 getLoadedCoprocessors()); 2704 } 2705 String msg = "***** ABORTING master " + this + ": " + reason + " *****"; 2706 if (cause != null) { 2707 LOG.error(HBaseMarkers.FATAL, msg, cause); 2708 } else { 2709 LOG.error(HBaseMarkers.FATAL, msg); 2710 } 2711 2712 try { 2713 stopMaster(); 2714 } catch (IOException e) { 2715 LOG.error("Exception occurred while stopping master", e); 2716 } 2717 } 2718 2719 @Override 2720 public ZKWatcher getZooKeeper() { 2721 return zooKeeper; 2722 } 2723 2724 @Override 2725 public MasterCoprocessorHost getMasterCoprocessorHost() { 2726 return cpHost; 2727 } 2728 2729 @Override 2730 public MasterQuotaManager getMasterQuotaManager() { 2731 return quotaManager; 2732 } 2733 2734 @Override 2735 public ProcedureExecutor<MasterProcedureEnv> getMasterProcedureExecutor() { 2736 return procedureExecutor; 2737 } 2738 2739 @Override 2740 public ServerName getServerName() { 2741 return this.serverName; 2742 } 2743 2744 @Override 2745 public AssignmentManager getAssignmentManager() { 2746 return this.assignmentManager; 2747 } 2748 2749 @Override 2750 public CatalogJanitor getCatalogJanitor() { 2751 return this.catalogJanitorChore; 2752 } 2753 2754 public MemoryBoundedLogMessageBuffer getRegionServerFatalLogBuffer() { 2755 return rsFatals; 2756 } 2757 2758 /** 2759 * Shutdown the cluster. 2760 * Master runs a coordinated stop of all RegionServers and then itself. 2761 */ 2762 public void shutdown() throws IOException { 2763 if (cpHost != null) { 2764 cpHost.preShutdown(); 2765 } 2766 // Tell the servermanager cluster shutdown has been called. This makes it so when Master is 2767 // last running server, it'll stop itself. Next, we broadcast the cluster shutdown by setting 2768 // the cluster status as down. RegionServers will notice this change in state and will start 2769 // shutting themselves down. When last has exited, Master can go down. 2770 if (this.serverManager != null) { 2771 this.serverManager.shutdownCluster(); 2772 } 2773 if (this.clusterStatusTracker != null) { 2774 try { 2775 this.clusterStatusTracker.setClusterDown(); 2776 } catch (KeeperException e) { 2777 LOG.error("ZooKeeper exception trying to set cluster as down in ZK", e); 2778 } 2779 } 2780 // Stop the procedure executor. Will stop any ongoing assign, unassign, server crash etc., 2781 // processing so we can go down. 2782 if (this.procedureExecutor != null) { 2783 this.procedureExecutor.stop(); 2784 } 2785 // Shutdown our cluster connection. This will kill any hosted RPCs that might be going on; 2786 // this is what we want especially if the Master is in startup phase doing call outs to 2787 // hbase:meta, etc. when cluster is down. Without ths connection close, we'd have to wait on 2788 // the rpc to timeout. 2789 if (this.clusterConnection != null) { 2790 this.clusterConnection.close(); 2791 } 2792 } 2793 2794 public void stopMaster() throws IOException { 2795 if (cpHost != null) { 2796 cpHost.preStopMaster(); 2797 } 2798 stop("Stopped by " + Thread.currentThread().getName()); 2799 } 2800 2801 @Override 2802 public void stop(String msg) { 2803 if (!isStopped()) { 2804 super.stop(msg); 2805 if (this.activeMasterManager != null) { 2806 this.activeMasterManager.stop(); 2807 } 2808 } 2809 } 2810 2811 void checkServiceStarted() throws ServerNotRunningYetException { 2812 if (!serviceStarted) { 2813 throw new ServerNotRunningYetException("Server is not running yet"); 2814 } 2815 } 2816 2817 public static class MasterStoppedException extends DoNotRetryIOException { 2818 MasterStoppedException() { 2819 super(); 2820 } 2821 } 2822 2823 void checkInitialized() throws PleaseHoldException, ServerNotRunningYetException, 2824 MasterNotRunningException, MasterStoppedException { 2825 checkServiceStarted(); 2826 if (!isInitialized()) { 2827 throw new PleaseHoldException("Master is initializing"); 2828 } 2829 if (isStopped()) { 2830 throw new MasterStoppedException(); 2831 } 2832 } 2833 2834 /** 2835 * Report whether this master is currently the active master or not. 2836 * If not active master, we are parked on ZK waiting to become active. 2837 * 2838 * This method is used for testing. 2839 * 2840 * @return true if active master, false if not. 2841 */ 2842 @Override 2843 public boolean isActiveMaster() { 2844 return activeMaster; 2845 } 2846 2847 /** 2848 * Report whether this master has completed with its initialization and is 2849 * ready. If ready, the master is also the active master. A standby master 2850 * is never ready. 2851 * 2852 * This method is used for testing. 2853 * 2854 * @return true if master is ready to go, false if not. 2855 */ 2856 @Override 2857 public boolean isInitialized() { 2858 return initialized.isReady(); 2859 } 2860 2861 /** 2862 * Report whether this master is in maintenance mode. 2863 * 2864 * @return true if master is in maintenanceMode 2865 */ 2866 @Override 2867 public boolean isInMaintenanceMode() throws IOException { 2868 if (!isInitialized()) { 2869 throw new PleaseHoldException("Master is initializing"); 2870 } 2871 return maintenanceModeTracker.isInMaintenanceMode(); 2872 } 2873 2874 @VisibleForTesting 2875 public void setInitialized(boolean isInitialized) { 2876 procedureExecutor.getEnvironment().setEventReady(initialized, isInitialized); 2877 } 2878 2879 @Override 2880 public ProcedureEvent getInitializedEvent() { 2881 return initialized; 2882 } 2883 2884 /** 2885 * Compute the average load across all region servers. 2886 * Currently, this uses a very naive computation - just uses the number of 2887 * regions being served, ignoring stats about number of requests. 2888 * @return the average load 2889 */ 2890 public double getAverageLoad() { 2891 if (this.assignmentManager == null) { 2892 return 0; 2893 } 2894 2895 RegionStates regionStates = this.assignmentManager.getRegionStates(); 2896 if (regionStates == null) { 2897 return 0; 2898 } 2899 return regionStates.getAverageLoad(); 2900 } 2901 2902 /* 2903 * @return the count of region split plans executed 2904 */ 2905 public long getSplitPlanCount() { 2906 return splitPlanCount; 2907 } 2908 2909 /* 2910 * @return the count of region merge plans executed 2911 */ 2912 public long getMergePlanCount() { 2913 return mergePlanCount; 2914 } 2915 2916 @Override 2917 public boolean registerService(Service instance) { 2918 /* 2919 * No stacking of instances is allowed for a single service name 2920 */ 2921 Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType(); 2922 String serviceName = CoprocessorRpcUtils.getServiceName(serviceDesc); 2923 if (coprocessorServiceHandlers.containsKey(serviceName)) { 2924 LOG.error("Coprocessor service "+serviceName+ 2925 " already registered, rejecting request from "+instance 2926 ); 2927 return false; 2928 } 2929 2930 coprocessorServiceHandlers.put(serviceName, instance); 2931 if (LOG.isDebugEnabled()) { 2932 LOG.debug("Registered master coprocessor service: service="+serviceName); 2933 } 2934 return true; 2935 } 2936 2937 /** 2938 * Utility for constructing an instance of the passed HMaster class. 2939 * @param masterClass 2940 * @return HMaster instance. 2941 */ 2942 public static HMaster constructMaster(Class<? extends HMaster> masterClass, 2943 final Configuration conf) { 2944 try { 2945 Constructor<? extends HMaster> c = masterClass.getConstructor(Configuration.class); 2946 return c.newInstance(conf); 2947 } catch(Exception e) { 2948 Throwable error = e; 2949 if (e instanceof InvocationTargetException && 2950 ((InvocationTargetException)e).getTargetException() != null) { 2951 error = ((InvocationTargetException)e).getTargetException(); 2952 } 2953 throw new RuntimeException("Failed construction of Master: " + masterClass.toString() + ". " 2954 , error); 2955 } 2956 } 2957 2958 /** 2959 * @see org.apache.hadoop.hbase.master.HMasterCommandLine 2960 */ 2961 public static void main(String [] args) { 2962 LOG.info("STARTING service " + HMaster.class.getSimpleName()); 2963 VersionInfo.logVersion(); 2964 new HMasterCommandLine(HMaster.class).doMain(args); 2965 } 2966 2967 public HFileCleaner getHFileCleaner() { 2968 return this.hfileCleaner; 2969 } 2970 2971 public LogCleaner getLogCleaner() { 2972 return this.logCleaner; 2973 } 2974 2975 /** 2976 * @return the underlying snapshot manager 2977 */ 2978 @Override 2979 public SnapshotManager getSnapshotManager() { 2980 return this.snapshotManager; 2981 } 2982 2983 /** 2984 * @return the underlying MasterProcedureManagerHost 2985 */ 2986 @Override 2987 public MasterProcedureManagerHost getMasterProcedureManagerHost() { 2988 return mpmHost; 2989 } 2990 2991 @Override 2992 public ClusterSchema getClusterSchema() { 2993 return this.clusterSchemaService; 2994 } 2995 2996 /** 2997 * Create a new Namespace. 2998 * @param namespaceDescriptor descriptor for new Namespace 2999 * @param nonceGroup Identifier for the source of the request, a client or process. 3000 * @param nonce A unique identifier for this operation from the client or process identified by 3001 * <code>nonceGroup</code> (the source must ensure each operation gets a unique id). 3002 * @return procedure id 3003 */ 3004 long createNamespace(final NamespaceDescriptor namespaceDescriptor, final long nonceGroup, 3005 final long nonce) throws IOException { 3006 checkInitialized(); 3007 3008 TableName.isLegalNamespaceName(Bytes.toBytes(namespaceDescriptor.getName())); 3009 3010 return MasterProcedureUtil.submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, 3011 nonceGroup, nonce) { 3012 @Override 3013 protected void run() throws IOException { 3014 getMaster().getMasterCoprocessorHost().preCreateNamespace(namespaceDescriptor); 3015 // We need to wait for the procedure to potentially fail due to "prepare" sanity 3016 // checks. This will block only the beginning of the procedure. See HBASE-19953. 3017 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch(); 3018 LOG.info(getClientIdAuditPrefix() + " creating " + namespaceDescriptor); 3019 // Execute the operation synchronously - wait for the operation to complete before 3020 // continuing. 3021 setProcId(getClusterSchema().createNamespace(namespaceDescriptor, getNonceKey(), latch)); 3022 latch.await(); 3023 getMaster().getMasterCoprocessorHost().postCreateNamespace(namespaceDescriptor); 3024 } 3025 3026 @Override 3027 protected String getDescription() { 3028 return "CreateNamespaceProcedure"; 3029 } 3030 }); 3031 } 3032 3033 /** 3034 * Modify an existing Namespace. 3035 * @param nonceGroup Identifier for the source of the request, a client or process. 3036 * @param nonce A unique identifier for this operation from the client or process identified by 3037 * <code>nonceGroup</code> (the source must ensure each operation gets a unique id). 3038 * @return procedure id 3039 */ 3040 long modifyNamespace(final NamespaceDescriptor newNsDescriptor, final long nonceGroup, 3041 final long nonce) throws IOException { 3042 checkInitialized(); 3043 3044 TableName.isLegalNamespaceName(Bytes.toBytes(newNsDescriptor.getName())); 3045 3046 return MasterProcedureUtil.submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, 3047 nonceGroup, nonce) { 3048 @Override 3049 protected void run() throws IOException { 3050 NamespaceDescriptor oldNsDescriptor = getNamespace(newNsDescriptor.getName()); 3051 getMaster().getMasterCoprocessorHost().preModifyNamespace(oldNsDescriptor, newNsDescriptor); 3052 // We need to wait for the procedure to potentially fail due to "prepare" sanity 3053 // checks. This will block only the beginning of the procedure. See HBASE-19953. 3054 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch(); 3055 LOG.info(getClientIdAuditPrefix() + " modify " + newNsDescriptor); 3056 // Execute the operation synchronously - wait for the operation to complete before 3057 // continuing. 3058 setProcId(getClusterSchema().modifyNamespace(newNsDescriptor, getNonceKey(), latch)); 3059 latch.await(); 3060 getMaster().getMasterCoprocessorHost().postModifyNamespace(oldNsDescriptor, 3061 newNsDescriptor); 3062 } 3063 3064 @Override 3065 protected String getDescription() { 3066 return "ModifyNamespaceProcedure"; 3067 } 3068 }); 3069 } 3070 3071 /** 3072 * Delete an existing Namespace. Only empty Namespaces (no tables) can be removed. 3073 * @param nonceGroup Identifier for the source of the request, a client or process. 3074 * @param nonce A unique identifier for this operation from the client or process identified by 3075 * <code>nonceGroup</code> (the source must ensure each operation gets a unique id). 3076 * @return procedure id 3077 */ 3078 long deleteNamespace(final String name, final long nonceGroup, final long nonce) 3079 throws IOException { 3080 checkInitialized(); 3081 3082 return MasterProcedureUtil.submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, 3083 nonceGroup, nonce) { 3084 @Override 3085 protected void run() throws IOException { 3086 getMaster().getMasterCoprocessorHost().preDeleteNamespace(name); 3087 LOG.info(getClientIdAuditPrefix() + " delete " + name); 3088 // Execute the operation synchronously - wait for the operation to complete before 3089 // continuing. 3090 // 3091 // We need to wait for the procedure to potentially fail due to "prepare" sanity 3092 // checks. This will block only the beginning of the procedure. See HBASE-19953. 3093 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch(); 3094 setProcId(submitProcedure( 3095 new DeleteNamespaceProcedure(procedureExecutor.getEnvironment(), name, latch))); 3096 latch.await(); 3097 // Will not be invoked in the face of Exception thrown by the Procedure's execution 3098 getMaster().getMasterCoprocessorHost().postDeleteNamespace(name); 3099 } 3100 3101 @Override 3102 protected String getDescription() { 3103 return "DeleteNamespaceProcedure"; 3104 } 3105 }); 3106 } 3107 3108 /** 3109 * Get a Namespace 3110 * @param name Name of the Namespace 3111 * @return Namespace descriptor for <code>name</code> 3112 */ 3113 NamespaceDescriptor getNamespace(String name) throws IOException { 3114 checkInitialized(); 3115 if (this.cpHost != null) this.cpHost.preGetNamespaceDescriptor(name); 3116 NamespaceDescriptor nsd = this.clusterSchemaService.getNamespace(name); 3117 if (this.cpHost != null) this.cpHost.postGetNamespaceDescriptor(nsd); 3118 return nsd; 3119 } 3120 3121 /** 3122 * Get all Namespaces 3123 * @return All Namespace descriptors 3124 */ 3125 List<NamespaceDescriptor> getNamespaces() throws IOException { 3126 checkInitialized(); 3127 final List<NamespaceDescriptor> nsds = new ArrayList<>(); 3128 if (cpHost != null) { 3129 cpHost.preListNamespaceDescriptors(nsds); 3130 } 3131 nsds.addAll(this.clusterSchemaService.getNamespaces()); 3132 if (this.cpHost != null) { 3133 this.cpHost.postListNamespaceDescriptors(nsds); 3134 } 3135 return nsds; 3136 } 3137 3138 @Override 3139 public List<TableName> listTableNamesByNamespace(String name) throws IOException { 3140 checkInitialized(); 3141 return listTableNames(name, null, true); 3142 } 3143 3144 @Override 3145 public List<TableDescriptor> listTableDescriptorsByNamespace(String name) throws IOException { 3146 checkInitialized(); 3147 return listTableDescriptors(name, null, null, true); 3148 } 3149 3150 @Override 3151 public boolean abortProcedure(final long procId, final boolean mayInterruptIfRunning) 3152 throws IOException { 3153 if (cpHost != null) { 3154 cpHost.preAbortProcedure(this.procedureExecutor, procId); 3155 } 3156 3157 final boolean result = this.procedureExecutor.abort(procId, mayInterruptIfRunning); 3158 3159 if (cpHost != null) { 3160 cpHost.postAbortProcedure(); 3161 } 3162 3163 return result; 3164 } 3165 3166 @Override 3167 public List<Procedure<?>> getProcedures() throws IOException { 3168 if (cpHost != null) { 3169 cpHost.preGetProcedures(); 3170 } 3171 3172 @SuppressWarnings({ "unchecked", "rawtypes" }) 3173 List<Procedure<?>> procList = (List) this.procedureExecutor.getProcedures(); 3174 3175 if (cpHost != null) { 3176 cpHost.postGetProcedures(procList); 3177 } 3178 3179 return procList; 3180 } 3181 3182 @Override 3183 public List<LockedResource> getLocks() throws IOException { 3184 if (cpHost != null) { 3185 cpHost.preGetLocks(); 3186 } 3187 3188 MasterProcedureScheduler procedureScheduler = 3189 procedureExecutor.getEnvironment().getProcedureScheduler(); 3190 3191 final List<LockedResource> lockedResources = procedureScheduler.getLocks(); 3192 3193 if (cpHost != null) { 3194 cpHost.postGetLocks(lockedResources); 3195 } 3196 3197 return lockedResources; 3198 } 3199 3200 /** 3201 * Returns the list of table descriptors that match the specified request 3202 * @param namespace the namespace to query, or null if querying for all 3203 * @param regex The regular expression to match against, or null if querying for all 3204 * @param tableNameList the list of table names, or null if querying for all 3205 * @param includeSysTables False to match only against userspace tables 3206 * @return the list of table descriptors 3207 */ 3208 public List<TableDescriptor> listTableDescriptors(final String namespace, final String regex, 3209 final List<TableName> tableNameList, final boolean includeSysTables) 3210 throws IOException { 3211 List<TableDescriptor> htds = new ArrayList<>(); 3212 if (cpHost != null) { 3213 cpHost.preGetTableDescriptors(tableNameList, htds, regex); 3214 } 3215 htds = getTableDescriptors(htds, namespace, regex, tableNameList, includeSysTables); 3216 if (cpHost != null) { 3217 cpHost.postGetTableDescriptors(tableNameList, htds, regex); 3218 } 3219 return htds; 3220 } 3221 3222 /** 3223 * Returns the list of table names that match the specified request 3224 * @param regex The regular expression to match against, or null if querying for all 3225 * @param namespace the namespace to query, or null if querying for all 3226 * @param includeSysTables False to match only against userspace tables 3227 * @return the list of table names 3228 */ 3229 public List<TableName> listTableNames(final String namespace, final String regex, 3230 final boolean includeSysTables) throws IOException { 3231 List<TableDescriptor> htds = new ArrayList<>(); 3232 if (cpHost != null) { 3233 cpHost.preGetTableNames(htds, regex); 3234 } 3235 htds = getTableDescriptors(htds, namespace, regex, null, includeSysTables); 3236 if (cpHost != null) { 3237 cpHost.postGetTableNames(htds, regex); 3238 } 3239 List<TableName> result = new ArrayList<>(htds.size()); 3240 for (TableDescriptor htd: htds) result.add(htd.getTableName()); 3241 return result; 3242 } 3243 3244 /** 3245 * @return list of table table descriptors after filtering by regex and whether to include system 3246 * tables, etc. 3247 * @throws IOException 3248 */ 3249 private List<TableDescriptor> getTableDescriptors(final List<TableDescriptor> htds, 3250 final String namespace, final String regex, final List<TableName> tableNameList, 3251 final boolean includeSysTables) 3252 throws IOException { 3253 if (tableNameList == null || tableNameList.isEmpty()) { 3254 // request for all TableDescriptors 3255 Collection<TableDescriptor> allHtds; 3256 if (namespace != null && namespace.length() > 0) { 3257 // Do a check on the namespace existence. Will fail if does not exist. 3258 this.clusterSchemaService.getNamespace(namespace); 3259 allHtds = tableDescriptors.getByNamespace(namespace).values(); 3260 } else { 3261 allHtds = tableDescriptors.getAll().values(); 3262 } 3263 for (TableDescriptor desc: allHtds) { 3264 if (tableStateManager.isTablePresent(desc.getTableName()) 3265 && (includeSysTables || !desc.getTableName().isSystemTable())) { 3266 htds.add(desc); 3267 } 3268 } 3269 } else { 3270 for (TableName s: tableNameList) { 3271 if (tableStateManager.isTablePresent(s)) { 3272 TableDescriptor desc = tableDescriptors.get(s); 3273 if (desc != null) { 3274 htds.add(desc); 3275 } 3276 } 3277 } 3278 } 3279 3280 // Retains only those matched by regular expression. 3281 if (regex != null) filterTablesByRegex(htds, Pattern.compile(regex)); 3282 return htds; 3283 } 3284 3285 /** 3286 * Removes the table descriptors that don't match the pattern. 3287 * @param descriptors list of table descriptors to filter 3288 * @param pattern the regex to use 3289 */ 3290 private static void filterTablesByRegex(final Collection<TableDescriptor> descriptors, 3291 final Pattern pattern) { 3292 final String defaultNS = NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR; 3293 Iterator<TableDescriptor> itr = descriptors.iterator(); 3294 while (itr.hasNext()) { 3295 TableDescriptor htd = itr.next(); 3296 String tableName = htd.getTableName().getNameAsString(); 3297 boolean matched = pattern.matcher(tableName).matches(); 3298 if (!matched && htd.getTableName().getNamespaceAsString().equals(defaultNS)) { 3299 matched = pattern.matcher(defaultNS + TableName.NAMESPACE_DELIM + tableName).matches(); 3300 } 3301 if (!matched) { 3302 itr.remove(); 3303 } 3304 } 3305 } 3306 3307 @Override 3308 public long getLastMajorCompactionTimestamp(TableName table) throws IOException { 3309 return getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS)) 3310 .getLastMajorCompactionTimestamp(table); 3311 } 3312 3313 @Override 3314 public long getLastMajorCompactionTimestampForRegion(byte[] regionName) throws IOException { 3315 return getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS)) 3316 .getLastMajorCompactionTimestamp(regionName); 3317 } 3318 3319 /** 3320 * Gets the mob file compaction state for a specific table. 3321 * Whether all the mob files are selected is known during the compaction execution, but 3322 * the statistic is done just before compaction starts, it is hard to know the compaction 3323 * type at that time, so the rough statistics are chosen for the mob file compaction. Only two 3324 * compaction states are available, CompactionState.MAJOR_AND_MINOR and CompactionState.NONE. 3325 * @param tableName The current table name. 3326 * @return If a given table is in mob file compaction now. 3327 */ 3328 public CompactionState getMobCompactionState(TableName tableName) { 3329 AtomicInteger compactionsCount = mobCompactionStates.get(tableName); 3330 if (compactionsCount != null && compactionsCount.get() != 0) { 3331 return CompactionState.MAJOR_AND_MINOR; 3332 } 3333 return CompactionState.NONE; 3334 } 3335 3336 public void reportMobCompactionStart(TableName tableName) throws IOException { 3337 IdLock.Entry lockEntry = null; 3338 try { 3339 lockEntry = mobCompactionLock.getLockEntry(tableName.hashCode()); 3340 AtomicInteger compactionsCount = mobCompactionStates.get(tableName); 3341 if (compactionsCount == null) { 3342 compactionsCount = new AtomicInteger(0); 3343 mobCompactionStates.put(tableName, compactionsCount); 3344 } 3345 compactionsCount.incrementAndGet(); 3346 } finally { 3347 if (lockEntry != null) { 3348 mobCompactionLock.releaseLockEntry(lockEntry); 3349 } 3350 } 3351 } 3352 3353 public void reportMobCompactionEnd(TableName tableName) throws IOException { 3354 IdLock.Entry lockEntry = null; 3355 try { 3356 lockEntry = mobCompactionLock.getLockEntry(tableName.hashCode()); 3357 AtomicInteger compactionsCount = mobCompactionStates.get(tableName); 3358 if (compactionsCount != null) { 3359 int count = compactionsCount.decrementAndGet(); 3360 // remove the entry if the count is 0. 3361 if (count == 0) { 3362 mobCompactionStates.remove(tableName); 3363 } 3364 } 3365 } finally { 3366 if (lockEntry != null) { 3367 mobCompactionLock.releaseLockEntry(lockEntry); 3368 } 3369 } 3370 } 3371 3372 /** 3373 * Requests mob compaction. 3374 * @param tableName The table the compact. 3375 * @param columns The compacted columns. 3376 * @param allFiles Whether add all mob files into the compaction. 3377 */ 3378 public void requestMobCompaction(TableName tableName, 3379 List<ColumnFamilyDescriptor> columns, boolean allFiles) throws IOException { 3380 mobCompactThread.requestMobCompaction(conf, fs, tableName, columns, allFiles); 3381 } 3382 3383 /** 3384 * Queries the state of the {@link LoadBalancerTracker}. If the balancer is not initialized, 3385 * false is returned. 3386 * 3387 * @return The state of the load balancer, or false if the load balancer isn't defined. 3388 */ 3389 public boolean isBalancerOn() { 3390 try { 3391 if (null == loadBalancerTracker || isInMaintenanceMode()) { 3392 return false; 3393 } 3394 } catch (IOException e) { 3395 return false; 3396 } 3397 return loadBalancerTracker.isBalancerOn(); 3398 } 3399 3400 /** 3401 * Queries the state of the {@link RegionNormalizerTracker}. If it's not initialized, 3402 * false is returned. 3403 */ 3404 public boolean isNormalizerOn() { 3405 try { 3406 return (null == regionNormalizerTracker || isInMaintenanceMode()) ? 3407 false: regionNormalizerTracker.isNormalizerOn(); 3408 } catch (IOException e) { 3409 return false; 3410 } 3411 } 3412 3413 /** 3414 * Queries the state of the {@link SplitOrMergeTracker}. If it is not initialized, 3415 * false is returned. If switchType is illegal, false will return. 3416 * @param switchType see {@link org.apache.hadoop.hbase.client.MasterSwitchType} 3417 * @return The state of the switch 3418 */ 3419 @Override 3420 public boolean isSplitOrMergeEnabled(MasterSwitchType switchType) { 3421 try { 3422 if (null == splitOrMergeTracker || isInMaintenanceMode()) { 3423 return false; 3424 } 3425 } catch (IOException e) { 3426 return false; 3427 } 3428 return splitOrMergeTracker.isSplitOrMergeEnabled(switchType); 3429 } 3430 3431 /** 3432 * Fetch the configured {@link LoadBalancer} class name. If none is set, a default is returned. 3433 * 3434 * @return The name of the {@link LoadBalancer} in use. 3435 */ 3436 public String getLoadBalancerClassName() { 3437 return conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, LoadBalancerFactory 3438 .getDefaultLoadBalancerClass().getName()); 3439 } 3440 3441 /** 3442 * @return RegionNormalizerTracker instance 3443 */ 3444 public RegionNormalizerTracker getRegionNormalizerTracker() { 3445 return regionNormalizerTracker; 3446 } 3447 3448 public SplitOrMergeTracker getSplitOrMergeTracker() { 3449 return splitOrMergeTracker; 3450 } 3451 3452 @Override 3453 public LoadBalancer getLoadBalancer() { 3454 return balancer; 3455 } 3456 3457 @Override 3458 public FavoredNodesManager getFavoredNodesManager() { 3459 return favoredNodesManager; 3460 } 3461 3462 @Override 3463 public void addReplicationPeer(String peerId, ReplicationPeerConfig peerConfig, boolean enabled) 3464 throws ReplicationException, IOException { 3465 if (cpHost != null) { 3466 cpHost.preAddReplicationPeer(peerId, peerConfig); 3467 } 3468 LOG.info(getClientIdAuditPrefix() + " creating replication peer, id=" + peerId + ", config=" 3469 + peerConfig + ", state=" + (enabled ? "ENABLED" : "DISABLED")); 3470 this.replicationManager.addReplicationPeer(peerId, peerConfig, enabled); 3471 if (cpHost != null) { 3472 cpHost.postAddReplicationPeer(peerId, peerConfig); 3473 } 3474 } 3475 3476 @Override 3477 public void removeReplicationPeer(String peerId) throws ReplicationException, IOException { 3478 if (cpHost != null) { 3479 cpHost.preRemoveReplicationPeer(peerId); 3480 } 3481 LOG.info(getClientIdAuditPrefix() + " removing replication peer, id=" + peerId); 3482 this.replicationManager.removeReplicationPeer(peerId); 3483 if (cpHost != null) { 3484 cpHost.postRemoveReplicationPeer(peerId); 3485 } 3486 } 3487 3488 @Override 3489 public void enableReplicationPeer(String peerId) throws ReplicationException, IOException { 3490 if (cpHost != null) { 3491 cpHost.preEnableReplicationPeer(peerId); 3492 } 3493 LOG.info(getClientIdAuditPrefix() + " enable replication peer, id=" + peerId); 3494 this.replicationManager.enableReplicationPeer(peerId); 3495 if (cpHost != null) { 3496 cpHost.postEnableReplicationPeer(peerId); 3497 } 3498 } 3499 3500 @Override 3501 public void disableReplicationPeer(String peerId) throws ReplicationException, IOException { 3502 if (cpHost != null) { 3503 cpHost.preDisableReplicationPeer(peerId); 3504 } 3505 LOG.info(getClientIdAuditPrefix() + " disable replication peer, id=" + peerId); 3506 this.replicationManager.disableReplicationPeer(peerId); 3507 if (cpHost != null) { 3508 cpHost.postDisableReplicationPeer(peerId); 3509 } 3510 } 3511 3512 @Override 3513 public ReplicationPeerConfig getReplicationPeerConfig(String peerId) throws ReplicationException, 3514 IOException { 3515 if (cpHost != null) { 3516 cpHost.preGetReplicationPeerConfig(peerId); 3517 } 3518 final ReplicationPeerConfig peerConfig = this.replicationManager.getPeerConfig(peerId); 3519 LOG.info(getClientIdAuditPrefix() + " get replication peer config, id=" + peerId + ", config=" 3520 + peerConfig); 3521 if (cpHost != null) { 3522 cpHost.postGetReplicationPeerConfig(peerId); 3523 } 3524 return peerConfig; 3525 } 3526 3527 @Override 3528 public void updateReplicationPeerConfig(String peerId, ReplicationPeerConfig peerConfig) 3529 throws ReplicationException, IOException { 3530 if (cpHost != null) { 3531 cpHost.preUpdateReplicationPeerConfig(peerId, peerConfig); 3532 } 3533 LOG.info(getClientIdAuditPrefix() + " update replication peer config, id=" + peerId 3534 + ", config=" + peerConfig); 3535 this.replicationManager.updatePeerConfig(peerId, peerConfig); 3536 if (cpHost != null) { 3537 cpHost.postUpdateReplicationPeerConfig(peerId, peerConfig); 3538 } 3539 } 3540 3541 @Override 3542 public List<ReplicationPeerDescription> listReplicationPeers(String regex) 3543 throws ReplicationException, IOException { 3544 if (cpHost != null) { 3545 cpHost.preListReplicationPeers(regex); 3546 } 3547 LOG.info(getClientIdAuditPrefix() + " list replication peers, regex=" + regex); 3548 Pattern pattern = regex == null ? null : Pattern.compile(regex); 3549 List<ReplicationPeerDescription> peers = this.replicationManager.listReplicationPeers(pattern); 3550 if (cpHost != null) { 3551 cpHost.postListReplicationPeers(regex); 3552 } 3553 return peers; 3554 } 3555 3556 /** 3557 * Mark region server(s) as decommissioned (previously called 'draining') to prevent additional 3558 * regions from getting assigned to them. Also unload the regions on the servers asynchronously.0 3559 * @param servers Region servers to decommission. 3560 * @throws HBaseIOException 3561 */ 3562 public void decommissionRegionServers(final List<ServerName> servers, final boolean offload) 3563 throws HBaseIOException { 3564 List<ServerName> serversAdded = new ArrayList<>(servers.size()); 3565 // Place the decommission marker first. 3566 String parentZnode = getZooKeeper().znodePaths.drainingZNode; 3567 for (ServerName server : servers) { 3568 try { 3569 String node = ZNodePaths.joinZNode(parentZnode, server.getServerName()); 3570 ZKUtil.createAndFailSilent(getZooKeeper(), node); 3571 } catch (KeeperException ke) { 3572 throw new HBaseIOException( 3573 this.zooKeeper.prefix("Unable to decommission '" + server.getServerName() + "'."), ke); 3574 } 3575 if (this.serverManager.addServerToDrainList(server)) { 3576 serversAdded.add(server); 3577 }; 3578 } 3579 // Move the regions off the decommissioned servers. 3580 if (offload) { 3581 final List<ServerName> destServers = this.serverManager.createDestinationServersList(); 3582 for (ServerName server : serversAdded) { 3583 final List<RegionInfo> regionsOnServer = 3584 this.assignmentManager.getRegionStates().getServerRegionInfoSet(server); 3585 for (RegionInfo hri : regionsOnServer) { 3586 ServerName dest = balancer.randomAssignment(hri, destServers); 3587 if (dest == null) { 3588 throw new HBaseIOException("Unable to determine a plan to move " + hri); 3589 } 3590 RegionPlan rp = new RegionPlan(hri, server, dest); 3591 this.assignmentManager.moveAsync(rp); 3592 } 3593 } 3594 } 3595 } 3596 3597 /** 3598 * List region servers marked as decommissioned (previously called 'draining') to not get regions 3599 * assigned to them. 3600 * @return List of decommissioned servers. 3601 */ 3602 public List<ServerName> listDecommissionedRegionServers() { 3603 return this.serverManager.getDrainingServersList(); 3604 } 3605 3606 /** 3607 * Remove decommission marker (previously called 'draining') from a region server to allow regions 3608 * assignments. Load regions onto the server asynchronously if a list of regions is given 3609 * @param server Region server to remove decommission marker from. 3610 * @throws HBaseIOException 3611 */ 3612 public void recommissionRegionServer(final ServerName server, 3613 final List<byte[]> encodedRegionNames) throws HBaseIOException { 3614 // Remove the server from decommissioned (draining) server list. 3615 String parentZnode = getZooKeeper().znodePaths.drainingZNode; 3616 String node = ZNodePaths.joinZNode(parentZnode, server.getServerName()); 3617 try { 3618 ZKUtil.deleteNodeFailSilent(getZooKeeper(), node); 3619 } catch (KeeperException ke) { 3620 throw new HBaseIOException( 3621 this.zooKeeper.prefix("Unable to recommission '" + server.getServerName() + "'."), ke); 3622 } 3623 this.serverManager.removeServerFromDrainList(server); 3624 3625 // Load the regions onto the server if we are given a list of regions. 3626 if (encodedRegionNames == null || encodedRegionNames.isEmpty()) { 3627 return; 3628 } 3629 if (!this.serverManager.isServerOnline(server)) { 3630 return; 3631 } 3632 for (byte[] encodedRegionName : encodedRegionNames) { 3633 RegionState regionState = 3634 assignmentManager.getRegionStates().getRegionState(Bytes.toString(encodedRegionName)); 3635 if (regionState == null) { 3636 LOG.warn("Unknown region " + Bytes.toStringBinary(encodedRegionName)); 3637 continue; 3638 } 3639 RegionInfo hri = regionState.getRegion(); 3640 if (server.equals(regionState.getServerName())) { 3641 LOG.info("Skipping move of region " + hri.getRegionNameAsString() 3642 + " because region already assigned to the same server " + server + "."); 3643 continue; 3644 } 3645 RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), server); 3646 this.assignmentManager.moveAsync(rp); 3647 } 3648 } 3649 3650 @Override 3651 public LockManager getLockManager() { 3652 return lockManager; 3653 } 3654 3655 public QuotaObserverChore getQuotaObserverChore() { 3656 return this.quotaObserverChore; 3657 } 3658 3659 public SpaceQuotaSnapshotNotifier getSpaceQuotaSnapshotNotifier() { 3660 return this.spaceQuotaSnapshotNotifier; 3661 } 3662 3663 /** 3664 * This method modifies the master's configuration in order to inject replication-related features 3665 */ 3666 @VisibleForTesting 3667 public static void decorateMasterConfiguration(Configuration conf) { 3668 String plugins = conf.get(HBASE_MASTER_LOGCLEANER_PLUGINS); 3669 String cleanerClass = ReplicationLogCleaner.class.getCanonicalName(); 3670 if (!plugins.contains(cleanerClass)) { 3671 conf.set(HBASE_MASTER_LOGCLEANER_PLUGINS, plugins + "," + cleanerClass); 3672 } 3673 if (ReplicationUtils.isReplicationForBulkLoadDataEnabled(conf)) { 3674 plugins = conf.get(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS); 3675 cleanerClass = ReplicationHFileCleaner.class.getCanonicalName(); 3676 if (!plugins.contains(cleanerClass)) { 3677 conf.set(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS, plugins + "," + cleanerClass); 3678 } 3679 } 3680 } 3681}