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
200      rsServices.getMetrics().recordThrottleException(e.getType(), ugi.getShortUserName(),
201        table.getNameAsString());
202
203      throw e;
204    }
205    return quota;
206  }
207
208  @Override
209  public OperationQuota checkBatchQuota(final Region region,
210    final OperationQuota.OperationType type) throws IOException, RpcThrottlingException {
211    switch (type) {
212      case GET:
213        return this.checkBatchQuota(region, 0, 1, false);
214      case MUTATE:
215        return this.checkBatchQuota(region, 1, 0, false);
216      case CHECK_AND_MUTATE:
217        return this.checkBatchQuota(region, 1, 1, true);
218    }
219    throw new RuntimeException("Invalid operation type: " + type);
220  }
221
222  @Override
223  public OperationQuota checkBatchQuota(final Region region,
224    final List<ClientProtos.Action> actions, boolean hasCondition)
225    throws IOException, RpcThrottlingException {
226    int numWrites = 0;
227    int numReads = 0;
228    boolean isAtomic = false;
229    for (final ClientProtos.Action action : actions) {
230      if (action.hasMutation()) {
231        numWrites++;
232        OperationQuota.OperationType operationType =
233          QuotaUtil.getQuotaOperationType(action, hasCondition);
234        if (operationType == OperationQuota.OperationType.CHECK_AND_MUTATE) {
235          numReads++;
236          // If any mutations in this batch are atomic, we will count the entire batch as atomic.
237          // This is a conservative approach, but it is the best that we can do without knowing
238          // the block bytes scanned of each individual action.
239          isAtomic = true;
240        }
241      } else if (action.hasGet()) {
242        numReads++;
243      }
244    }
245    return checkBatchQuota(region, numWrites, numReads, isAtomic);
246  }
247
248  /**
249   * Check the quota for the current (rpc-context) user. Returns the OperationQuota used to get the
250   * available quota and to report the data/usage of the operation.
251   * @param region    the region where the operation will be performed
252   * @param numWrites number of writes to perform
253   * @param numReads  number of short-reads to perform
254   * @return the OperationQuota
255   * @throws RpcThrottlingException if the operation cannot be executed due to quota exceeded.
256   */
257  @Override
258  public OperationQuota checkBatchQuota(final Region region, final int numWrites,
259    final int numReads, boolean isAtomic) throws IOException, RpcThrottlingException {
260    Optional<User> user = RpcServer.getRequestUser();
261    UserGroupInformation ugi;
262    if (user.isPresent()) {
263      ugi = user.get().getUGI();
264    } else {
265      ugi = User.getCurrent().getUGI();
266    }
267    TableDescriptor tableDescriptor = region.getTableDescriptor();
268    TableName table = tableDescriptor.getTableName();
269
270    OperationQuota quota = getQuota(ugi, table, region.getMinBlockSizeBytes());
271    try {
272      quota.checkBatchQuota(numWrites, numReads, isAtomic);
273    } catch (RpcThrottlingException e) {
274      LOG.debug("Throttling exception for user=" + ugi.getUserName() + " table=" + table
275        + " numWrites=" + numWrites + " numReads=" + numReads + ": " + e.getMessage());
276
277      rsServices.getMetrics().recordThrottleException(e.getType(), ugi.getShortUserName(),
278        table.getNameAsString());
279
280      throw e;
281    }
282    return quota;
283  }
284}