View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
3    * agreements. See the NOTICE file distributed with this work for additional information regarding
4    * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
5    * "License"); you may not use this file except in compliance with the License. You may obtain a
6    * copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable
7    * law or agreed to in writing, software distributed under the License is distributed on an "AS IS"
8    * BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License
9    * for the specific language governing permissions and limitations under the License.
10   */
11  
12  package org.apache.hadoop.hbase.quotas;
13  
14  import static org.junit.Assert.assertEquals;
15  
16  import java.util.concurrent.TimeUnit;
17  
18  import org.apache.commons.logging.Log;
19  import org.apache.commons.logging.LogFactory;
20  import org.apache.hadoop.hbase.HBaseTestingUtility;
21  import org.apache.hadoop.hbase.HConstants;
22  import org.apache.hadoop.hbase.TableName;
23  import org.apache.hadoop.hbase.client.Admin;
24  import org.apache.hadoop.hbase.client.Get;
25  import org.apache.hadoop.hbase.client.HTable;
26  import org.apache.hadoop.hbase.client.Put;
27  import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
28  import org.apache.hadoop.hbase.security.User;
29  import org.apache.hadoop.hbase.testclassification.MediumTests;
30  import org.apache.hadoop.hbase.util.Bytes;
31  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
32  import org.apache.hadoop.hbase.util.EnvironmentEdgeManagerTestHelper;
33  import org.apache.hadoop.hbase.util.IncrementingEnvironmentEdge;
34  import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
35  import org.junit.After;
36  import org.junit.AfterClass;
37  import org.junit.BeforeClass;
38  import org.junit.Test;
39  import org.junit.experimental.categories.Category;
40  
41  @Category({ MediumTests.class })
42  public class TestQuotaThrottle {
43    final Log LOG = LogFactory.getLog(getClass());
44  
45    private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
46  
47    private final static byte[] FAMILY = Bytes.toBytes("cf");
48    private final static byte[] QUALIFIER = Bytes.toBytes("q");
49  
50    private final static TableName[] TABLE_NAMES = new TableName[] {
51        TableName.valueOf("TestQuotaAdmin0"), TableName.valueOf("TestQuotaAdmin1"),
52        TableName.valueOf("TestQuotaAdmin2") };
53  
54    private static HTable[] tables;
55  
56    @BeforeClass
57    public static void setUpBeforeClass() throws Exception {
58      TEST_UTIL.getConfiguration().setBoolean(QuotaUtil.QUOTA_CONF_KEY, true);
59      TEST_UTIL.getConfiguration().setInt("hbase.hstore.compactionThreshold", 10);
60      TEST_UTIL.getConfiguration().setInt("hbase.regionserver.msginterval", 100);
61      TEST_UTIL.getConfiguration().setInt("hbase.client.pause", 250);
62      TEST_UTIL.getConfiguration().setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 6);
63      TEST_UTIL.getConfiguration().setBoolean("hbase.master.enabletable.roundrobin", true);
64      TEST_UTIL.startMiniCluster(1);
65      TEST_UTIL.waitTableAvailable(QuotaTableUtil.QUOTA_TABLE_NAME);
66      QuotaCache.setTEST_FORCE_REFRESH(true);
67  
68      tables = new HTable[TABLE_NAMES.length];
69      for (int i = 0; i < TABLE_NAMES.length; ++i) {
70        tables[i] = TEST_UTIL.createTable(TABLE_NAMES[i], FAMILY);
71      }
72    }
73  
74    @AfterClass
75    public static void tearDownAfterClass() throws Exception {
76      for (int i = 0; i < tables.length; ++i) {
77        if (tables[i] != null) {
78          tables[i].close();
79          TEST_UTIL.deleteTable(TABLE_NAMES[i]);
80        }
81      }
82  
83      TEST_UTIL.shutdownMiniCluster();
84    }
85  
86    @After
87    public void tearDown() throws Exception {
88      for (RegionServerThread rst : TEST_UTIL.getMiniHBaseCluster().getRegionServerThreads()) {
89        RegionServerQuotaManager quotaManager = rst.getRegionServer().getRegionServerQuotaManager();
90        QuotaCache quotaCache = quotaManager.getQuotaCache();
91        quotaCache.getNamespaceQuotaCache().clear();
92        quotaCache.getTableQuotaCache().clear();
93        quotaCache.getUserQuotaCache().clear();
94      }
95    }
96  
97    @Test(timeout = 60000)
98    public void testUserGlobalThrottle() throws Exception {
99      final Admin admin = TEST_UTIL.getHBaseAdmin();
100     final String userName = User.getCurrent().getShortName();
101 
102     // Add 6req/min limit
103     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_NUMBER, 6,
104       TimeUnit.MINUTES));
105     triggerUserCacheRefresh(false, TABLE_NAMES);
106 
107     // should execute at max 6 requests
108     assertEquals(6, doPuts(100, tables));
109 
110     // wait a minute and you should get other 6 requests executed
111     waitMinuteQuota();
112     assertEquals(6, doPuts(100, tables));
113 
114     // Remove all the limits
115     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName));
116     triggerUserCacheRefresh(true, TABLE_NAMES);
117     assertEquals(60, doPuts(60, tables));
118     assertEquals(60, doGets(60, tables));
119   }
120 
121   @Test(timeout = 60000)
122   public void testUserTableThrottle() throws Exception {
123     final Admin admin = TEST_UTIL.getHBaseAdmin();
124     final String userName = User.getCurrent().getShortName();
125 
126     // Add 6req/min limit
127     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, TABLE_NAMES[0],
128       ThrottleType.REQUEST_NUMBER, 6, TimeUnit.MINUTES));
129     triggerUserCacheRefresh(false, TABLE_NAMES[0]);
130 
131     // should execute at max 6 requests on tables[0] and have no limit on tables[1]
132     assertEquals(6, doPuts(100, tables[0]));
133     assertEquals(30, doPuts(30, tables[1]));
134 
135     // wait a minute and you should get other 6 requests executed
136     waitMinuteQuota();
137     assertEquals(6, doPuts(100, tables[0]));
138 
139     // Remove all the limits
140     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName, TABLE_NAMES[0]));
141     triggerUserCacheRefresh(true, TABLE_NAMES);
142     assertEquals(60, doPuts(60, tables));
143     assertEquals(60, doGets(60, tables));
144   }
145 
146   @Test(timeout = 60000)
147   public void testUserNamespaceThrottle() throws Exception {
148     final Admin admin = TEST_UTIL.getHBaseAdmin();
149     final String userName = User.getCurrent().getShortName();
150     final String NAMESPACE = "default";
151 
152     // Add 6req/min limit
153     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, NAMESPACE,
154       ThrottleType.REQUEST_NUMBER, 6, TimeUnit.MINUTES));
155     triggerUserCacheRefresh(false, TABLE_NAMES[0]);
156 
157     // should execute at max 6 requests on tables[0] and have no limit on tables[1]
158     assertEquals(6, doPuts(100, tables[0]));
159 
160     // wait a minute and you should get other 6 requests executed
161     waitMinuteQuota();
162     assertEquals(6, doPuts(100, tables[1]));
163 
164     // Remove all the limits
165     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName, NAMESPACE));
166     triggerUserCacheRefresh(true, TABLE_NAMES);
167     assertEquals(60, doPuts(60, tables));
168     assertEquals(60, doGets(60, tables));
169   }
170 
171   @Test(timeout = 60000)
172   public void testTableGlobalThrottle() throws Exception {
173     final Admin admin = TEST_UTIL.getHBaseAdmin();
174 
175     // Add 6req/min limit
176     admin.setQuota(QuotaSettingsFactory.throttleTable(TABLE_NAMES[0], ThrottleType.REQUEST_NUMBER,
177       6, TimeUnit.MINUTES));
178     triggerTableCacheRefresh(false, TABLE_NAMES[0]);
179 
180     // should execute at max 6 requests
181     assertEquals(6, doPuts(100, tables[0]));
182     // should have no limits
183     assertEquals(30, doPuts(30, tables[1]));
184 
185     // wait a minute and you should get other 6 requests executed
186     waitMinuteQuota();
187     assertEquals(6, doPuts(100, tables[0]));
188 
189     // Remove all the limits
190     admin.setQuota(QuotaSettingsFactory.unthrottleTable(TABLE_NAMES[0]));
191     triggerTableCacheRefresh(true, TABLE_NAMES[0]);
192     assertEquals(80, doGets(80, tables[0], tables[1]));
193   }
194 
195   @Test(timeout = 60000)
196   public void testNamespaceGlobalThrottle() throws Exception {
197     final Admin admin = TEST_UTIL.getHBaseAdmin();
198     final String NAMESPACE = "default";
199 
200     // Add 6req/min limit
201     admin.setQuota(QuotaSettingsFactory.throttleNamespace(NAMESPACE, ThrottleType.REQUEST_NUMBER,
202       6, TimeUnit.MINUTES));
203     triggerNamespaceCacheRefresh(false, TABLE_NAMES[0]);
204 
205     // should execute at max 6 requests
206     assertEquals(6, doPuts(100, tables[0]));
207 
208     // wait a minute and you should get other 6 requests executed
209     waitMinuteQuota();
210     assertEquals(6, doPuts(100, tables[1]));
211 
212     admin.setQuota(QuotaSettingsFactory.unthrottleNamespace(NAMESPACE));
213     triggerNamespaceCacheRefresh(true, TABLE_NAMES[0]);
214     assertEquals(40, doPuts(40, tables[0]));
215   }
216 
217   @Test(timeout = 60000)
218   public void testUserAndTableThrottle() throws Exception {
219     final Admin admin = TEST_UTIL.getHBaseAdmin();
220     final String userName = User.getCurrent().getShortName();
221 
222     // Add 6req/min limit for the user on tables[0]
223     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, TABLE_NAMES[0],
224       ThrottleType.REQUEST_NUMBER, 6, TimeUnit.MINUTES));
225     triggerUserCacheRefresh(false, TABLE_NAMES[0]);
226     // Add 12req/min limit for the user
227     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_NUMBER, 12,
228       TimeUnit.MINUTES));
229     triggerUserCacheRefresh(false, TABLE_NAMES[1], TABLE_NAMES[2]);
230     // Add 8req/min limit for the tables[1]
231     admin.setQuota(QuotaSettingsFactory.throttleTable(TABLE_NAMES[1], ThrottleType.REQUEST_NUMBER,
232       8, TimeUnit.MINUTES));
233     triggerTableCacheRefresh(false, TABLE_NAMES[1]);
234     // Add a lower table level throttle on tables[0]
235     admin.setQuota(QuotaSettingsFactory.throttleTable(TABLE_NAMES[0], ThrottleType.REQUEST_NUMBER,
236       3, TimeUnit.MINUTES));
237     triggerTableCacheRefresh(false, TABLE_NAMES[0]);
238 
239     // should execute at max 12 requests
240     assertEquals(12, doGets(100, tables[2]));
241 
242     // should execute at max 8 requests
243     waitMinuteQuota();
244     assertEquals(8, doGets(100, tables[1]));
245 
246     // should execute at max 3 requests
247     waitMinuteQuota();
248     assertEquals(3, doPuts(100, tables[0]));
249 
250     // Remove all the throttling rules
251     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName, TABLE_NAMES[0]));
252     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName));
253     triggerUserCacheRefresh(true, TABLE_NAMES[0], TABLE_NAMES[1]);
254 
255     admin.setQuota(QuotaSettingsFactory.unthrottleTable(TABLE_NAMES[1]));
256     triggerTableCacheRefresh(true, TABLE_NAMES[1]);
257     waitMinuteQuota();
258     assertEquals(40, doGets(40, tables[1]));
259 
260     admin.setQuota(QuotaSettingsFactory.unthrottleTable(TABLE_NAMES[0]));
261     triggerTableCacheRefresh(true, TABLE_NAMES[0]);
262     waitMinuteQuota();
263     assertEquals(40, doGets(40, tables[0]));
264   }
265 
266   @Test(timeout = 60000)
267   public void testUserGlobalBypassThrottle() throws Exception {
268     final Admin admin = TEST_UTIL.getHBaseAdmin();
269     final String userName = User.getCurrent().getShortName();
270     final String NAMESPACE = "default";
271 
272     // Add 6req/min limit for tables[0]
273     admin.setQuota(QuotaSettingsFactory.throttleTable(TABLE_NAMES[0], ThrottleType.REQUEST_NUMBER,
274       6, TimeUnit.MINUTES));
275     triggerTableCacheRefresh(false, TABLE_NAMES[0]);
276     // Add 13req/min limit for the user
277     admin.setQuota(QuotaSettingsFactory.throttleNamespace(NAMESPACE, ThrottleType.REQUEST_NUMBER,
278       13, TimeUnit.MINUTES));
279     triggerNamespaceCacheRefresh(false, TABLE_NAMES[1]);
280 
281     // should execute at max 6 requests on table[0] and (13 - 6) on table[1]
282     assertEquals(6, doPuts(100, tables[0]));
283     assertEquals(7, doGets(100, tables[1]));
284     waitMinuteQuota();
285 
286     // Set the global bypass for the user
287     admin.setQuota(QuotaSettingsFactory.bypassGlobals(userName, true));
288     admin.setQuota(QuotaSettingsFactory.throttleUser(userName, TABLE_NAMES[2],
289       ThrottleType.REQUEST_NUMBER, 6, TimeUnit.MINUTES));
290     triggerUserCacheRefresh(false, TABLE_NAMES[2]);
291     assertEquals(30, doGets(30, tables[0]));
292     assertEquals(30, doGets(30, tables[1]));
293     waitMinuteQuota();
294 
295     // Remove the global bypass
296     // should execute at max 6 requests on table[0] and (13 - 6) on table[1]
297     admin.setQuota(QuotaSettingsFactory.bypassGlobals(userName, false));
298     admin.setQuota(QuotaSettingsFactory.unthrottleUser(userName, TABLE_NAMES[2]));
299     triggerUserCacheRefresh(true, TABLE_NAMES[2]);
300     assertEquals(6, doPuts(100, tables[0]));
301     assertEquals(7, doGets(100, tables[1]));
302 
303     // unset throttle
304     admin.setQuota(QuotaSettingsFactory.unthrottleTable(TABLE_NAMES[0]));
305     admin.setQuota(QuotaSettingsFactory.unthrottleNamespace(NAMESPACE));
306     waitMinuteQuota();
307     triggerTableCacheRefresh(true, TABLE_NAMES[0]);
308     triggerNamespaceCacheRefresh(true, TABLE_NAMES[1]);
309     assertEquals(30, doGets(30, tables[0]));
310     assertEquals(30, doGets(30, tables[1]));
311   }
312 
313   private int doPuts(int maxOps, final HTable... tables) throws Exception {
314     int count = 0;
315     try {
316       while (count < maxOps) {
317         Put put = new Put(Bytes.toBytes("row-" + count));
318         put.addColumn(FAMILY, QUALIFIER, Bytes.toBytes("data-" + count));
319         for (final HTable table : tables) {
320           table.put(put);
321         }
322         count += tables.length;
323       }
324     } catch (RetriesExhaustedWithDetailsException e) {
325       for (Throwable t : e.getCauses()) {
326         if (!(t instanceof ThrottlingException)) {
327           throw e;
328         }
329       }
330       LOG.error("put failed after nRetries=" + count, e);
331     }
332     return count;
333   }
334 
335   private long doGets(int maxOps, final HTable... tables) throws Exception {
336     int count = 0;
337     try {
338       while (count < maxOps) {
339         Get get = new Get(Bytes.toBytes("row-" + count));
340         for (final HTable table : tables) {
341           table.get(get);
342         }
343         count += tables.length;
344       }
345     } catch (ThrottlingException e) {
346       LOG.error("get failed after nRetries=" + count, e);
347     }
348     return count;
349   }
350 
351   private void triggerUserCacheRefresh(boolean bypass, TableName... tables) throws Exception {
352     triggerCacheRefresh(bypass, true, false, false, tables);
353   }
354 
355   private void triggerTableCacheRefresh(boolean bypass, TableName... tables) throws Exception {
356     triggerCacheRefresh(bypass, false, true, false, tables);
357   }
358 
359   private void triggerNamespaceCacheRefresh(boolean bypass, TableName... tables) throws Exception {
360     triggerCacheRefresh(bypass, false, false, true, tables);
361   }
362 
363   private void triggerCacheRefresh(boolean bypass, boolean userLimiter, boolean tableLimiter,
364       boolean nsLimiter, final TableName... tables) throws Exception {
365     for (RegionServerThread rst : TEST_UTIL.getMiniHBaseCluster().getRegionServerThreads()) {
366       RegionServerQuotaManager quotaManager = rst.getRegionServer().getRegionServerQuotaManager();
367       QuotaCache quotaCache = quotaManager.getQuotaCache();
368 
369       quotaCache.triggerCacheRefresh();
370       Thread.sleep(250);
371 
372       for (TableName table : tables) {
373         quotaCache.getTableLimiter(table);
374       }
375 
376       boolean isUpdated = false;
377       while (!isUpdated) {
378         isUpdated = true;
379         for (TableName table : tables) {
380           boolean isBypass = true;
381           if (userLimiter) {
382             isBypass &= quotaCache.getUserLimiter(User.getCurrent().getUGI(), table).isBypass();
383           }
384           if (tableLimiter) {
385             isBypass &= quotaCache.getTableLimiter(table).isBypass();
386           }
387           if (nsLimiter) {
388             isBypass &= quotaCache.getNamespaceLimiter(table.getNamespaceAsString()).isBypass();
389           }
390           if (isBypass != bypass) {
391             isUpdated = false;
392             Thread.sleep(250);
393             break;
394           }
395         }
396       }
397 
398       LOG.debug("QuotaCache");
399       LOG.debug(quotaCache.getNamespaceQuotaCache());
400       LOG.debug(quotaCache.getTableQuotaCache());
401       LOG.debug(quotaCache.getUserQuotaCache());
402     }
403   }
404 
405   private void waitMinuteQuota() {
406     EnvironmentEdgeManagerTestHelper.injectEdge(new IncrementingEnvironmentEdge(
407         EnvironmentEdgeManager.currentTime() + 70000));
408   }
409 }