001/* 002 * 003 * Licensed to the Apache Software Foundation (ASF) under one 004 * or more contributor license agreements. See the NOTICE file 005 * distributed with this work for additional information 006 * regarding copyright ownership. The ASF licenses this file 007 * to you under the Apache License, Version 2.0 (the 008 * "License"); you may not use this file except in compliance 009 * with the License. You may obtain a copy of the License at 010 * 011 * http://www.apache.org/licenses/LICENSE-2.0 012 * 013 * Unless required by applicable law or agreed to in writing, software 014 * distributed under the License is distributed on an "AS IS" BASIS, 015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 016 * See the License for the specific language governing permissions and 017 * limitations under the License. 018 */ 019 020package org.apache.hadoop.hbase.regionserver.slowlog; 021 022import java.io.IOException; 023import java.lang.reflect.Field; 024import java.util.concurrent.CompletableFuture; 025import java.util.concurrent.TimeUnit; 026import org.apache.hadoop.conf.Configuration; 027import org.apache.hadoop.hbase.HBaseClassTestRule; 028import org.apache.hadoop.hbase.HBaseTestingUtility; 029import org.apache.hadoop.hbase.HConstants; 030import org.apache.hadoop.hbase.MetaTableAccessor; 031import org.apache.hadoop.hbase.client.Connection; 032import org.apache.hadoop.hbase.client.Result; 033import org.apache.hadoop.hbase.client.ResultScanner; 034import org.apache.hadoop.hbase.client.Scan; 035import org.apache.hadoop.hbase.client.Table; 036import org.apache.hadoop.hbase.regionserver.HRegionServer; 037import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos; 038import org.apache.hadoop.hbase.slowlog.SlowLogTableAccessor; 039import org.apache.hadoop.hbase.testclassification.MasterTests; 040import org.apache.hadoop.hbase.testclassification.MediumTests; 041import org.apache.hbase.thirdparty.com.google.common.util.concurrent.Uninterruptibles; 042import org.junit.AfterClass; 043import org.junit.Assert; 044import org.junit.Before; 045import org.junit.BeforeClass; 046import org.junit.ClassRule; 047import org.junit.Test; 048import org.junit.experimental.categories.Category; 049import org.slf4j.Logger; 050import org.slf4j.LoggerFactory; 051 052/** 053 * Tests for SlowLog System Table 054 */ 055@Category({ MasterTests.class, MediumTests.class }) 056public class TestSlowLogAccessor { 057 058 @ClassRule 059 public static final HBaseClassTestRule CLASS_RULE = 060 HBaseClassTestRule.forClass(TestSlowLogAccessor.class); 061 062 private static final Logger LOG = LoggerFactory.getLogger(TestSlowLogRecorder.class); 063 064 private static final HBaseTestingUtility HBASE_TESTING_UTILITY = new HBaseTestingUtility(); 065 066 private SlowLogRecorder slowLogRecorder; 067 068 @BeforeClass 069 public static void setup() throws Exception { 070 try { 071 HBASE_TESTING_UTILITY.shutdownMiniHBaseCluster(); 072 } catch (IOException e) { 073 LOG.debug("No worries."); 074 } 075 Configuration conf = HBASE_TESTING_UTILITY.getConfiguration(); 076 conf.setBoolean(HConstants.SLOW_LOG_BUFFER_ENABLED_KEY, true); 077 conf.setBoolean(HConstants.SLOW_LOG_SYS_TABLE_ENABLED_KEY, true); 078 conf.setInt("hbase.slowlog.systable.chore.duration", 900); 079 conf.setInt("hbase.regionserver.slowlog.ringbuffer.size", 50000); 080 HBASE_TESTING_UTILITY.startMiniCluster(); 081 } 082 083 @AfterClass 084 public static void teardown() throws Exception { 085 HBASE_TESTING_UTILITY.shutdownMiniHBaseCluster(); 086 } 087 088 @Before 089 public void setUp() throws Exception { 090 HRegionServer hRegionServer = HBASE_TESTING_UTILITY.getMiniHBaseCluster().getRegionServer(0); 091 Field slowLogRecorder = HRegionServer.class.getDeclaredField("slowLogRecorder"); 092 slowLogRecorder.setAccessible(true); 093 this.slowLogRecorder = (SlowLogRecorder) slowLogRecorder.get(hRegionServer); 094 } 095 096 @Test 097 public void testSlowLogRecords() throws Exception { 098 099 AdminProtos.SlowLogResponseRequest request = 100 AdminProtos.SlowLogResponseRequest.newBuilder().setLimit(15).build(); 101 102 slowLogRecorder.clearSlowLogPayloads(); 103 Assert.assertEquals(slowLogRecorder.getSlowLogPayloads(request).size(), 0); 104 105 int i = 0; 106 107 Connection connection = waitForSlowLogTableCreation(); 108 // add 5 records initially 109 for (; i < 5; i++) { 110 RpcLogDetails rpcLogDetails = TestSlowLogRecorder 111 .getRpcLogDetails("userName_" + (i + 1), "client_" + (i + 1), "class_" + (i + 1)); 112 slowLogRecorder.addSlowLogPayload(rpcLogDetails); 113 } 114 115 // add 2 more records 116 for (; i < 7; i++) { 117 RpcLogDetails rpcLogDetails = TestSlowLogRecorder 118 .getRpcLogDetails("userName_" + (i + 1), "client_" + (i + 1), "class_" + (i + 1)); 119 slowLogRecorder.addSlowLogPayload(rpcLogDetails); 120 } 121 122 // add 3 more records 123 for (; i < 10; i++) { 124 RpcLogDetails rpcLogDetails = TestSlowLogRecorder 125 .getRpcLogDetails("userName_" + (i + 1), "client_" + (i + 1), "class_" + (i + 1)); 126 slowLogRecorder.addSlowLogPayload(rpcLogDetails); 127 } 128 129 // add 4 more records 130 for (; i < 14; i++) { 131 RpcLogDetails rpcLogDetails = TestSlowLogRecorder 132 .getRpcLogDetails("userName_" + (i + 1), "client_" + (i + 1), "class_" + (i + 1)); 133 slowLogRecorder.addSlowLogPayload(rpcLogDetails); 134 } 135 136 Assert.assertNotEquals(-1, HBASE_TESTING_UTILITY 137 .waitFor(3000, () -> slowLogRecorder.getSlowLogPayloads(request).size() == 14)); 138 139 Assert.assertNotEquals(-1, 140 HBASE_TESTING_UTILITY.waitFor(3000, () -> getTableCount(connection) == 14)); 141 } 142 143 private int getTableCount(Connection connection) { 144 try (Table table = connection.getTable(SlowLogTableAccessor.SLOW_LOG_TABLE_NAME)) { 145 ResultScanner resultScanner = table.getScanner(new Scan().setReadType(Scan.ReadType.STREAM)); 146 int count = 0; 147 for (Result result : resultScanner) { 148 ++count; 149 } 150 return count; 151 } catch (Exception e) { 152 return 0; 153 } 154 } 155 156 private Connection waitForSlowLogTableCreation() { 157 Connection connection = 158 HBASE_TESTING_UTILITY.getMiniHBaseCluster().getRegionServer(0).getConnection(); 159 Assert.assertNotEquals(-1, HBASE_TESTING_UTILITY.waitFor(2000, () -> { 160 try { 161 return MetaTableAccessor.tableExists(connection, SlowLogTableAccessor.SLOW_LOG_TABLE_NAME); 162 } catch (IOException e) { 163 return false; 164 } 165 })); 166 return connection; 167 } 168 169 @Test 170 public void testHigherSlowLogs() throws Exception { 171 Connection connection = waitForSlowLogTableCreation(); 172 173 slowLogRecorder.clearSlowLogPayloads(); 174 AdminProtos.SlowLogResponseRequest request = 175 AdminProtos.SlowLogResponseRequest.newBuilder().setLimit(500000).build(); 176 Assert.assertEquals(slowLogRecorder.getSlowLogPayloads(request).size(), 0); 177 178 for (int j = 0; j < 100; j++) { 179 CompletableFuture.runAsync(() -> { 180 for (int i = 0; i < 350; i++) { 181 if (i == 300) { 182 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS); 183 } 184 RpcLogDetails rpcLogDetails = TestSlowLogRecorder 185 .getRpcLogDetails("userName_" + (i + 1), "client_" + (i + 1), "class_" + (i + 1)); 186 slowLogRecorder.addSlowLogPayload(rpcLogDetails); 187 } 188 }); 189 } 190 191 Assert.assertNotEquals(-1, HBASE_TESTING_UTILITY.waitFor(7000, () -> { 192 int count = slowLogRecorder.getSlowLogPayloads(request).size(); 193 LOG.debug("RingBuffer records count: {}", count); 194 return count > 2000; 195 })); 196 197 Assert.assertNotEquals(-1, HBASE_TESTING_UTILITY.waitFor(7000, () -> { 198 int count = getTableCount(connection); 199 LOG.debug("SlowLog Table records count: {}", count); 200 return count > 2000; 201 })); 202 } 203 204}