Skip to content

HBASE-30348 Throttles should work for system tables (excluding quotas, meta, namespace) #8774

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
rmdmattingly wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

HBASE-30348 Throttles should work for system tables (excluding quotas, meta, namespace) #8774

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
190 changes: 133 additions & 57 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,16 @@
import java.io.IOException;
import java.time.Duration;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ClusterMetrics;
import org.apache.hadoop.hbase.ClusterMetrics.Option;
Expand All @@ -35,6 +41,7 @@
import org.apache.hadoop.hbase.ipc.RpcServer;
import org.apache.hadoop.hbase.regionserver.RegionServerServices;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
Expand All @@ -44,6 +51,7 @@
import org.apache.hbase.thirdparty.com.google.common.cache.CacheBuilder;
import org.apache.hbase.thirdparty.com.google.common.cache.CacheLoader;
import org.apache.hbase.thirdparty.com.google.common.cache.LoadingCache;
import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
* Cache that keeps track of the quota settings for the users and tables that are interacting with
Expand All @@ -66,7 +74,9 @@ public class QuotaCache implements Stoppable {
"hbase.quota.user.override.key";
private static final int REFRESH_DEFAULT_PERIOD = 43_200_000; // 12 hours

private final Object initializerLock = new Object();
private static final long INITIAL_LOAD_RETRY_MS = 1_000;

private final AtomicBoolean initialLoadStarted = new AtomicBoolean(false);
private volatile boolean initialized = false;

private volatile Map<String, QuotaState> namespaceQuotaCache = new ConcurrentHashMap<>();
Expand All @@ -85,6 +95,14 @@ public class QuotaCache implements Stoppable {
private QuotaRefresherChore refreshChore;
private boolean stopped = true;

// The initial load runs here rather than on the requesting thread, so that no request ever waits
// for it. The load reads hbase:quota and asks the master for cluster metrics, neither of which is
// necessarily available yet when the first request arrives.
private final ScheduledExecutorService initialLoadExecutor =
Executors.newSingleThreadScheduledExecutor(
new ThreadFactoryBuilder().setNameFormat("quota-cache-initial-load-%d").setDaemon(true)
.setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build());

public QuotaCache(final RegionServerServices rsServices) {
this.rsServices = rsServices;
this.userOverrideRequestAttributeKey =
Expand All @@ -109,6 +127,7 @@ public void stop(final String why) {
LOG.debug("Stopping QuotaRefresherChore chore.");
refreshChore.shutdown(true);
}
initialLoadExecutor.shutdownNow();
stopped = true;
}

Expand All @@ -117,14 +136,34 @@ public boolean isStopped() {
return stopped;
}

/**
* Starts the initial load if it has not been started, without waiting for it. Until it completes,
* lookups return default quota state, which does not throttle.
*/
private void ensureInitialized() {
if (!initialized) {
synchronized (initializerLock) {
if (!initialized) {
refreshChore.chore();
initialized = true;
}
}
if (initialized || !initialLoadStarted.compareAndSet(false, true)) {
return;
}
try {
initialLoadExecutor.execute(this::loadInitialQuotas);
} catch (RejectedExecutionException e) {
LOG.debug("Not loading quota cache because it is stopped", e);
}
}

private void loadInitialQuotas() {
if (refreshChore.refresh()) {
initialized = true;
LOG.info("Quota cache loaded, quotas are now enforced");
return;
}
LOG.warn("Quota cache not fully loaded, serving default quotas and retrying in {}ms",
INITIAL_LOAD_RETRY_MS);
try {
initialLoadExecutor.schedule(this::loadInitialQuotas, INITIAL_LOAD_RETRY_MS,
TimeUnit.MILLISECONDS);
} catch (RejectedExecutionException e) {
LOG.debug("Not retrying quota cache load because it is stopped", e);
}
}

Expand Down Expand Up @@ -155,7 +194,7 @@ private Map<String, QuotaState> fetchNamespaceQuotaStateEntries() throws IOExcep
* @return the limiter associated to the specified user/table
*/
public QuotaLimiter getUserLimiter(final UserGroupInformation ugi, final TableName table) {
if (table.isSystemTable()) {
if (QuotaUtil.isThrottleExempt(table)) {
return NoopQuotaLimiter.get();
}
return getUserQuotaState(ugi).getTableLimiter(table);
Expand Down Expand Up @@ -256,6 +295,11 @@ void forceSynchronousCacheRefresh() {
refreshChore.chore();
}

/** visible for testing */
boolean isInitialized() {
return initialized;
}

/** visible for testing */
Map<String, QuotaState> getNamespaceQuotaCache() {
return namespaceQuotaCache;
Expand Down Expand Up @@ -309,56 +353,90 @@ public synchronized boolean triggerNow() {

@Override
protected void chore() {
synchronized (this) {
LOG.info("Reloading quota cache from hbase:quota table");
updateQuotaFactors();

try {
Map<String, UserQuotaState> newUserQuotaCache =
new ConcurrentHashMap<>(fetchUserQuotaStateEntries());
updateNewCacheFromOld(userQuotaCache, newUserQuotaCache);
userQuotaCache = newUserQuotaCache;
} catch (IOException e) {
LOG.error("Error while fetching user quotas", e);
}
refresh();
}

try {
Map<String, QuotaState> newRegionServerQuotaCache =
new ConcurrentHashMap<>(fetchRegionServerQuotaStateEntries());
updateNewCacheFromOld(regionServerQuotaCache, newRegionServerQuotaCache);
regionServerQuotaCache = newRegionServerQuotaCache;
} catch (IOException e) {
LOG.error("Error while fetching region server quotas", e);
}
/**
* Reloads the cache from hbase:quota.
* @return true if every kind of quota was loaded
*/
synchronized boolean refresh() {
LOG.info("Reloading quota cache from hbase:quota table");
Set<TableName> tablesWithQuotas = getTablesWithQuotas();
updateQuotaFactors(tablesWithQuotas);
boolean loaded = fetchQuotas();

Set<TableName> newTablesWithQuotas = getTablesWithQuotas();
if (!tablesWithQuotas.containsAll(newTablesWithQuotas)) {
LOG.info("Reloading quota cache to compute quota factors for newly limited tables");
updateQuotaFactors(newTablesWithQuotas);
loaded = fetchQuotas();
}
return loaded;
}

try {
Map<TableName, QuotaState> newTableQuotaCache =
new ConcurrentHashMap<>(fetchTableQuotaStateEntries());
updateNewCacheFromOld(tableQuotaCache, newTableQuotaCache);
tableQuotaCache = newTableQuotaCache;
} catch (IOException e) {
LOG.error("Error while refreshing table quotas", e);
}
private Set<TableName> getTablesWithQuotas() {
Set<TableName> tables = new HashSet<>(tableQuotaCache.keySet());
for (UserQuotaState userQuotaState : userQuotaCache.values()) {
tables.addAll(userQuotaState.getTablesWithLimiters());
}
return tables;
}

try {
Map<String, QuotaState> newNamespaceQuotaCache =
new ConcurrentHashMap<>(fetchNamespaceQuotaStateEntries());
updateNewCacheFromOld(namespaceQuotaCache, newNamespaceQuotaCache);
namespaceQuotaCache = newNamespaceQuotaCache;
} catch (IOException e) {
LOG.error("Error while refreshing namespace quotas", e);
}
private boolean fetchQuotas() {
boolean loaded = true;

fetchExceedThrottleQuota();
try {
Map<String, UserQuotaState> newUserQuotaCache =
new ConcurrentHashMap<>(fetchUserQuotaStateEntries());
updateNewCacheFromOld(userQuotaCache, newUserQuotaCache);
userQuotaCache = newUserQuotaCache;
} catch (IOException e) {
LOG.error("Error while fetching user quotas", e);
loaded = false;
}

try {
Map<String, QuotaState> newRegionServerQuotaCache =
new ConcurrentHashMap<>(fetchRegionServerQuotaStateEntries());
updateNewCacheFromOld(regionServerQuotaCache, newRegionServerQuotaCache);
regionServerQuotaCache = newRegionServerQuotaCache;
} catch (IOException e) {
LOG.error("Error while fetching region server quotas", e);
loaded = false;
}

try {
Map<TableName, QuotaState> newTableQuotaCache =
new ConcurrentHashMap<>(fetchTableQuotaStateEntries());
updateNewCacheFromOld(tableQuotaCache, newTableQuotaCache);
tableQuotaCache = newTableQuotaCache;
} catch (IOException e) {
LOG.error("Error while refreshing table quotas", e);
loaded = false;
}

try {
Map<String, QuotaState> newNamespaceQuotaCache =
new ConcurrentHashMap<>(fetchNamespaceQuotaStateEntries());
updateNewCacheFromOld(namespaceQuotaCache, newNamespaceQuotaCache);
namespaceQuotaCache = newNamespaceQuotaCache;
} catch (IOException e) {
LOG.error("Error while refreshing namespace quotas", e);
loaded = false;
}

return fetchExceedThrottleQuota() && loaded;
}

private void fetchExceedThrottleQuota() {
private boolean fetchExceedThrottleQuota() {
try {
QuotaCache.this.exceedThrottleQuotaEnabled =
QuotaUtil.isExceedThrottleQuotaEnabled(rsServices.getConnection());
return true;
} catch (IOException e) {
LOG.warn("Unable to read if exceed throttle quota enabled from quota table", e);
return false;
}
}

Expand All @@ -367,19 +445,17 @@ private void fetchExceedThrottleQuota() {
* user/namespace/user over namespace quota, use [1 / RSNum] as machine factor. For table/user
* over table quota, use [1 / TotalTableRegionNum * MachineTableRegionNum] as machine factor.
*/
private void updateQuotaFactors() {
boolean hasTableQuotas = !tableQuotaCache.entrySet().isEmpty()
|| userQuotaCache.values().stream().anyMatch(UserQuotaState::hasTableLimiters);
if (hasTableQuotas) {
updateTableMachineQuotaFactors();
} else {
private void updateQuotaFactors(Set<TableName> tablesWithQuotas) {
if (tablesWithQuotas.isEmpty()) {
updateOnlyMachineQuotaFactors();
} else {
updateTableMachineQuotaFactors(tablesWithQuotas);
}
}

/**
* This method is cheaper than {@link #updateTableMachineQuotaFactors()} and should be used if
* we don't have any table quotas in the cache.
* This method is cheaper than {@link #updateTableMachineQuotaFactors(Set)} and should be used
* if we don't have any table quotas in the cache.
*/
private void updateOnlyMachineQuotaFactors() {
Optional<Integer> rsSize = regionServersSize.get();
Expand All @@ -394,7 +470,7 @@ private void updateOnlyMachineQuotaFactors() {
* This will call {@link #updateMachineQuotaFactors(int)}, and then update the table machine
* factors as well. This relies on a more expensive query for ClusterMetrics.
*/
private void updateTableMachineQuotaFactors() {
private void updateTableMachineQuotaFactors(Set<TableName> tablesWithQuotas) {
Optional<ClusterMetrics> clusterMetricsMaybe = tableRegionStatesClusterMetrics.get();
if (!clusterMetricsMaybe.isPresent()) {
tableRegionStatesClusterMetrics.refresh();
Expand All @@ -407,7 +483,7 @@ private void updateTableMachineQuotaFactors() {
clusterMetrics.getTableRegionStatesCount();

// Update table machine quota factors
for (TableName tableName : tableQuotaCache.keySet()) {
for (TableName tableName : tablesWithQuotas) {
if (tableRegionStatesCount.containsKey(tableName)) {
double factor = 1;
try {
Expand Down
11 changes: 11 additions & 0 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,17 @@ public static boolean isQuotaEnabled(final Configuration conf) {
return conf.getBoolean(QUOTA_CONF_KEY, QUOTA_ENABLED_DEFAULT);
}

/**
* Returns true if the given table can never be throttled. These tables are read while a master is
* still initializing, before quotas can be served, so throttling them could prevent a throttle
* from ever being lifted.
*/
@SuppressWarnings("deprecation") // hbase:namespace still exists on pre-3.x clusters
public static boolean isThrottleExempt(final TableName tableName) {
return TableName.META_TABLE_NAME.equals(tableName) || QUOTA_TABLE_NAME.equals(tableName)
|| TableName.NAMESPACE_TABLE_NAME.equals(tableName);
}

/*
* ========================================================================= Quota "settings"
* helpers
Expand Down
2 changes: 1 addition & 1 deletion hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RegionServerRpcQuotaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ QuotaCache getQuotaCache() {
*/
public OperationQuota getQuota(final UserGroupInformation ugi, final TableName table,
final int blockSizeBytes) {
if (isQuotaEnabled() && !table.isSystemTable() && isRpcThrottleEnabled()) {
if (isQuotaEnabled() && !QuotaUtil.isThrottleExempt(table) && isRpcThrottleEnabled()) {
UserQuotaState userQuotaState = quotaCache.getUserQuotaState(ugi);
QuotaLimiter userLimiter = userQuotaState.getTableLimiter(table);

Expand Down
4 changes: 2 additions & 2 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/UserQuotaState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,8 +111,8 @@ public void setQuotas(Configuration conf, final String namespace, Quotas quotas)
namespaceLimiters = setLimiter(conf, namespaceLimiters, namespace, quotas);
}

public boolean hasTableLimiters() {
return tableLimiters != null && !tableLimiters.isEmpty();
public synchronized Set<TableName> getTablesWithLimiters() {
return tableLimiters == null ? new HashSet<>() : new HashSet<>(tableLimiters.keySet());
}

private <K> Map<K, QuotaLimiter> setLimiter(Configuration conf, Map<K, QuotaLimiter> limiters,
Expand Down
12 changes: 12 additions & 0 deletions hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hbase.HBaseTestingUtil;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.testclassification.MediumTests;
Expand Down Expand Up @@ -63,6 +64,7 @@ public void testUserQuotaLookup() throws Exception {
final Admin admin = TEST_UTIL.getAdmin();
admin.setQuota(QuotaSettingsFactory.throttleUser("my_user", ThrottleType.READ_NUMBER, 3737,
TimeUnit.MINUTES));
awaitInitialLoad(quotaCache);

// Setting a quota and then looking it up from the cache should work, even if the cache has not
// refreshed
Expand Down Expand Up @@ -93,6 +95,7 @@ public void testTableQuotaLookup() throws Exception {
final Admin admin = TEST_UTIL.getAdmin();
admin.setQuota(QuotaSettingsFactory.throttleTable(TableName.valueOf("my_table"),
ThrottleType.READ_NUMBER, 3737, TimeUnit.MINUTES));
awaitInitialLoad(quotaCache);

// Setting a quota and then looking it up from the cache should work, even if the cache has not
// refreshed
Expand All @@ -119,6 +122,7 @@ public void testNamespaceQuotaLookup() throws Exception {
final Admin admin = TEST_UTIL.getAdmin();
admin.setQuota(QuotaSettingsFactory.throttleNamespace("my_namespace", ThrottleType.READ_NUMBER,
3737, TimeUnit.MINUTES));
awaitInitialLoad(quotaCache);

// Setting a quota and then looking it up from the cache should work, even if the cache has not
// refreshed
Expand All @@ -145,6 +149,7 @@ public void testRegionServerQuotaLookup() throws Exception {
final Admin admin = TEST_UTIL.getAdmin();
admin.setQuota(QuotaSettingsFactory.throttleRegionServer("my_region_server",
ThrottleType.READ_NUMBER, 3737, TimeUnit.MINUTES));
awaitInitialLoad(quotaCache);

// Setting a quota and then looking it up from the cache should work, even if the cache has not
// refreshed
Expand All @@ -163,4 +168,11 @@ public void testRegionServerQuotaLookup() throws Exception {
quotaLimiter = quotaCache.getRegionServerQuotaLimiter("my_region_server2");
assertEquals(Long.MAX_VALUE, quotaLimiter.getReadNumLimit());
}

private static void awaitInitialLoad(QuotaCache quotaCache) {
TEST_UTIL.waitFor(30_000, () -> {
quotaCache.getNamespaceLimiter(NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR);
return quotaCache.isInitialized();
});
}
}
Loading
Loading