View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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               // Something to keep the jvm from removing the loop.
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               // IGNORED
200             } catch (InterruptedException ie) {
201               // IGNORED
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 }