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.quotas; 019 020import java.io.IOException; 021import java.util.Collections; 022import java.util.HashMap; 023import java.util.HashSet; 024import java.util.Iterator; 025import java.util.Map; 026import java.util.Map.Entry; 027import java.util.Optional; 028import java.util.Set; 029import java.util.concurrent.ConcurrentHashMap; 030import org.apache.commons.lang3.builder.HashCodeBuilder; 031import org.apache.hadoop.conf.Configuration; 032import org.apache.hadoop.fs.FileSystem; 033import org.apache.hadoop.hbase.DoNotRetryIOException; 034import org.apache.hadoop.hbase.NamespaceDescriptor; 035import org.apache.hadoop.hbase.RegionStateListener; 036import org.apache.hadoop.hbase.TableName; 037import org.apache.hadoop.hbase.client.Connection; 038import org.apache.hadoop.hbase.client.RegionInfo; 039import org.apache.hadoop.hbase.master.MasterServices; 040import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch; 041import org.apache.hadoop.hbase.master.procedure.SwitchRpcThrottleProcedure; 042import org.apache.hadoop.hbase.namespace.NamespaceAuditor; 043import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot.SpaceQuotaStatus; 044import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 045import org.apache.yetus.audience.InterfaceAudience; 046import org.apache.yetus.audience.InterfaceStability; 047import org.slf4j.Logger; 048import org.slf4j.LoggerFactory; 049 050import org.apache.hbase.thirdparty.com.google.common.collect.HashMultimap; 051import org.apache.hbase.thirdparty.com.google.common.collect.Maps; 052import org.apache.hbase.thirdparty.com.google.protobuf.TextFormat; 053 054import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; 055import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsRpcThrottleEnabledRequest; 056import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsRpcThrottleEnabledResponse; 057import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SetQuotaRequest; 058import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SetQuotaResponse; 059import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchExceedThrottleQuotaRequest; 060import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchExceedThrottleQuotaResponse; 061import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchRpcThrottleRequest; 062import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SwitchRpcThrottleResponse; 063import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.FileArchiveNotificationRequest; 064import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.FileArchiveNotificationRequest.FileWithSize; 065 066/** 067 * Master Quota Manager. It is responsible for initialize the quota table on the first-run and 068 * provide the admin operations to interact with the quota table. TODO: FUTURE: The master will be 069 * responsible to notify each RS of quota changes and it will do the "quota aggregation" when the 070 * QuotaScope is CLUSTER. 071 */ 072@InterfaceAudience.Private 073@InterfaceStability.Evolving 074public class MasterQuotaManager implements RegionStateListener { 075 private static final Logger LOG = LoggerFactory.getLogger(MasterQuotaManager.class); 076 private static final Map<RegionInfo, Long> EMPTY_MAP = 077 Collections.unmodifiableMap(new HashMap<>()); 078 079 private final MasterServices masterServices; 080 private NamedLock<String> namespaceLocks; 081 private NamedLock<TableName> tableLocks; 082 private NamedLock<String> userLocks; 083 private NamedLock<String> regionServerLocks; 084 private boolean initialized = false; 085 private NamespaceAuditor namespaceQuotaManager; 086 private ConcurrentHashMap<RegionInfo, SizeSnapshotWithTimestamp> regionSizes; 087 088 public MasterQuotaManager(MasterServices masterServices) { 089 this.masterServices = masterServices; 090 } 091 092 public void start() throws IOException { 093 // If the user doesn't want the quota support skip all the initializations. 094 if (!QuotaUtil.isQuotaEnabled(masterServices.getConfiguration())) { 095 LOG.info("Quota support disabled"); 096 return; 097 } 098 099 // Create the quota table if missing 100 if (!masterServices.getTableDescriptors().exists(QuotaUtil.QUOTA_TABLE_NAME)) { 101 LOG.info("Quota table not found. Creating..."); 102 createQuotaTable(); 103 } 104 105 LOG.info("Initializing quota support"); 106 namespaceLocks = new NamedLock<>(); 107 tableLocks = new NamedLock<>(); 108 userLocks = new NamedLock<>(); 109 regionServerLocks = new NamedLock<>(); 110 regionSizes = new ConcurrentHashMap<>(); 111 112 namespaceQuotaManager = new NamespaceAuditor(masterServices); 113 namespaceQuotaManager.start(); 114 initialized = true; 115 } 116 117 public void stop() { 118 } 119 120 public boolean isQuotaInitialized() { 121 return initialized && namespaceQuotaManager.isInitialized(); 122 } 123 124 /* 125 * ========================================================================== Admin operations to 126 * manage the quota table 127 */ 128 public SetQuotaResponse setQuota(final SetQuotaRequest req) 129 throws IOException, InterruptedException { 130 checkQuotaSupport(); 131 132 if (req.hasUserName()) { 133 userLocks.lock(req.getUserName()); 134 try { 135 if (req.hasTableName()) { 136 setUserQuota(req.getUserName(), ProtobufUtil.toTableName(req.getTableName()), req); 137 } else if (req.hasNamespace()) { 138 setUserQuota(req.getUserName(), req.getNamespace(), req); 139 } else { 140 setUserQuota(req.getUserName(), req); 141 } 142 } finally { 143 userLocks.unlock(req.getUserName()); 144 } 145 } else if (req.hasTableName()) { 146 TableName table = ProtobufUtil.toTableName(req.getTableName()); 147 tableLocks.lock(table); 148 try { 149 setTableQuota(table, req); 150 } finally { 151 tableLocks.unlock(table); 152 } 153 } else if (req.hasNamespace()) { 154 namespaceLocks.lock(req.getNamespace()); 155 try { 156 setNamespaceQuota(req.getNamespace(), req); 157 } finally { 158 namespaceLocks.unlock(req.getNamespace()); 159 } 160 } else if (req.hasRegionServer()) { 161 regionServerLocks.lock(req.getRegionServer()); 162 try { 163 setRegionServerQuota(req.getRegionServer(), req); 164 } finally { 165 regionServerLocks.unlock(req.getRegionServer()); 166 } 167 } else { 168 throw new DoNotRetryIOException(new UnsupportedOperationException( 169 "a user, a table, a namespace or region server must be specified")); 170 } 171 return SetQuotaResponse.newBuilder().build(); 172 } 173 174 public void setUserQuota(final String userName, final SetQuotaRequest req) 175 throws IOException, InterruptedException { 176 setQuota(req, new SetQuotaOperations() { 177 @Override 178 public GlobalQuotaSettingsImpl fetch() throws IOException { 179 return new GlobalQuotaSettingsImpl(req.getUserName(), null, null, null, 180 QuotaUtil.getUserQuota(masterServices.getConnection(), userName)); 181 } 182 183 @Override 184 public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 185 QuotaUtil.addUserQuota(masterServices.getConnection(), userName, quotaPojo.toQuotas()); 186 } 187 188 @Override 189 public void delete() throws IOException { 190 QuotaUtil.deleteUserQuota(masterServices.getConnection(), userName); 191 } 192 193 @Override 194 public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 195 masterServices.getMasterCoprocessorHost().preSetUserQuota(userName, quotaPojo); 196 } 197 198 @Override 199 public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 200 masterServices.getMasterCoprocessorHost().postSetUserQuota(userName, quotaPojo); 201 } 202 }); 203 } 204 205 public void setUserQuota(final String userName, final TableName table, final SetQuotaRequest req) 206 throws IOException, InterruptedException { 207 setQuota(req, new SetQuotaOperations() { 208 @Override 209 public GlobalQuotaSettingsImpl fetch() throws IOException { 210 return new GlobalQuotaSettingsImpl(userName, table, null, null, 211 QuotaUtil.getUserQuota(masterServices.getConnection(), userName, table)); 212 } 213 214 @Override 215 public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 216 QuotaUtil.addUserQuota(masterServices.getConnection(), userName, table, 217 quotaPojo.toQuotas()); 218 } 219 220 @Override 221 public void delete() throws IOException { 222 QuotaUtil.deleteUserQuota(masterServices.getConnection(), userName, table); 223 } 224 225 @Override 226 public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 227 masterServices.getMasterCoprocessorHost().preSetUserQuota(userName, table, quotaPojo); 228 } 229 230 @Override 231 public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 232 masterServices.getMasterCoprocessorHost().postSetUserQuota(userName, table, quotaPojo); 233 } 234 }); 235 } 236 237 public void setUserQuota(final String userName, final String namespace, final SetQuotaRequest req) 238 throws IOException, InterruptedException { 239 setQuota(req, new SetQuotaOperations() { 240 @Override 241 public GlobalQuotaSettingsImpl fetch() throws IOException { 242 return new GlobalQuotaSettingsImpl(userName, null, namespace, null, 243 QuotaUtil.getUserQuota(masterServices.getConnection(), userName, namespace)); 244 } 245 246 @Override 247 public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 248 QuotaUtil.addUserQuota(masterServices.getConnection(), userName, namespace, 249 quotaPojo.toQuotas()); 250 } 251 252 @Override 253 public void delete() throws IOException { 254 QuotaUtil.deleteUserQuota(masterServices.getConnection(), userName, namespace); 255 } 256 257 @Override 258 public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 259 masterServices.getMasterCoprocessorHost().preSetUserQuota(userName, namespace, quotaPojo); 260 } 261 262 @Override 263 public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 264 masterServices.getMasterCoprocessorHost().postSetUserQuota(userName, namespace, quotaPojo); 265 } 266 }); 267 } 268 269 public void setTableQuota(final TableName table, final SetQuotaRequest req) 270 throws IOException, InterruptedException { 271 setQuota(req, new SetQuotaOperations() { 272 @Override 273 public GlobalQuotaSettingsImpl fetch() throws IOException { 274 return new GlobalQuotaSettingsImpl(null, table, null, null, 275 QuotaUtil.getTableQuota(masterServices.getConnection(), table)); 276 } 277 278 @Override 279 public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 280 QuotaUtil.addTableQuota(masterServices.getConnection(), table, quotaPojo.toQuotas()); 281 } 282 283 @Override 284 public void delete() throws IOException { 285 SpaceQuotaSnapshot currSnapshotOfTable = 286 QuotaTableUtil.getCurrentSnapshotFromQuotaTable(masterServices.getConnection(), table); 287 QuotaUtil.deleteTableQuota(masterServices.getConnection(), table); 288 if (currSnapshotOfTable != null) { 289 SpaceQuotaStatus quotaStatus = currSnapshotOfTable.getQuotaStatus(); 290 if ( 291 SpaceViolationPolicy.DISABLE == quotaStatus.getPolicy().orElse(null) 292 && quotaStatus.isInViolation() 293 ) { 294 QuotaUtil.enableTableIfNotEnabled(masterServices.getConnection(), table); 295 } 296 } 297 } 298 299 @Override 300 public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 301 masterServices.getMasterCoprocessorHost().preSetTableQuota(table, quotaPojo); 302 } 303 304 @Override 305 public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 306 masterServices.getMasterCoprocessorHost().postSetTableQuota(table, quotaPojo); 307 } 308 }); 309 } 310 311 public void setNamespaceQuota(final String namespace, final SetQuotaRequest req) 312 throws IOException, InterruptedException { 313 setQuota(req, new SetQuotaOperations() { 314 @Override 315 public GlobalQuotaSettingsImpl fetch() throws IOException { 316 return new GlobalQuotaSettingsImpl(null, null, namespace, null, 317 QuotaUtil.getNamespaceQuota(masterServices.getConnection(), namespace)); 318 } 319 320 @Override 321 public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 322 QuotaUtil.addNamespaceQuota(masterServices.getConnection(), namespace, 323 quotaPojo.toQuotas()); 324 } 325 326 @Override 327 public void delete() throws IOException { 328 QuotaUtil.deleteNamespaceQuota(masterServices.getConnection(), namespace); 329 } 330 331 @Override 332 public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 333 masterServices.getMasterCoprocessorHost().preSetNamespaceQuota(namespace, quotaPojo); 334 } 335 336 @Override 337 public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 338 masterServices.getMasterCoprocessorHost().postSetNamespaceQuota(namespace, quotaPojo); 339 } 340 }); 341 } 342 343 public void setRegionServerQuota(final String regionServer, final SetQuotaRequest req) 344 throws IOException, InterruptedException { 345 setQuota(req, new SetQuotaOperations() { 346 @Override 347 public GlobalQuotaSettingsImpl fetch() throws IOException { 348 return new GlobalQuotaSettingsImpl(null, null, null, regionServer, 349 QuotaUtil.getRegionServerQuota(masterServices.getConnection(), regionServer)); 350 } 351 352 @Override 353 public void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 354 QuotaUtil.addRegionServerQuota(masterServices.getConnection(), regionServer, 355 quotaPojo.toQuotas()); 356 } 357 358 @Override 359 public void delete() throws IOException { 360 QuotaUtil.deleteRegionServerQuota(masterServices.getConnection(), regionServer); 361 } 362 363 @Override 364 public void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 365 masterServices.getMasterCoprocessorHost().preSetRegionServerQuota(regionServer, quotaPojo); 366 } 367 368 @Override 369 public void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException { 370 masterServices.getMasterCoprocessorHost().postSetRegionServerQuota(regionServer, quotaPojo); 371 } 372 }); 373 } 374 375 public void setNamespaceQuota(NamespaceDescriptor desc) throws IOException { 376 if (initialized) { 377 this.namespaceQuotaManager.addNamespace(desc); 378 } 379 } 380 381 public void removeNamespaceQuota(String namespace) throws IOException { 382 if (initialized) { 383 this.namespaceQuotaManager.deleteNamespace(namespace); 384 } 385 } 386 387 public SwitchRpcThrottleResponse switchRpcThrottle(SwitchRpcThrottleRequest request) 388 throws IOException { 389 boolean rpcThrottle = request.getRpcThrottleEnabled(); 390 if (initialized) { 391 masterServices.getMasterCoprocessorHost().preSwitchRpcThrottle(rpcThrottle); 392 boolean oldRpcThrottle = masterServices.getRpcThrottleStateStore().get(); 393 if (rpcThrottle != oldRpcThrottle) { 394 LOG.info("{} switch rpc throttle from {} to {}", masterServices.getClientIdAuditPrefix(), 395 oldRpcThrottle, rpcThrottle); 396 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch(); 397 SwitchRpcThrottleProcedure procedure = 398 new SwitchRpcThrottleProcedure(rpcThrottle, masterServices.getServerName(), latch); 399 masterServices.getMasterProcedureExecutor().submitProcedure(procedure); 400 latch.await(); 401 } else { 402 LOG.warn("Skip switch rpc throttle to {} because it's the same with old value", 403 rpcThrottle); 404 } 405 SwitchRpcThrottleResponse response = SwitchRpcThrottleResponse.newBuilder() 406 .setPreviousRpcThrottleEnabled(oldRpcThrottle).build(); 407 masterServices.getMasterCoprocessorHost().postSwitchRpcThrottle(oldRpcThrottle, rpcThrottle); 408 return response; 409 } else { 410 LOG.warn("Skip switch rpc throttle to {} because rpc quota is disabled", rpcThrottle); 411 return SwitchRpcThrottleResponse.newBuilder().setPreviousRpcThrottleEnabled(false).build(); 412 } 413 } 414 415 public IsRpcThrottleEnabledResponse isRpcThrottleEnabled(IsRpcThrottleEnabledRequest request) 416 throws IOException { 417 if (initialized) { 418 masterServices.getMasterCoprocessorHost().preIsRpcThrottleEnabled(); 419 boolean enabled = isRpcThrottleEnabled(); 420 IsRpcThrottleEnabledResponse response = 421 IsRpcThrottleEnabledResponse.newBuilder().setRpcThrottleEnabled(enabled).build(); 422 masterServices.getMasterCoprocessorHost().postIsRpcThrottleEnabled(enabled); 423 return response; 424 } else { 425 LOG.warn("Skip get rpc throttle because rpc quota is disabled"); 426 return IsRpcThrottleEnabledResponse.newBuilder().setRpcThrottleEnabled(false).build(); 427 } 428 } 429 430 public boolean isRpcThrottleEnabled() throws IOException { 431 return initialized && masterServices.getRpcThrottleStateStore().get(); 432 } 433 434 public SwitchExceedThrottleQuotaResponse 435 switchExceedThrottleQuota(SwitchExceedThrottleQuotaRequest request) throws IOException { 436 boolean enabled = request.getExceedThrottleQuotaEnabled(); 437 if (initialized) { 438 masterServices.getMasterCoprocessorHost().preSwitchExceedThrottleQuota(enabled); 439 boolean previousEnabled = 440 QuotaUtil.isExceedThrottleQuotaEnabled(masterServices.getConnection()); 441 if (previousEnabled == enabled) { 442 LOG.warn("Skip switch exceed throttle quota to {} because it's the same with old value", 443 enabled); 444 } else { 445 QuotaUtil.switchExceedThrottleQuota(masterServices.getConnection(), enabled); 446 LOG.info("{} switch exceed throttle quota from {} to {}", 447 masterServices.getClientIdAuditPrefix(), previousEnabled, enabled); 448 } 449 SwitchExceedThrottleQuotaResponse response = SwitchExceedThrottleQuotaResponse.newBuilder() 450 .setPreviousExceedThrottleQuotaEnabled(previousEnabled).build(); 451 masterServices.getMasterCoprocessorHost().postSwitchExceedThrottleQuota(previousEnabled, 452 enabled); 453 return response; 454 } else { 455 LOG.warn("Skip switch exceed throttle quota to {} because quota is disabled", enabled); 456 return SwitchExceedThrottleQuotaResponse.newBuilder() 457 .setPreviousExceedThrottleQuotaEnabled(false).build(); 458 } 459 } 460 461 public boolean isExceedThrottleQuotaEnabled() throws IOException { 462 return initialized 463 ? QuotaUtil.isExceedThrottleQuotaEnabled(masterServices.getConnection()) 464 : false; 465 } 466 467 private void setQuota(final SetQuotaRequest req, final SetQuotaOperations quotaOps) 468 throws IOException, InterruptedException { 469 if (req.hasRemoveAll() && req.getRemoveAll() == true) { 470 quotaOps.preApply(null); 471 quotaOps.delete(); 472 quotaOps.postApply(null); 473 return; 474 } 475 476 // Apply quota changes 477 GlobalQuotaSettingsImpl currentQuota = quotaOps.fetch(); 478 if (LOG.isTraceEnabled()) { 479 LOG.trace( 480 "Current quota for request(" + TextFormat.shortDebugString(req) + "): " + currentQuota); 481 } 482 // Call the appropriate "pre" CP hook with the current quota value (may be null) 483 quotaOps.preApply(currentQuota); 484 // Translate the protobuf request back into a POJO 485 QuotaSettings newQuota = QuotaSettings.buildFromProto(req); 486 if (LOG.isTraceEnabled()) { 487 LOG.trace("Deserialized quota from request: " + newQuota); 488 } 489 490 // Merge the current quota settings with the new quota settings the user provided. 491 // 492 // NB: while SetQuotaRequest technically allows for multi types of quotas to be set in one 493 // message, the Java API (in Admin/AsyncAdmin) does not. Assume there is only one type. 494 GlobalQuotaSettingsImpl mergedQuota = currentQuota.merge(newQuota); 495 if (LOG.isTraceEnabled()) { 496 LOG.trace("Computed merged quota from current quota and user request: " + mergedQuota); 497 } 498 499 // Submit new changes 500 if (mergedQuota == null) { 501 quotaOps.delete(); 502 } else { 503 quotaOps.update(mergedQuota); 504 } 505 // Advertise the final result via the "post" CP hook 506 quotaOps.postApply(mergedQuota); 507 } 508 509 public void checkNamespaceTableAndRegionQuota(TableName tName, int regions) throws IOException { 510 if (initialized) { 511 namespaceQuotaManager.checkQuotaToCreateTable(tName, regions); 512 } 513 } 514 515 public void checkAndUpdateNamespaceRegionQuota(TableName tName, int regions) throws IOException { 516 if (initialized) { 517 namespaceQuotaManager.checkQuotaToUpdateRegion(tName, regions); 518 } 519 } 520 521 /** Returns cached region count, or -1 if quota manager is disabled or table status not found */ 522 public int getRegionCountOfTable(TableName tName) throws IOException { 523 if (initialized) { 524 return namespaceQuotaManager.getRegionCountOfTable(tName); 525 } 526 return -1; 527 } 528 529 @Override 530 public void onRegionMerged(RegionInfo mergedRegion) throws IOException { 531 if (initialized) { 532 namespaceQuotaManager.updateQuotaForRegionMerge(mergedRegion); 533 } 534 } 535 536 @Override 537 public void onRegionSplit(RegionInfo hri) throws IOException { 538 if (initialized) { 539 namespaceQuotaManager.checkQuotaToSplitRegion(hri); 540 } 541 } 542 543 /** 544 * Remove table from namespace quota. 545 * @param tName - The table name to update quota usage. 546 * @throws IOException Signals that an I/O exception has occurred. 547 */ 548 public void removeTableFromNamespaceQuota(TableName tName) throws IOException { 549 if (initialized) { 550 namespaceQuotaManager.removeFromNamespaceUsage(tName); 551 } 552 } 553 554 public NamespaceAuditor getNamespaceQuotaManager() { 555 return this.namespaceQuotaManager; 556 } 557 558 /** 559 * Encapsulates CRUD quota operations for some subject. 560 */ 561 private static interface SetQuotaOperations { 562 /** 563 * Fetches the current quota settings for the subject. 564 */ 565 GlobalQuotaSettingsImpl fetch() throws IOException; 566 567 /** 568 * Deletes the quota for the subject. 569 */ 570 void delete() throws IOException; 571 572 /** 573 * Persist the given quota for the subject. 574 */ 575 void update(GlobalQuotaSettingsImpl quotaPojo) throws IOException; 576 577 /** 578 * Performs some action before {@link #update(GlobalQuotaSettingsImpl)} with the current quota 579 * for the subject. 580 */ 581 void preApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException; 582 583 /** 584 * Performs some action after {@link #update(GlobalQuotaSettingsImpl)} with the resulting quota 585 * from the request action for the subject. 586 */ 587 void postApply(GlobalQuotaSettingsImpl quotaPojo) throws IOException; 588 } 589 590 /* 591 * ========================================================================== Helpers 592 */ 593 594 private void checkQuotaSupport() throws IOException { 595 if (!QuotaUtil.isQuotaEnabled(masterServices.getConfiguration())) { 596 throw new DoNotRetryIOException(new UnsupportedOperationException("quota support disabled")); 597 } 598 if (!initialized) { 599 long maxWaitTime = masterServices.getConfiguration() 600 .getLong("hbase.master.wait.for.quota.manager.init", 30000); // default is 30 seconds. 601 long startTime = EnvironmentEdgeManager.currentTime(); 602 do { 603 try { 604 Thread.sleep(100); 605 } catch (InterruptedException e) { 606 LOG.warn("Interrupted while waiting for Quota Manager to be initialized."); 607 break; 608 } 609 } while (!initialized && (EnvironmentEdgeManager.currentTime() - startTime) < maxWaitTime); 610 if (!initialized) { 611 throw new IOException("Quota manager is uninitialized, please retry later."); 612 } 613 } 614 } 615 616 private void createQuotaTable() throws IOException { 617 masterServices.createSystemTable(QuotaUtil.QUOTA_TABLE_DESC); 618 } 619 620 private static class NamedLock<T> { 621 private final HashSet<T> locks = new HashSet<>(); 622 623 public void lock(final T name) throws InterruptedException { 624 synchronized (locks) { 625 while (locks.contains(name)) { 626 locks.wait(); 627 } 628 locks.add(name); 629 } 630 } 631 632 public void unlock(final T name) { 633 synchronized (locks) { 634 locks.remove(name); 635 locks.notifyAll(); 636 } 637 } 638 } 639 640 @Override 641 public void onRegionSplitReverted(RegionInfo hri) throws IOException { 642 if (initialized) { 643 this.namespaceQuotaManager.removeRegionFromNamespaceUsage(hri); 644 } 645 } 646 647 /** 648 * Holds the size of a region at the given time, millis since the epoch. 649 */ 650 private static class SizeSnapshotWithTimestamp { 651 private final long size; 652 private final long time; 653 654 public SizeSnapshotWithTimestamp(long size, long time) { 655 this.size = size; 656 this.time = time; 657 } 658 659 public long getSize() { 660 return size; 661 } 662 663 public long getTime() { 664 return time; 665 } 666 667 @Override 668 public boolean equals(Object o) { 669 if (o instanceof SizeSnapshotWithTimestamp) { 670 SizeSnapshotWithTimestamp other = (SizeSnapshotWithTimestamp) o; 671 return size == other.size && time == other.time; 672 } 673 return false; 674 } 675 676 @Override 677 public int hashCode() { 678 HashCodeBuilder hcb = new HashCodeBuilder(); 679 return hcb.append(size).append(time).toHashCode(); 680 } 681 682 @Override 683 public String toString() { 684 StringBuilder sb = new StringBuilder(32); 685 sb.append("SizeSnapshotWithTimestamp={size=").append(size).append("B, "); 686 sb.append("time=").append(time).append("}"); 687 return sb.toString(); 688 } 689 } 690 691 void initializeRegionSizes() { 692 assert regionSizes == null; 693 this.regionSizes = new ConcurrentHashMap<>(); 694 } 695 696 public void addRegionSize(RegionInfo hri, long size, long time) { 697 if (regionSizes == null) { 698 return; 699 } 700 regionSizes.put(hri, new SizeSnapshotWithTimestamp(size, time)); 701 } 702 703 public Map<RegionInfo, Long> snapshotRegionSizes() { 704 if (regionSizes == null) { 705 return EMPTY_MAP; 706 } 707 708 Map<RegionInfo, Long> copy = new HashMap<>(); 709 for (Entry<RegionInfo, SizeSnapshotWithTimestamp> entry : regionSizes.entrySet()) { 710 copy.put(entry.getKey(), entry.getValue().getSize()); 711 } 712 return copy; 713 } 714 715 int pruneEntriesOlderThan(long timeToPruneBefore, QuotaObserverChore quotaObserverChore) { 716 if (regionSizes == null) { 717 return 0; 718 } 719 int numEntriesRemoved = 0; 720 Iterator<Entry<RegionInfo, SizeSnapshotWithTimestamp>> iterator = 721 regionSizes.entrySet().iterator(); 722 while (iterator.hasNext()) { 723 RegionInfo regionInfo = iterator.next().getKey(); 724 long currentEntryTime = regionSizes.get(regionInfo).getTime(); 725 // do not prune the entries if table is in violation and 726 // violation policy is disable to avoid cycle of enable/disable. 727 // Please refer HBASE-22012 for more details. 728 // prune entries older than time. 729 if ( 730 currentEntryTime < timeToPruneBefore 731 && !isInViolationAndPolicyDisable(regionInfo.getTable(), quotaObserverChore) 732 ) { 733 iterator.remove(); 734 numEntriesRemoved++; 735 } 736 } 737 return numEntriesRemoved; 738 } 739 740 /** 741 * Method to check if a table is in violation and policy set on table is DISABLE. 742 * @param tableName tableName to check. 743 * @param quotaObserverChore QuotaObserverChore instance 744 * @return returns true if table is in violation and policy is disable else false. 745 */ 746 private boolean isInViolationAndPolicyDisable(TableName tableName, 747 QuotaObserverChore quotaObserverChore) { 748 boolean isInViolationAtTable = false; 749 boolean isInViolationAtNamespace = false; 750 SpaceViolationPolicy tablePolicy = null; 751 SpaceViolationPolicy namespacePolicy = null; 752 // Get Current Snapshot for the given table 753 SpaceQuotaSnapshot tableQuotaSnapshot = quotaObserverChore.getTableQuotaSnapshot(tableName); 754 SpaceQuotaSnapshot namespaceQuotaSnapshot = 755 quotaObserverChore.getNamespaceQuotaSnapshot(tableName.getNamespaceAsString()); 756 if (tableQuotaSnapshot != null) { 757 // check if table in violation 758 isInViolationAtTable = tableQuotaSnapshot.getQuotaStatus().isInViolation(); 759 Optional<SpaceViolationPolicy> policy = tableQuotaSnapshot.getQuotaStatus().getPolicy(); 760 if (policy.isPresent()) { 761 tablePolicy = policy.get(); 762 } 763 } 764 if (namespaceQuotaSnapshot != null) { 765 // check namespace in violation 766 isInViolationAtNamespace = namespaceQuotaSnapshot.getQuotaStatus().isInViolation(); 767 Optional<SpaceViolationPolicy> policy = namespaceQuotaSnapshot.getQuotaStatus().getPolicy(); 768 if (policy.isPresent()) { 769 namespacePolicy = policy.get(); 770 } 771 } 772 return (tablePolicy == SpaceViolationPolicy.DISABLE && isInViolationAtTable) 773 || (namespacePolicy == SpaceViolationPolicy.DISABLE && isInViolationAtNamespace); 774 } 775 776 public void processFileArchivals(FileArchiveNotificationRequest request, Connection conn, 777 Configuration conf, FileSystem fs) throws IOException { 778 final HashMultimap<TableName, Entry<String, Long>> archivedFilesByTable = HashMultimap.create(); 779 // Group the archived files by table 780 for (FileWithSize fileWithSize : request.getArchivedFilesList()) { 781 TableName tn = ProtobufUtil.toTableName(fileWithSize.getTableName()); 782 archivedFilesByTable.put(tn, 783 Maps.immutableEntry(fileWithSize.getName(), fileWithSize.getSize())); 784 } 785 if (LOG.isTraceEnabled()) { 786 LOG.trace("Grouped archived files by table: " + archivedFilesByTable); 787 } 788 // Report each set of files to the appropriate object 789 for (TableName tn : archivedFilesByTable.keySet()) { 790 final Set<Entry<String, Long>> filesWithSize = archivedFilesByTable.get(tn); 791 final FileArchiverNotifier notifier = 792 FileArchiverNotifierFactoryImpl.getInstance().get(conn, conf, fs, tn); 793 notifier.addArchivedFiles(filesWithSize); 794 } 795 } 796 797 /** 798 * Removes each region size entry where the RegionInfo references the provided TableName. 799 * @param tableName tableName. 800 */ 801 public void removeRegionSizesForTable(TableName tableName) { 802 regionSizes.keySet().removeIf(regionInfo -> regionInfo.getTable().equals(tableName)); 803 } 804}