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 org.apache.hadoop.conf.Configuration; 021import org.apache.yetus.audience.InterfaceAudience; 022import org.apache.yetus.audience.InterfaceStability; 023import org.slf4j.Logger; 024import org.slf4j.LoggerFactory; 025 026import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; 027 028/* 029 * Internal class used to check and consume quota if exceed throttle quota is enabled. Exceed 030 * throttle quota means, user can over consume user/namespace/table quota if region server has 031 * additional available quota because other users don't consume at the same time. 032 * 033 * There are some limits when enable exceed throttle quota: 034 * 1. Must set at least one read and one write region server throttle quota; 035 * 2. All region server throttle quotas must be in seconds time unit. Because once previous requests 036 * exceed their quota and consume region server quota, quota in other time units may be refilled in 037 * a long time, this may affect later requests. 038 */ 039@InterfaceAudience.Private 040@InterfaceStability.Evolving 041public class ExceedOperationQuota extends DefaultOperationQuota { 042 private static final Logger LOG = LoggerFactory.getLogger(ExceedOperationQuota.class); 043 private QuotaLimiter regionServerLimiter; 044 045 public ExceedOperationQuota(final Configuration conf, int blockSizeBytes, 046 final double requestsPerSecond, QuotaLimiter regionServerLimiter, 047 final QuotaLimiter... limiters) { 048 super(conf, blockSizeBytes, requestsPerSecond, limiters); 049 this.regionServerLimiter = regionServerLimiter; 050 } 051 052 @Override 053 public void checkBatchQuota(int numWrites, int numReads, boolean isAtomic) 054 throws RpcThrottlingException { 055 Runnable estimateQuota = () -> updateEstimateConsumeBatchQuota(numWrites, numReads); 056 CheckQuotaRunnable checkQuota = () -> super.checkBatchQuota(numWrites, numReads, isAtomic); 057 checkQuota(estimateQuota, checkQuota, numWrites, numReads, 0, isAtomic); 058 } 059 060 @Override 061 public void checkScanQuota(ClientProtos.ScanRequest scanRequest, long maxScannerResultSize, 062 long maxBlockBytesScanned, long prevBlockBytesScannedDifference) throws RpcThrottlingException { 063 Runnable estimateQuota = () -> updateEstimateConsumeScanQuota(scanRequest, maxScannerResultSize, 064 maxBlockBytesScanned, prevBlockBytesScannedDifference); 065 CheckQuotaRunnable checkQuota = () -> super.checkScanQuota(scanRequest, maxScannerResultSize, 066 maxBlockBytesScanned, prevBlockBytesScannedDifference); 067 checkQuota(estimateQuota, checkQuota, 0, 0, 1, false); 068 } 069 070 private void checkQuota(Runnable estimateQuota, CheckQuotaRunnable checkQuota, int numWrites, 071 int numReads, int numScans, boolean isAtomic) throws RpcThrottlingException { 072 if (regionServerLimiter.isBypass()) { 073 // If region server limiter is bypass, which means no region server quota is set, check and 074 // throttle by all other quotas. In this condition, exceed throttle quota will not work. 075 LOG.warn("Exceed throttle quota is enabled but no region server quotas found"); 076 checkQuota.run(); 077 } else { 078 // 1. Update estimate quota which will be consumed 079 estimateQuota.run(); 080 // 2. Check if region server limiter is enough. If not, throw RpcThrottlingException. 081 regionServerLimiter.checkQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, 082 writeCapacityUnitConsumed, readCapacityUnitConsumed, isAtomic, handlerUsageTimeConsumed); 083 // 3. Check if other limiters are enough. If not, exceed other limiters because region server 084 // limiter is enough. 085 boolean exceed = false; 086 try { 087 checkQuota.run(); 088 } catch (RpcThrottlingException e) { 089 exceed = true; 090 if (LOG.isDebugEnabled()) { 091 LOG.debug("Read/Write requests num exceeds quota: writes:{} reads:{}, scans:{}, " 092 + "try use region server quota", numWrites, numReads, numScans); 093 } 094 } 095 // 4. Region server limiter is enough and grab estimated consume quota. 096 readAvailable = Math.max(readAvailable, regionServerLimiter.getReadAvailable()); 097 regionServerLimiter.grabQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, 098 writeCapacityUnitConsumed, writeCapacityUnitConsumed, isAtomic, handlerUsageTimeConsumed); 099 if (exceed) { 100 // 5. Other quota limiter is exceeded and has not been grabbed (because throw 101 // RpcThrottlingException in Step 3), so grab it. 102 for (final QuotaLimiter limiter : limiters) { 103 limiter.grabQuota(numWrites, writeConsumed, numReads + numScans, readConsumed, 104 writeCapacityUnitConsumed, writeCapacityUnitConsumed, isAtomic, 0L); 105 } 106 } 107 } 108 } 109 110 @Override 111 public void close() { 112 super.close(); 113 if (writeDiff != 0) { 114 regionServerLimiter.consumeWrite(writeDiff, writeCapacityUnitDiff, false); 115 } 116 if (readDiff != 0) { 117 regionServerLimiter.consumeRead(readDiff, readCapacityUnitDiff, false); 118 } 119 if (handlerUsageTimeDiff != 0) { 120 regionServerLimiter.consumeTime(handlerUsageTimeDiff); 121 } 122 } 123 124 private interface CheckQuotaRunnable { 125 void run() throws RpcThrottlingException; 126 } 127}