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