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.List; 022import java.util.Optional; 023import java.util.concurrent.TimeUnit; 024import java.util.function.Supplier; 025import org.apache.hadoop.conf.Configuration; 026import org.apache.hadoop.hbase.TableName; 027import org.apache.hadoop.hbase.client.TableDescriptor; 028import org.apache.hadoop.hbase.conf.ConfigurationObserver; 029import org.apache.hadoop.hbase.ipc.RpcScheduler; 030import org.apache.hadoop.hbase.ipc.RpcServer; 031import org.apache.hadoop.hbase.regionserver.Region; 032import org.apache.hadoop.hbase.regionserver.RegionServerServices; 033import org.apache.hadoop.hbase.security.User; 034import org.apache.hadoop.security.UserGroupInformation; 035import org.apache.yetus.audience.InterfaceAudience; 036import org.apache.yetus.audience.InterfaceStability; 037import org.slf4j.Logger; 038import org.slf4j.LoggerFactory; 039 040import org.apache.hbase.thirdparty.com.google.common.base.Suppliers; 041 042import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; 043 044/** 045 * Region Server Quota Manager. It is responsible to provide access to the quota information of each 046 * user/table. The direct user of this class is the RegionServer that will get and check the 047 * user/table quota for each operation (put, get, scan). For system tables and user/table with a 048 * quota specified, the quota check will be a noop. 049 */ 050@InterfaceAudience.Private 051@InterfaceStability.Evolving 052public class RegionServerRpcQuotaManager implements RpcQuotaManager, ConfigurationObserver { 053 private static final Logger LOG = LoggerFactory.getLogger(RegionServerRpcQuotaManager.class); 054 055 private final RegionServerServices rsServices; 056 057 private QuotaCache quotaCache = null; 058 private volatile boolean rpcThrottleEnabled; 059 // Storage for quota rpc throttle 060 private RpcThrottleStorage rpcThrottleStorage; 061 private final Supplier<Double> requestsPerSecondSupplier; 062 063 public RegionServerRpcQuotaManager(final RegionServerServices rsServices) { 064 this.rsServices = rsServices; 065 rpcThrottleStorage = 066 new RpcThrottleStorage(rsServices.getZooKeeper(), rsServices.getConfiguration()); 067 this.requestsPerSecondSupplier = Suppliers.memoizeWithExpiration( 068 () -> rsServices.getMetrics().getRegionServerWrapper().getRequestsPerSecond(), 1, 069 TimeUnit.MINUTES); 070 } 071 072 public void start(final RpcScheduler rpcScheduler) throws IOException { 073 if (!QuotaUtil.isQuotaEnabled(rsServices.getConfiguration())) { 074 LOG.info("Quota support disabled"); 075 return; 076 } 077 078 LOG.info("Initializing RPC quota support"); 079 080 // Initialize quota cache 081 quotaCache = new QuotaCache(rsServices); 082 quotaCache.start(); 083 rpcThrottleEnabled = rpcThrottleStorage.isRpcThrottleEnabled(); 084 LOG.info("Start rpc quota manager and rpc throttle enabled is {}", rpcThrottleEnabled); 085 } 086 087 public void stop() { 088 if (isQuotaEnabled()) { 089 quotaCache.stop("shutdown"); 090 } 091 } 092 093 public void reload() { 094 quotaCache.forceSynchronousCacheRefresh(); 095 } 096 097 @Override 098 public void onConfigurationChange(Configuration conf) { 099 reload(); 100 } 101 102 protected boolean isRpcThrottleEnabled() { 103 return rpcThrottleEnabled; 104 } 105 106 private boolean isQuotaEnabled() { 107 return quotaCache != null; 108 } 109 110 public void switchRpcThrottle(boolean enable) throws IOException { 111 if (isQuotaEnabled()) { 112 if (rpcThrottleEnabled != enable) { 113 boolean previousEnabled = rpcThrottleEnabled; 114 rpcThrottleEnabled = rpcThrottleStorage.isRpcThrottleEnabled(); 115 LOG.info("Switch rpc throttle from {} to {}", previousEnabled, rpcThrottleEnabled); 116 } else { 117 LOG.warn( 118 "Skip switch rpc throttle because previous value {} is the same as current value {}", 119 rpcThrottleEnabled, enable); 120 } 121 } else { 122 LOG.warn("Skip switch rpc throttle to {} because rpc quota is disabled", enable); 123 } 124 } 125 126 QuotaCache getQuotaCache() { 127 return quotaCache; 128 } 129 130 /** 131 * Returns the quota for an operation. 132 * @param ugi the user that is executing the operation 133 * @param table the table where the operation will be executed 134 * @return the OperationQuota 135 */ 136 public OperationQuota getQuota(final UserGroupInformation ugi, final TableName table, 137 final int blockSizeBytes) { 138 if (isQuotaEnabled() && !table.isSystemTable() && isRpcThrottleEnabled()) { 139 UserQuotaState userQuotaState = quotaCache.getUserQuotaState(ugi); 140 QuotaLimiter userLimiter = userQuotaState.getTableLimiter(table); 141 142 boolean useNoop = userLimiter.isBypass(); 143 if (userQuotaState.hasBypassGlobals()) { 144 if (LOG.isTraceEnabled()) { 145 LOG.trace("get quota for ugi=" + ugi + " table=" + table + " userLimiter=" + userLimiter); 146 } 147 if (!useNoop) { 148 return new DefaultOperationQuota(this.rsServices.getConfiguration(), blockSizeBytes, 149 requestsPerSecondSupplier.get(), userLimiter); 150 } 151 } else { 152 QuotaLimiter nsLimiter = quotaCache.getNamespaceLimiter(table.getNamespaceAsString()); 153 QuotaLimiter tableLimiter = quotaCache.getTableLimiter(table); 154 QuotaLimiter rsLimiter = 155 quotaCache.getRegionServerQuotaLimiter(QuotaTableUtil.QUOTA_REGION_SERVER_ROW_KEY); 156 useNoop &= tableLimiter.isBypass() && nsLimiter.isBypass() && rsLimiter.isBypass(); 157 boolean exceedThrottleQuotaEnabled = quotaCache.isExceedThrottleQuotaEnabled(); 158 if (LOG.isTraceEnabled()) { 159 LOG.trace("get quota for ugi=" + ugi + " table=" + table + " userLimiter=" + userLimiter 160 + " tableLimiter=" + tableLimiter + " nsLimiter=" + nsLimiter + " rsLimiter=" 161 + rsLimiter + " exceedThrottleQuotaEnabled=" + exceedThrottleQuotaEnabled); 162 } 163 if (!useNoop) { 164 if (exceedThrottleQuotaEnabled) { 165 return new ExceedOperationQuota(this.rsServices.getConfiguration(), blockSizeBytes, 166 requestsPerSecondSupplier.get(), rsLimiter, userLimiter, tableLimiter, nsLimiter); 167 } else { 168 return new DefaultOperationQuota(this.rsServices.getConfiguration(), blockSizeBytes, 169 requestsPerSecondSupplier.get(), userLimiter, tableLimiter, nsLimiter, rsLimiter); 170 } 171 } 172 } 173 } 174 return NoopOperationQuota.get(); 175 } 176 177 @Override 178 public OperationQuota checkScanQuota(final Region region, 179 final ClientProtos.ScanRequest scanRequest, long maxScannerResultSize, 180 long maxBlockBytesScanned, long prevBlockBytesScannedDifference) 181 throws IOException, RpcThrottlingException { 182 Optional<User> user = RpcServer.getRequestUser(); 183 UserGroupInformation ugi; 184 if (user.isPresent()) { 185 ugi = user.get().getUGI(); 186 } else { 187 ugi = User.getCurrent().getUGI(); 188 } 189 TableDescriptor tableDescriptor = region.getTableDescriptor(); 190 TableName table = tableDescriptor.getTableName(); 191 192 OperationQuota quota = getQuota(ugi, table, region.getMinBlockSizeBytes()); 193 try { 194 quota.checkScanQuota(scanRequest, maxScannerResultSize, maxBlockBytesScanned, 195 prevBlockBytesScannedDifference); 196 } catch (RpcThrottlingException e) { 197 LOG.debug("Throttling exception for user=" + ugi.getUserName() + " table=" + table + " scan=" 198 + scanRequest.getScannerId() + ": " + e.getMessage()); 199 throw e; 200 } 201 return quota; 202 } 203 204 @Override 205 public OperationQuota checkBatchQuota(final Region region, 206 final OperationQuota.OperationType type) throws IOException, RpcThrottlingException { 207 switch (type) { 208 case GET: 209 return this.checkBatchQuota(region, 0, 1, false); 210 case MUTATE: 211 return this.checkBatchQuota(region, 1, 0, false); 212 case CHECK_AND_MUTATE: 213 return this.checkBatchQuota(region, 1, 1, true); 214 } 215 throw new RuntimeException("Invalid operation type: " + type); 216 } 217 218 @Override 219 public OperationQuota checkBatchQuota(final Region region, 220 final List<ClientProtos.Action> actions, boolean hasCondition) 221 throws IOException, RpcThrottlingException { 222 int numWrites = 0; 223 int numReads = 0; 224 boolean isAtomic = false; 225 for (final ClientProtos.Action action : actions) { 226 if (action.hasMutation()) { 227 numWrites++; 228 OperationQuota.OperationType operationType = 229 QuotaUtil.getQuotaOperationType(action, hasCondition); 230 if (operationType == OperationQuota.OperationType.CHECK_AND_MUTATE) { 231 numReads++; 232 // If any mutations in this batch are atomic, we will count the entire batch as atomic. 233 // This is a conservative approach, but it is the best that we can do without knowing 234 // the block bytes scanned of each individual action. 235 isAtomic = true; 236 } 237 } else if (action.hasGet()) { 238 numReads++; 239 } 240 } 241 return checkBatchQuota(region, numWrites, numReads, isAtomic); 242 } 243 244 /** 245 * Check the quota for the current (rpc-context) user. Returns the OperationQuota used to get the 246 * available quota and to report the data/usage of the operation. 247 * @param region the region where the operation will be performed 248 * @param numWrites number of writes to perform 249 * @param numReads number of short-reads to perform 250 * @return the OperationQuota 251 * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded. 252 */ 253 @Override 254 public OperationQuota checkBatchQuota(final Region region, final int numWrites, 255 final int numReads, boolean isAtomic) throws IOException, RpcThrottlingException { 256 Optional<User> user = RpcServer.getRequestUser(); 257 UserGroupInformation ugi; 258 if (user.isPresent()) { 259 ugi = user.get().getUGI(); 260 } else { 261 ugi = User.getCurrent().getUGI(); 262 } 263 TableDescriptor tableDescriptor = region.getTableDescriptor(); 264 TableName table = tableDescriptor.getTableName(); 265 266 OperationQuota quota = getQuota(ugi, table, region.getMinBlockSizeBytes()); 267 try { 268 quota.checkBatchQuota(numWrites, numReads, isAtomic); 269 } catch (RpcThrottlingException e) { 270 LOG.debug("Throttling exception for user=" + ugi.getUserName() + " table=" + table 271 + " numWrites=" + numWrites + " numReads=" + numReads + ": " + e.getMessage()); 272 throw e; 273 } 274 return quota; 275 } 276}