KuduSpanReceiver extends SpanReceiver { public KuduSpanReceiver(HTraceConfiguration conf) { String masterHost; Integer masterPort; Integer workerCount; Integer bossCount; Boolean isStatisticsEnabled; Long adminOperationTimeout; Long operationTimeout; Long socketReadTimeout; masterHost = conf.get(KuduReceiverConstants.KUDU_MASTER_HOST_KEY, KuduReceiverConstants.DEFAULT_KUDU_MASTER_HOST); masterPort = Integer.valueOf(conf.get(KuduReceiverConstants.KUDU_MASTER_PORT_KEY, KuduReceiverConstants.DEFAULT_KUDU_MASTER_PORT)); if (conf.get(KuduReceiverConstants.KUDU_CLIENT_BOSS_COUNT_KEY) != null) { bossCount = Integer.valueOf(conf.get(KuduReceiverConstants.KUDU_CLIENT_BOSS_COUNT_KEY)); } else { bossCount = null; } if (conf.get(KuduReceiverConstants.KUDU_CLIENT_WORKER_COUNT_KEY) != null) { workerCount = Integer.valueOf(conf.get(KuduReceiverConstants.KUDU_CLIENT_WORKER_COUNT_KEY)); } else { workerCount = null; } if (conf.get(KuduReceiverConstants.KUDU_CLIENT_STATISTICS_ENABLED_KEY) != null) { isStatisticsEnabled = Boolean.valueOf(conf.get(KuduReceiverConstants.KUDU_CLIENT_STATISTICS_ENABLED_KEY)); } else { isStatisticsEnabled = null; } if (conf.get(KuduReceiverConstants.KUDU_CLIENT_TIMEOUT_ADMIN_OPERATION_KEY) != null) { adminOperationTimeout = Long.valueOf(conf.get(KuduReceiverConstants.KUDU_CLIENT_TIMEOUT_ADMIN_OPERATION_KEY)); } else { adminOperationTimeout = null; } if (conf.get(KuduReceiverConstants.KUDU_CLIENT_TIMEOUT_OPERATION_KEY) != null) { operationTimeout = Long.valueOf(conf.get(KuduReceiverConstants.KUDU_CLIENT_TIMEOUT_OPERATION_KEY)); } else { operationTimeout = null; } if (conf.get(KuduReceiverConstants.KUDU_CLIENT_TIMEOUT_SOCKET_READ_KEY) != null) { socketReadTimeout = Long.valueOf(conf.get(KuduReceiverConstants.KUDU_CLIENT_TIMEOUT_SOCKET_READ_KEY)); } else { socketReadTimeout = null; } this.clientConf = new KuduClientConfiguration(masterHost, masterPort, workerCount, bossCount, isStatisticsEnabled, adminOperationTimeout, operationTimeout, socketReadTimeout); this.table_span = conf.get(KuduReceiverConstants.KUDU_SPAN_TABLE_KEY, KuduReceiverConstants.DEFAULT_KUDU_SPAN_TABLE); this.table_timeline = conf.get(KuduReceiverConstants.KUDU_SPAN_TIMELINE_ANNOTATION_TABLE_KEY, KuduReceiverConstants.DEFAULT_KUDU_SPAN_TIMELINE_ANNOTATION_TABLE); this.table_span_parent = conf.get(KuduReceiverConstants.KUDU_SPAN_PARENT_TABLE_KEY, KuduReceiverConstants.DEFAULT_KUDU_SPAN_PARENT_TABLE); this.column_span_trace_id = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_TRACE_ID; this.column_span_start_time = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_START_TIME; this.column_span_stop_time = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_STOP_TIME; this.column_span_span_id = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_SPAN_ID; this.column_parent_id_low = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_ID_LOW; this.column_parent_id_high = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_ID_HIGH; this.column_parent_child_span_id = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_CHILD_SPANID; this.column_span_description = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_DESCRIPTION; this.column_span_parent = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_PARENT; this.column_timeline_time = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_TIME; this.column_timeline_message = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_MESSAGE; this.column_timeline_span_id = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_SPANID; this.column_timeline_timeline_id = KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_TIMELINEID; if (this.session == null) { if (this.client == null) { client = clientConf.buildClient(); } session = client.newSession(); } try { tableSpan = client.openTable(table_span); tableTimeline = client.openTable(table_timeline); tableParent = client.openTable(table_span_parent); } catch (java.lang.Exception ex) { LOG.warn("Failed to open kudu tables to store Spans. " + ex.getMessage()); } } KuduSpanReceiver(HTraceConfiguration conf); @Override void close(); @Override void receiveSpan(Span span);  }
@Ignore @Test public void TestKuduSpanReceiver() throws Exception { createTable(); Tracer tracer = new Tracer.Builder(). name("testKuduSpanReceiver"). tracerPool(new TracerPool("testKuduSpanReceiver")). conf(HTraceConfiguration.fromKeyValuePairs( "sampler.classes", "AlwaysSampler", "span.receiver.classes", "org.apache.htrace.impl.KuduSpanReceiver", KuduReceiverConstants.KUDU_MASTER_HOST_KEY, BaseKuduTest.getMasterAddresses().split(":")[0], KuduReceiverConstants.KUDU_MASTER_PORT_KEY, BaseKuduTest.getMasterAddresses().split(":")[1])) .build(); TraceScope scope = tracer.newScope("testKuduScope"); scope.addTimelineAnnotation("test"); Span testSpan = scope.getSpan(); TraceScope childScope = tracer.newScope("testKuduChildScope", new SpanId(100,200)); Span childScopeSpan = childScope.getSpan(); childScope.close(); scope.close(); tracer.close(); KuduClient client = BaseKuduTest.syncClient; List<String> spanColumns = new ArrayList<>(); spanColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_TRACE_ID); spanColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_SPAN_ID); spanColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_DESCRIPTION); spanColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_START_TIME); spanColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_STOP_TIME); KuduScanner scanner = client.newScannerBuilder(client.openTable(KuduReceiverConstants.DEFAULT_KUDU_SPAN_TABLE)) .setProjectedColumnNames(spanColumns) .build(); MilliSpan dbSpan = null; while (scanner.hasMoreRows()) { RowResultIterator results = scanner.nextRows(); while (results.hasNext()) { RowResult result = results.next(); long traceId = result.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_TRACE_ID); MilliSpan.Builder builder = new MilliSpan.Builder() .spanId(new SpanId(result.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_SPAN_ID), result.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_TRACE_ID))) .description(result.getString(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_DESCRIPTION)) .begin(result.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_START_TIME)) .end(result.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_SPAN_STOP_TIME)); List<String> parentColumns = new ArrayList<>(); parentColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_ID_LOW); parentColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_ID_HIGH); parentColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_CHILD_SPANID); KuduScanner parentScanner = client .newScannerBuilder(client.openTable(KuduReceiverConstants.DEFAULT_KUDU_SPAN_PARENT_TABLE)) .setProjectedColumnNames(parentColumns) .addPredicate(KuduPredicate .newComparisonPredicate(new ColumnSchema.ColumnSchemaBuilder (KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_CHILD_SPANID, Type.INT64) .build(), KuduPredicate.ComparisonOp.EQUAL, traceId)) .build(); List<SpanId> parentList = new LinkedList<SpanId>(); while (parentScanner.hasMoreRows()) { RowResultIterator parentResults = parentScanner.nextRows(); while (parentResults.hasNext()) { RowResult parentRow = parentResults.next(); parentList.add(new SpanId( parentRow.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_ID_HIGH), parentRow.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_PARENT_ID_LOW))); } } SpanId[] array = new SpanId[parentList.size()]; parentList.toArray(array); builder.parents(array); List<String> timelineColumns = new ArrayList<>(); timelineColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_TIME); timelineColumns.add(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_MESSAGE); KuduScanner timelineScanner = client .newScannerBuilder(client.openTable(KuduReceiverConstants.DEFAULT_KUDU_SPAN_TIMELINE_ANNOTATION_TABLE)) .setProjectedColumnNames(timelineColumns) .addPredicate(KuduPredicate .newComparisonPredicate(new ColumnSchema.ColumnSchemaBuilder (KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_SPANID, Type.INT64) .build(), KuduPredicate.ComparisonOp.EQUAL, traceId)) .build(); List<TimelineAnnotation> timelineList = new LinkedList<TimelineAnnotation>(); while (timelineScanner.hasMoreRows()) { RowResultIterator timelineResults = timelineScanner.nextRows(); while (timelineResults.hasNext()) { RowResult timelineRow = timelineResults.next(); timelineList.add(new TimelineAnnotation (timelineRow.getLong(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_TIME), timelineRow.getString(KuduReceiverConstants.DEFAULT_KUDU_COLUMN_TIMELINE_MESSAGE))); } } builder.timeline(timelineList); dbSpan = builder.build(); break; } } Assert.assertEquals(testSpan.getSpanId().getHigh(), dbSpan.getSpanId().getHigh()); Assert.assertEquals(testSpan.getSpanId().getLow(), dbSpan.getSpanId().getLow()); Assert.assertEquals(testSpan.getStartTimeMillis(), dbSpan.getStartTimeMillis()); Assert.assertEquals(testSpan.getStopTimeMillis(), dbSpan.getStopTimeMillis()); Assert.assertEquals(testSpan.getDescription(), dbSpan.getDescription()); Assert.assertEquals(testSpan.getTimelineAnnotations().get(0).getMessage(), dbSpan.getTimelineAnnotations().get(0).getMessage()); Assert.assertEquals(testSpan.getTimelineAnnotations().get(0).getTime(), dbSpan.getTimelineAnnotations().get(0).getTime()); syncClient.deleteTable(KuduReceiverConstants.DEFAULT_KUDU_SPAN_TABLE); syncClient.deleteTable(KuduReceiverConstants.DEFAULT_KUDU_SPAN_TIMELINE_ANNOTATION_TABLE); }