@Test public void testKeepAll() throws InterruptedException { try (TestReceiver receiver = new TestReceiver(HTraceConfiguration .fromMap(Collections.singletonMap(AsyncSpanReceiver.SPAN_MIN_MS, "0")))) { receiver.receiveSpan(createSpan(0)); while (receiver.getQueueSize() > 0) { Thread.sleep(500); } assertEquals(0, receiver.getQueueSize()); assertEquals(1, receiver.getSpansSent()); } }
@Override public void receiveSpan(Span s) { if (s.getStopTimeMillis() - s.getStartTimeMillis() < minSpanSize) { return; } Map<String,String> data = s.getKVAnnotations(); SpanKey dest = getSpanKey(data); if (dest != null) { List<Annotation> annotations = convertToAnnotations(s.getTimelineAnnotations()); if (sendQueueSize.get() > maxQueueSize) { long now = System.currentTimeMillis(); if (now - lastNotificationOfDroppedSpans > 60 * 1000) { log.warn("Tracing spans are being dropped because there are already" + " {} spans queued for delivery.\n" + "This does not affect performance, security or data integrity," + " but distributed tracing information is being lost.", maxQueueSize); lastNotificationOfDroppedSpans = now; } return; } sendQueue.add(new RemoteSpan(host, service == null ? processId : service, s.getTraceId(), s.getSpanId(), Longs.asList(s.getParents()), s.getStartTimeMillis(), s.getStopTimeMillis(), s.getDescription(), data, annotations)); sendQueueSize.incrementAndGet(); } }
AsyncSpanReceiver implements SpanReceiver { @Override public void receiveSpan(Span s) { if (s.getStopTimeMillis() - s.getStartTimeMillis() < minSpanSize) { return; } Map<String,String> data = s.getKVAnnotations(); SpanKey dest = getSpanKey(data); if (dest != null) { List<Annotation> annotations = convertToAnnotations(s.getTimelineAnnotations()); if (sendQueueSize.get() > maxQueueSize) { long now = System.currentTimeMillis(); if (now - lastNotificationOfDroppedSpans > 60 * 1000) { log.warn("Tracing spans are being dropped because there are already" + " {} spans queued for delivery.\n" + "This does not affect performance, security or data integrity," + " but distributed tracing information is being lost.", maxQueueSize); lastNotificationOfDroppedSpans = now; } return; } sendQueue.add(new RemoteSpan(host, service == null ? processId : service, s.getTraceId(), s.getSpanId(), Longs.asList(s.getParents()), s.getStartTimeMillis(), s.getStopTimeMillis(), s.getDescription(), data, annotations)); sendQueueSize.incrementAndGet(); } } }
AsyncSpanReceiver implements SpanReceiver { @Override public void receiveSpan(Span s) { if (s.getStopTimeMillis() - s.getStartTimeMillis() < minSpanSize) { return; } Map<String,String> data = s.getKVAnnotations(); SpanKey dest = getSpanKey(data); if (dest != null) { List<Annotation> annotations = convertToAnnotations(s.getTimelineAnnotations()); if (sendQueueSize.get() > maxQueueSize) { long now = System.currentTimeMillis(); if (now - lastNotificationOfDroppedSpans > 60 * 1000) { log.warn("Tracing spans are being dropped because there are already" + " {} spans queued for delivery.\n" + "This does not affect performance, security or data integrity," + " but distributed tracing information is being lost.", maxQueueSize); lastNotificationOfDroppedSpans = now; } return; } sendQueue.add(new RemoteSpan(host, service == null ? processId : service, s.getTraceId(), s.getSpanId(), Longs.asList(s.getParents()), s.getStartTimeMillis(), s.getStopTimeMillis(), s.getDescription(), data, annotations)); sendQueueSize.incrementAndGet(); } } AsyncSpanReceiver();  AsyncSpanReceiver(HTraceConfiguration conf); }
AsyncSpanReceiver implements SpanReceiver { @Override public void receiveSpan(Span s) { if (s.getStopTimeMillis() - s.getStartTimeMillis() < minSpanSize) { return; } Map<String,String> data = s.getKVAnnotations(); SpanKey dest = getSpanKey(data); if (dest != null) { List<Annotation> annotations = convertToAnnotations(s.getTimelineAnnotations()); if (sendQueueSize.get() > maxQueueSize) { long now = System.currentTimeMillis(); if (now - lastNotificationOfDroppedSpans > 60 * 1000) { log.warn("Tracing spans are being dropped because there are already" + " {} spans queued for delivery.\n" + "This does not affect performance, security or data integrity," + " but distributed tracing information is being lost.", maxQueueSize); lastNotificationOfDroppedSpans = now; } return; } sendQueue.add(new RemoteSpan(host, service == null ? processId : service, s.getTraceId(), s.getSpanId(), Longs.asList(s.getParents()), s.getStartTimeMillis(), s.getStopTimeMillis(), s.getDescription(), data, annotations)); sendQueueSize.incrementAndGet(); } } AsyncSpanReceiver();  AsyncSpanReceiver(HTraceConfiguration conf); static Map<String,String> convertToStrings(Map<byte[],byte[]> bytesMap); static List<Annotation> convertToAnnotations(List<TimelineAnnotation> annotations); @Override void receiveSpan(Span s); @Override void close(); }
AsyncSpanReceiver implements SpanReceiver { @Override public void receiveSpan(Span s) { if (s.getStopTimeMillis() - s.getStartTimeMillis() < minSpanSize) { return; } Map<String,String> data = s.getKVAnnotations(); SpanKey dest = getSpanKey(data); if (dest != null) { List<Annotation> annotations = convertToAnnotations(s.getTimelineAnnotations()); if (sendQueueSize.get() > maxQueueSize) { long now = System.currentTimeMillis(); if (now - lastNotificationOfDroppedSpans > 60 * 1000) { log.warn("Tracing spans are being dropped because there are already" + " {} spans queued for delivery.\n" + "This does not affect performance, security or data integrity," + " but distributed tracing information is being lost.", maxQueueSize); lastNotificationOfDroppedSpans = now; } return; } sendQueue.add(new RemoteSpan(host, service == null ? processId : service, s.getTraceId(), s.getSpanId(), Longs.asList(s.getParents()), s.getStartTimeMillis(), s.getStopTimeMillis(), s.getDescription(), data, annotations)); sendQueueSize.incrementAndGet(); } } AsyncSpanReceiver();  AsyncSpanReceiver(HTraceConfiguration conf); static Map<String,String> convertToStrings(Map<byte[],byte[]> bytesMap); static List<Annotation> convertToAnnotations(List<TimelineAnnotation> annotations); @Override void receiveSpan(Span s); @Override void close(); static final String SEND_TIMER_MILLIS; static final String QUEUE_SIZE; static final String SPAN_MIN_MS; }