HBaseSpanReceiver extends SpanReceiver { public HBaseSpanReceiver(HTraceConfiguration conf) { this.queue = new ArrayBlockingQueue<Span>(1000); this.hconf = HBaseConfiguration.create(); this.table = Bytes.toBytes(conf.get(TABLE_KEY, DEFAULT_TABLE)); this.cf = Bytes.toBytes(conf.get(COLUMNFAMILY_KEY, DEFAULT_COLUMNFAMILY)); this.icf = Bytes.toBytes(conf.get(INDEXFAMILY_KEY, DEFAULT_INDEXFAMILY)); this.maxSpanBatchSize = conf.getInt(MAX_SPAN_BATCH_SIZE_KEY, DEFAULT_MAX_SPAN_BATCH_SIZE); String quorum = conf.get(COLLECTOR_QUORUM_KEY, DEFAULT_COLLECTOR_QUORUM); hconf.set(HConstants.ZOOKEEPER_QUORUM, quorum); String znodeParent = conf.get(ZOOKEEPER_ZNODE_PARENT_KEY, DEFAULT_ZOOKEEPER_ZNODE_PARENT); hconf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, znodeParent); int clientPort = conf.getInt(ZOOKEEPER_CLIENT_PORT_KEY, DEFAULT_ZOOKEEPER_CLIENT_PORT); hconf.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, clientPort); if (this.service != null) { this.service.shutdownNow(); this.service = null; } int numThreads = conf.getInt(NUM_THREADS_KEY, DEFAULT_NUM_THREADS); this.service = Executors.newFixedThreadPool(numThreads, tf); for (int i = 0; i < numThreads; i++) { this.service.submit(new WriteSpanRunnable()); } } HBaseSpanReceiver(HTraceConfiguration conf); @Override void close(); @Override void receiveSpan(Span span); static void main(String[] args); static final String COLLECTOR_QUORUM_KEY; static final String DEFAULT_COLLECTOR_QUORUM; static final String ZOOKEEPER_CLIENT_PORT_KEY; static final int DEFAULT_ZOOKEEPER_CLIENT_PORT; static final String ZOOKEEPER_ZNODE_PARENT_KEY; static final String DEFAULT_ZOOKEEPER_ZNODE_PARENT; static final String NUM_THREADS_KEY; static final int DEFAULT_NUM_THREADS; static final String MAX_SPAN_BATCH_SIZE_KEY; static final int DEFAULT_MAX_SPAN_BATCH_SIZE; static final String TABLE_KEY; static final String DEFAULT_TABLE; static final String COLUMNFAMILY_KEY; static final String DEFAULT_COLUMNFAMILY; static final String INDEXFAMILY_KEY; static final String DEFAULT_INDEXFAMILY; static final byte[] INDEX_SPAN_QUAL; static final byte[] INDEX_TIME_QUAL; }
@Ignore @Test public void testHBaseSpanReceiver() { Table htable = createTable(UTIL); Configuration conf = UTIL.getConfiguration(); Tracer tracer = new Tracer.Builder(). name("testHBaseSpanReceiver"). tracerPool(new TracerPool("testHBaseSpanReceiver")). conf(HTraceConfiguration.fromKeyValuePairs( HBaseSpanReceiver.COLLECTOR_QUORUM_KEY, conf.get(HConstants.ZOOKEEPER_QUORUM), HBaseSpanReceiver.ZOOKEEPER_CLIENT_PORT_KEY, "" + conf.getInt(HConstants.ZOOKEEPER_CLIENT_PORT, 2181), HBaseSpanReceiver.ZOOKEEPER_ZNODE_PARENT_KEY, conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT) )).build(); TraceCreator tc = new TraceCreator(tracer); tc.createThreadedTrace(); tc.createSimpleTrace(); tc.createSampleRpcTrace(); tracer.close(); Scan scan = new Scan(); scan.addFamily(Bytes.toBytes(HBaseSpanReceiver.DEFAULT_COLUMNFAMILY)); scan.setMaxVersions(1); ArrayList<Span> spans = new ArrayList<Span>(); try { ResultScanner scanner = htable.getScanner(scan); Result result = null; while ((result = scanner.next()) != null) { for (Cell cell : result.listCells()) { InputStream in = new ByteArrayInputStream(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()); spans.add(new TestSpan(SpanProtos.Span.parseFrom(in))); } } } catch (IOException e) { Assert.fail("failed to get spans from HBase. " + e.getMessage()); } TraceGraph traceGraph = new TraceGraph(spans); Collection<Span> roots = traceGraph.getSpansByParent().find(SpanId.INVALID); Assert.assertTrue("Trace tree must have roots", !roots.isEmpty()); Assert.assertEquals(3, roots.size()); Map<String, Span> descs = new HashMap<String, Span>(); for (Span root : roots) { descs.put(root.getDescription(), root); } Assert.assertTrue(descs.keySet().contains(TraceCreator.RPC_TRACE_ROOT)); Assert.assertTrue(descs.keySet().contains(TraceCreator.SIMPLE_TRACE_ROOT)); Assert.assertTrue(descs.keySet().contains(TraceCreator.THREADED_TRACE_ROOT)); SpansByParent spansByParentId = traceGraph.getSpansByParent(); Span rpcRoot = descs.get(TraceCreator.RPC_TRACE_ROOT); Assert.assertEquals(1, spansByParentId.find(rpcRoot.getSpanId()).size()); Span rpcChild1 = spansByParentId.find(rpcRoot.getSpanId()).iterator().next(); Assert.assertEquals(1, spansByParentId.find(rpcChild1.getSpanId()).size()); Span rpcChild2 = spansByParentId.find(rpcChild1.getSpanId()).iterator().next(); Assert.assertEquals(1, spansByParentId.find(rpcChild2.getSpanId()).size()); Span rpcChild3 = spansByParentId.find(rpcChild2.getSpanId()).iterator().next(); Assert.assertEquals(0, spansByParentId.find(rpcChild3.getSpanId()).size()); Scan iscan = new Scan(); iscan.addColumn(Bytes.toBytes(HBaseSpanReceiver.DEFAULT_INDEXFAMILY), HBaseSpanReceiver.INDEX_SPAN_QUAL); try { ResultScanner scanner = htable.getScanner(iscan); Result result = null; while ((result = scanner.next()) != null) { for (Cell cell : result.listCells()) { InputStream in = new ByteArrayInputStream(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()); Assert.assertEquals(SpanProtos.Span.parseFrom(in).getParentId(), 0); } } } catch (IOException e) { Assert.fail("failed to get spans from index family. " + e.getMessage()); } }