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 */ 018 019package org.apache.hadoop.hbase.quotas; 020 021import java.io.IOException; 022import java.util.List; 023import java.util.Optional; 024 025import org.apache.hadoop.hbase.TableName; 026import org.apache.hadoop.hbase.ipc.RpcScheduler; 027import org.apache.hadoop.hbase.ipc.RpcServer; 028import org.apache.hadoop.hbase.regionserver.Region; 029import org.apache.hadoop.hbase.regionserver.RegionServerServices; 030import org.apache.hadoop.hbase.security.User; 031import org.apache.hadoop.security.UserGroupInformation; 032import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; 033import org.apache.yetus.audience.InterfaceAudience; 034import org.apache.yetus.audience.InterfaceStability; 035import org.slf4j.Logger; 036import org.slf4j.LoggerFactory; 037 038import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; 039 040/** 041 * Region Server Quota Manager. 042 * It is responsible to provide access to the quota information of each user/table. 043 * 044 * The direct user of this class is the RegionServer that will get and check the 045 * user/table quota for each operation (put, get, scan). 046 * For system tables and user/table with a quota specified, the quota check will be a noop. 047 */ 048@InterfaceAudience.Private 049@InterfaceStability.Evolving 050public class RegionServerRpcQuotaManager { 051 private static final Logger LOG = LoggerFactory.getLogger(RegionServerRpcQuotaManager.class); 052 053 private final RegionServerServices rsServices; 054 055 private QuotaCache quotaCache = null; 056 private volatile boolean rpcThrottleEnabled; 057 // Storage for quota rpc throttle 058 private RpcThrottleStorage rpcThrottleStorage; 059 060 public RegionServerRpcQuotaManager(final RegionServerServices rsServices) { 061 this.rsServices = rsServices; 062 rpcThrottleStorage = 063 new RpcThrottleStorage(rsServices.getZooKeeper(), rsServices.getConfiguration()); 064 } 065 066 public void start(final RpcScheduler rpcScheduler) throws IOException { 067 if (!QuotaUtil.isQuotaEnabled(rsServices.getConfiguration())) { 068 LOG.info("Quota support disabled"); 069 return; 070 } 071 072 LOG.info("Initializing RPC quota support"); 073 074 // Initialize quota cache 075 quotaCache = new QuotaCache(rsServices); 076 quotaCache.start(); 077 rpcThrottleEnabled = rpcThrottleStorage.isRpcThrottleEnabled(); 078 LOG.info("Start rpc quota manager and rpc throttle enabled is {}", rpcThrottleEnabled); 079 } 080 081 public void stop() { 082 if (isQuotaEnabled()) { 083 quotaCache.stop("shutdown"); 084 } 085 } 086 087 protected boolean isRpcThrottleEnabled() { 088 return rpcThrottleEnabled; 089 } 090 091 private boolean isQuotaEnabled() { 092 return quotaCache != null; 093 } 094 095 public void switchRpcThrottle(boolean enable) throws IOException { 096 if (isQuotaEnabled()) { 097 if (rpcThrottleEnabled != enable) { 098 boolean previousEnabled = rpcThrottleEnabled; 099 rpcThrottleEnabled = rpcThrottleStorage.isRpcThrottleEnabled(); 100 LOG.info("Switch rpc throttle from {} to {}", previousEnabled, rpcThrottleEnabled); 101 } else { 102 LOG.warn( 103 "Skip switch rpc throttle because previous value {} is the same as current value {}", 104 rpcThrottleEnabled, enable); 105 } 106 } else { 107 LOG.warn("Skip switch rpc throttle to {} because rpc quota is disabled", enable); 108 } 109 } 110 111 QuotaCache getQuotaCache() { 112 return quotaCache; 113 } 114 115 /** 116 * Returns the quota for an operation. 117 * 118 * @param ugi the user that is executing the operation 119 * @param table the table where the operation will be executed 120 * @return the OperationQuota 121 */ 122 public OperationQuota getQuota(final UserGroupInformation ugi, final TableName table) { 123 if (isQuotaEnabled() && !table.isSystemTable() && isRpcThrottleEnabled()) { 124 UserQuotaState userQuotaState = quotaCache.getUserQuotaState(ugi); 125 QuotaLimiter userLimiter = userQuotaState.getTableLimiter(table); 126 boolean useNoop = userLimiter.isBypass(); 127 if (userQuotaState.hasBypassGlobals()) { 128 if (LOG.isTraceEnabled()) { 129 LOG.trace("get quota for ugi=" + ugi + " table=" + table + " userLimiter=" + userLimiter); 130 } 131 if (!useNoop) { 132 return new DefaultOperationQuota(this.rsServices.getConfiguration(), userLimiter); 133 } 134 } else { 135 QuotaLimiter nsLimiter = quotaCache.getNamespaceLimiter(table.getNamespaceAsString()); 136 QuotaLimiter tableLimiter = quotaCache.getTableLimiter(table); 137 QuotaLimiter rsLimiter = quotaCache 138 .getRegionServerQuotaLimiter(QuotaTableUtil.QUOTA_REGION_SERVER_ROW_KEY); 139 useNoop &= tableLimiter.isBypass() && nsLimiter.isBypass() && rsLimiter.isBypass(); 140 boolean exceedThrottleQuotaEnabled = quotaCache.isExceedThrottleQuotaEnabled(); 141 if (LOG.isTraceEnabled()) { 142 LOG.trace("get quota for ugi=" + ugi + " table=" + table + " userLimiter=" + userLimiter 143 + " tableLimiter=" + tableLimiter + " nsLimiter=" + nsLimiter + " rsLimiter=" 144 + rsLimiter + " exceedThrottleQuotaEnabled=" + exceedThrottleQuotaEnabled); 145 } 146 if (!useNoop) { 147 if (exceedThrottleQuotaEnabled) { 148 return new ExceedOperationQuota(this.rsServices.getConfiguration(), rsLimiter, 149 userLimiter, tableLimiter, nsLimiter); 150 } else { 151 return new DefaultOperationQuota(this.rsServices.getConfiguration(), userLimiter, 152 tableLimiter, nsLimiter, rsLimiter); 153 } 154 } 155 } 156 } 157 return NoopOperationQuota.get(); 158 } 159 160 /** 161 * Check the quota for the current (rpc-context) user. 162 * Returns the OperationQuota used to get the available quota and 163 * to report the data/usage of the operation. 164 * @param region the region where the operation will be performed 165 * @param type the operation type 166 * @return the OperationQuota 167 * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded. 168 */ 169 public OperationQuota checkQuota(final Region region, 170 final OperationQuota.OperationType type) throws IOException, RpcThrottlingException { 171 switch (type) { 172 case SCAN: return checkQuota(region, 0, 0, 1); 173 case GET: return checkQuota(region, 0, 1, 0); 174 case MUTATE: return checkQuota(region, 1, 0, 0); 175 } 176 throw new RuntimeException("Invalid operation type: " + type); 177 } 178 179 /** 180 * Check the quota for the current (rpc-context) user. 181 * Returns the OperationQuota used to get the available quota and 182 * to report the data/usage of the operation. 183 * @param region the region where the operation will be performed 184 * @param actions the "multi" actions to perform 185 * @return the OperationQuota 186 * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded. 187 */ 188 public OperationQuota checkQuota(final Region region, 189 final List<ClientProtos.Action> actions) throws IOException, RpcThrottlingException { 190 int numWrites = 0; 191 int numReads = 0; 192 for (final ClientProtos.Action action: actions) { 193 if (action.hasMutation()) { 194 numWrites++; 195 } else if (action.hasGet()) { 196 numReads++; 197 } 198 } 199 return checkQuota(region, numWrites, numReads, 0); 200 } 201 202 /** 203 * Check the quota for the current (rpc-context) user. 204 * Returns the OperationQuota used to get the available quota and 205 * to report the data/usage of the operation. 206 * @param region the region where the operation will be performed 207 * @param numWrites number of writes to perform 208 * @param numReads number of short-reads to perform 209 * @param numScans number of scan to perform 210 * @return the OperationQuota 211 * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded. 212 */ 213 private OperationQuota checkQuota(final Region region, 214 final int numWrites, final int numReads, final int numScans) 215 throws IOException, RpcThrottlingException { 216 Optional<User> user = RpcServer.getRequestUser(); 217 UserGroupInformation ugi; 218 if (user.isPresent()) { 219 ugi = user.get().getUGI(); 220 } else { 221 ugi = User.getCurrent().getUGI(); 222 } 223 TableName table = region.getTableDescriptor().getTableName(); 224 225 OperationQuota quota = getQuota(ugi, table); 226 try { 227 quota.checkQuota(numWrites, numReads, numScans); 228 } catch (RpcThrottlingException e) { 229 LOG.debug("Throttling exception for user=" + ugi.getUserName() + 230 " table=" + table + " numWrites=" + numWrites + 231 " numReads=" + numReads + " numScans=" + numScans + 232 ": " + e.getMessage()); 233 throw e; 234 } 235 return quota; 236 } 237}