1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.trace;
20
21 import org.apache.commons.cli.CommandLine;
22 import org.apache.hadoop.conf.Configuration;
23 import org.apache.hadoop.hbase.HBaseConfiguration;
24 import org.apache.hadoop.hbase.IntegrationTestingUtility;
25 import org.apache.hadoop.hbase.client.HTable;
26 import org.apache.hadoop.hbase.testclassification.IntegrationTests;
27 import org.apache.hadoop.hbase.TableName;
28 import org.apache.hadoop.hbase.client.Admin;
29 import org.apache.hadoop.hbase.client.BufferedMutator;
30 import org.apache.hadoop.hbase.client.Get;
31 import org.apache.hadoop.hbase.client.Put;
32 import org.apache.hadoop.hbase.client.Result;
33 import org.apache.hadoop.hbase.client.ResultScanner;
34 import org.apache.hadoop.hbase.client.Scan;
35 import org.apache.hadoop.hbase.client.Table;
36 import org.apache.hadoop.hbase.util.AbstractHBaseTool;
37 import org.apache.hadoop.hbase.util.Bytes;
38 import org.apache.hadoop.util.ToolRunner;
39 import org.apache.htrace.Sampler;
40 import org.apache.htrace.Trace;
41 import org.apache.htrace.TraceScope;
42 import org.junit.Test;
43 import org.junit.experimental.categories.Category;
44
45 import java.io.IOException;
46 import java.util.Random;
47 import java.util.concurrent.ExecutorService;
48 import java.util.concurrent.Executors;
49 import java.util.concurrent.LinkedBlockingQueue;
50 import java.util.concurrent.TimeUnit;
51
52 @Category(IntegrationTests.class)
53 public class IntegrationTestSendTraceRequests extends AbstractHBaseTool {
54
55 public static final String TABLE_ARG = "t";
56 public static final String CF_ARG = "f";
57
58 public static final String TABLE_NAME_DEFAULT = "SendTracesTable";
59 public static final String COLUMN_FAMILY_DEFAULT = "D";
60 private TableName tableName = TableName.valueOf(TABLE_NAME_DEFAULT);
61 private byte[] familyName = Bytes.toBytes(COLUMN_FAMILY_DEFAULT);
62 private IntegrationTestingUtility util;
63 private Random random = new Random();
64 private Admin admin;
65 private SpanReceiverHost receiverHost;
66
67 public static void main(String[] args) throws Exception {
68 Configuration configuration = HBaseConfiguration.create();
69 IntegrationTestingUtility.setUseDistributedCluster(configuration);
70 IntegrationTestSendTraceRequests tool = new IntegrationTestSendTraceRequests();
71 ToolRunner.run(configuration, tool, args);
72 }
73
74 @Override
75 protected void addOptions() {
76 addOptWithArg(TABLE_ARG, "The table name to target. Will be created if not there already.");
77 addOptWithArg(CF_ARG, "The family to target");
78 }
79
80 @Override
81 public void processOptions(CommandLine cmd) {
82 String tableNameString = cmd.getOptionValue(TABLE_ARG, TABLE_NAME_DEFAULT);
83 String familyString = cmd.getOptionValue(CF_ARG, COLUMN_FAMILY_DEFAULT);
84
85 this.tableName = TableName.valueOf(tableNameString);
86 this.familyName = Bytes.toBytes(familyString);
87 }
88
89 @Override
90 public int doWork() throws Exception {
91 internalDoWork();
92 return 0;
93 }
94
95 @Test
96 public void internalDoWork() throws Exception {
97 util = createUtil();
98 admin = util.getHBaseAdmin();
99 setupReceiver();
100
101 deleteTable();
102 createTable();
103 LinkedBlockingQueue<Long> rks = insertData();
104
105 ExecutorService service = Executors.newFixedThreadPool(20);
106 doScans(service, rks);
107 doGets(service, rks);
108
109 service.shutdown();
110 service.awaitTermination(100, TimeUnit.SECONDS);
111 Thread.sleep(90000);
112 receiverHost.closeReceivers();
113 util.restoreCluster();
114 util = null;
115 }
116
117 private void doScans(ExecutorService service, final LinkedBlockingQueue<Long> rks) {
118
119 for (int i = 0; i < 100; i++) {
120 Runnable runnable = new Runnable() {
121 private TraceScope innerScope = null;
122 private final LinkedBlockingQueue<Long> rowKeyQueue = rks;
123 @Override
124 public void run() {
125 ResultScanner rs = null;
126 try {
127 innerScope = Trace.startSpan("Scan", Sampler.ALWAYS);
128 Table ht = new HTable(util.getConfiguration(), tableName);
129 Scan s = new Scan();
130 s.setStartRow(Bytes.toBytes(rowKeyQueue.take()));
131 s.setBatch(7);
132 rs = ht.getScanner(s);
133
134 long accum = 0;
135
136 for(int x = 0; x < 1000; x++) {
137 Result r = rs.next();
138 accum |= Bytes.toLong(r.getRow());
139 }
140
141 innerScope.getSpan().addTimelineAnnotation("Accum result = " + accum);
142
143 ht.close();
144 ht = null;
145 } catch (IOException e) {
146 e.printStackTrace();
147
148 innerScope.getSpan().addKVAnnotation(
149 Bytes.toBytes("exception"),
150 Bytes.toBytes(e.getClass().getSimpleName()));
151
152 } catch (Exception e) {
153 } finally {
154 if (innerScope != null) innerScope.close();
155 if (rs != null) rs.close();
156 }
157
158 }
159 };
160 service.submit(runnable);
161 }
162
163 }
164
165 private void doGets(ExecutorService service, final LinkedBlockingQueue<Long> rowKeys)
166 throws IOException {
167 for (int i = 0; i < 100; i++) {
168 Runnable runnable = new Runnable() {
169 private TraceScope innerScope = null;
170 private final LinkedBlockingQueue<Long> rowKeyQueue = rowKeys;
171
172 @Override
173 public void run() {
174
175
176 Table ht = null;
177 try {
178 ht = new HTable(util.getConfiguration(), tableName);
179 } catch (IOException e) {
180 e.printStackTrace();
181 }
182
183 long accum = 0;
184 for (int x = 0; x < 5; x++) {
185 try {
186 innerScope = Trace.startSpan("gets", Sampler.ALWAYS);
187 long rk = rowKeyQueue.take();
188 Result r1 = ht.get(new Get(Bytes.toBytes(rk)));
189 if (r1 != null) {
190 accum |= Bytes.toLong(r1.getRow());
191 }
192 Result r2 = ht.get(new Get(Bytes.toBytes(rk)));
193 if (r2 != null) {
194 accum |= Bytes.toLong(r2.getRow());
195 }
196 innerScope.getSpan().addTimelineAnnotation("Accum = " + accum);
197
198 } catch (IOException e) {
199
200 } catch (InterruptedException ie) {
201
202 } finally {
203 if (innerScope != null) innerScope.close();
204 }
205 }
206
207 }
208 };
209 service.submit(runnable);
210 }
211 }
212
213 private void createTable() throws IOException {
214 TraceScope createScope = null;
215 try {
216 createScope = Trace.startSpan("createTable", Sampler.ALWAYS);
217 util.createTable(tableName, familyName);
218 } finally {
219 if (createScope != null) createScope.close();
220 }
221 }
222
223 private void deleteTable() throws IOException {
224 TraceScope deleteScope = null;
225
226 try {
227 if (admin.tableExists(tableName)) {
228 deleteScope = Trace.startSpan("deleteTable", Sampler.ALWAYS);
229 util.deleteTable(tableName);
230 }
231 } finally {
232 if (deleteScope != null) deleteScope.close();
233 }
234 }
235
236 private LinkedBlockingQueue<Long> insertData() throws IOException, InterruptedException {
237 LinkedBlockingQueue<Long> rowKeys = new LinkedBlockingQueue<Long>(25000);
238 BufferedMutator ht = util.getConnection().getBufferedMutator(this.tableName);
239 byte[] value = new byte[300];
240 for (int x = 0; x < 5000; x++) {
241 TraceScope traceScope = Trace.startSpan("insertData", Sampler.ALWAYS);
242 try {
243 for (int i = 0; i < 5; i++) {
244 long rk = random.nextLong();
245 rowKeys.add(rk);
246 Put p = new Put(Bytes.toBytes(rk));
247 for (int y = 0; y < 10; y++) {
248 random.nextBytes(value);
249 p.add(familyName, Bytes.toBytes(random.nextLong()), value);
250 }
251 ht.mutate(p);
252 }
253 if ((x % 1000) == 0) {
254 admin.flush(tableName);
255 }
256 } finally {
257 traceScope.close();
258 }
259 }
260 admin.flush(tableName);
261 return rowKeys;
262 }
263
264 private IntegrationTestingUtility createUtil() throws Exception {
265 Configuration conf = getConf();
266 if (this.util == null) {
267 IntegrationTestingUtility u;
268 if (conf == null) {
269 u = new IntegrationTestingUtility();
270 } else {
271 u = new IntegrationTestingUtility(conf);
272 }
273 util = u;
274 util.initializeCluster(1);
275
276 }
277 return this.util;
278 }
279
280 private void setupReceiver() {
281 Configuration conf = new Configuration(util.getConfiguration());
282 conf.setBoolean("hbase.zipkin.is-in-client-mode", true);
283
284 this.receiverHost = SpanReceiverHost.getInstance(conf);
285 }
286 }