DependencyLinkV2SpanIterator implements Iterator<Span> { @Override public Span next() { if (!hasNext()) throw new NoSuchElementException(); Record row = delegate.peek(); long spanId = row.getValue(ZipkinSpans.ZIPKIN_SPANS.ID); boolean error = false; String lcService = null, srService = null, csService = null, caService = null, saService = null, maService = null, mrService = null, msService = null; while (hasNext()) { if (spanId != delegate.peek().getValue(ZipkinSpans.ZIPKIN_SPANS.ID)) { break; } Record next = delegate.next(); String key = emptyToNull(next, ZIPKIN_ANNOTATIONS.A_KEY); String value = emptyToNull(next, ZIPKIN_ANNOTATIONS.ENDPOINT_SERVICE_NAME); if (key == null || value == null) continue; switch (key) { case "lc": lcService = value; break; case "ca": caService = value; break; case "cs": csService = value; break; case "ma": maService = value; break; case "mr": mrService = value; break; case "ms": msService = value; break; case "sa": saService = value; break; case "sr": srService = value; break; case "error": error = V1BinaryAnnotation.TYPE_STRING == next.get(ZIPKIN_ANNOTATIONS.A_TYPE); } } if (caService == null) caService = csService; if (saService != null && saService.equals(caService)) caService = null; long parentId = maybeGet(row, ZipkinSpans.ZIPKIN_SPANS.PARENT_ID, 0L); Span.Builder result = Span.newBuilder().traceId(traceIdHi, traceIdLo).parentId(parentId).id(spanId); if (error) { result.putTag("error", "" ); } if (srService != null) { return result .kind(Span.Kind.SERVER) .localEndpoint(ep(srService)) .remoteEndpoint(ep(caService)) .build(); } else if (saService != null) { Endpoint localEndpoint = ep(caService); if (localEndpoint == null) localEndpoint = ep(lcService); return result .kind(csService != null ? Span.Kind.CLIENT : null) .localEndpoint(localEndpoint) .remoteEndpoint(ep(saService)) .build(); } else if (csService != null) { return result.kind(Span.Kind.SERVER).localEndpoint(ep(caService)).build(); } else if (mrService != null) { return result .kind(Span.Kind.CONSUMER) .localEndpoint(ep(mrService)) .remoteEndpoint(ep(maService)) .build(); } else if (msService != null) { return result .kind(Span.Kind.PRODUCER) .localEndpoint(ep(msService)) .remoteEndpoint(ep(maService)) .build(); } return result.build(); } DependencyLinkV2SpanIterator(PeekingIterator<Record> delegate, long traceIdHi, long traceIdLo); @Override boolean hasNext(); @Override Span next(); @Override void remove();  }
@Test public void whenNoServiceLabelsExist_kindIsUnknown() { DependencyLinkV2SpanIterator iterator = iterator(newRecord().values(traceIdHigh, traceId, parentId, spanId, "cs", -1, null)); Span span = iterator.next(); assertThat(span.kind()).isNull(); assertThat(span.localEndpoint()).isNull(); assertThat(span.remoteEndpoint()).isNull(); }
@Test public void whenOnlyAddressLabelsExist_kindIsNull() { DependencyLinkV2SpanIterator iterator = iterator( newRecord().values(traceIdHigh, traceId, parentId, spanId, "ca", TYPE_BOOLEAN, "s1"), newRecord().values(traceIdHigh, traceId, parentId, spanId, "sa", TYPE_BOOLEAN, "s2")); Span span = iterator.next(); assertThat(span.kind()).isNull(); assertThat(span.localServiceName()).isEqualTo("s1"); assertThat(span.remoteServiceName()).isEqualTo("s2"); }
@Test public void whenServerLabelsAreMissing_kindIsUnknownAndLabelsAreCleared() { DependencyLinkV2SpanIterator iterator = iterator( newRecord().values(traceIdHigh, traceId, parentId, spanId, "ca", TYPE_BOOLEAN, "s1")); Span span = iterator.next(); assertThat(span.kind()).isNull(); assertThat(span.localEndpoint()).isNull(); assertThat(span.remoteEndpoint()).isNull(); }
@Test public void whenSrServiceExists_kindIsServer() { DependencyLinkV2SpanIterator iterator = iterator(newRecord().values(traceIdHigh, traceId, parentId, spanId, "sr", -1, "service")); Span span = iterator.next(); assertThat(span.kind()).isEqualTo(Span.Kind.SERVER); assertThat(span.localServiceName()).isEqualTo("service"); assertThat(span.remoteEndpoint()).isNull(); }
@Test public void errorAnnotationIgnored() { DependencyLinkV2SpanIterator iterator = iterator( newRecord().values(traceIdHigh, traceId, parentId, spanId, "error", -1, "service")); Span span = iterator.next(); assertThat(span.tags()).isEmpty(); assertThat(span.annotations()).isEmpty(); }
@Test public void errorTagAdded() { DependencyLinkV2SpanIterator iterator = iterator( newRecord() .values(traceIdHigh, traceId, parentId, spanId, "error", TYPE_STRING, "foo")); Span span = iterator.next(); assertThat(span.tags()).containsOnly(entry("error", "")); }
@Test public void whenSrAndCaServiceExists_caIsThePeer() { DependencyLinkV2SpanIterator iterator = iterator( newRecord().values(traceIdHigh, traceId, parentId, spanId, "ca", TYPE_BOOLEAN, "s1"), newRecord().values(traceIdHigh, traceId, parentId, spanId, "sr", -1, "s2")); Span span = iterator.next(); assertThat(span.kind()).isEqualTo(Span.Kind.SERVER); assertThat(span.localServiceName()).isEqualTo("s2"); assertThat(span.remoteServiceName()).isEqualTo("s1"); }
@Test public void whenSrAndCsServiceExists_caIsThePeer() { DependencyLinkV2SpanIterator iterator = iterator( newRecord().values(traceIdHigh, traceId, parentId, spanId, "cs", -1, "s1"), newRecord().values(traceIdHigh, traceId, parentId, spanId, "sr", -1, "s2")); Span span = iterator.next(); assertThat(span.kind()).isEqualTo(Span.Kind.SERVER); assertThat(span.localServiceName()).isEqualTo("s2"); assertThat(span.remoteServiceName()).isEqualTo("s1"); }
@Test public void whenCrAndCaServiceExists_caIsThePeer() { DependencyLinkV2SpanIterator iterator = iterator( newRecord().values(traceIdHigh, traceId, parentId, spanId, "cs", -1, "foo"), newRecord().values(traceIdHigh, traceId, parentId, spanId, "ca", TYPE_BOOLEAN, "s1"), newRecord().values(traceIdHigh, traceId, parentId, spanId, "sr", -1, "s2")); Span span = iterator.next(); assertThat(span.kind()).isEqualTo(Span.Kind.SERVER); assertThat(span.localServiceName()).isEqualTo("s2"); assertThat(span.remoteServiceName()).isEqualTo("s1"); }
@Test public void specialCasesFinagleLocalSocketLabeling_client() { DependencyLinkV2SpanIterator iterator = iterator( newRecord().values(traceIdHigh, traceId, parentId, spanId, "cs", -1, "service"), newRecord() .values(traceIdHigh, traceId, parentId, spanId, "ca", TYPE_BOOLEAN, "service"), newRecord() .values(traceIdHigh, traceId, parentId, spanId, "sa", TYPE_BOOLEAN, "service")); Span span = iterator.next(); assertThat(span.kind()).isEqualTo(Span.Kind.CLIENT); assertThat(span.localEndpoint()).isNull(); assertThat(span.remoteServiceName()).isEqualTo("service"); }
@Test public void specialCasesFinagleLocalSocketLabeling_server() { DependencyLinkV2SpanIterator iterator = iterator( newRecord() .values(traceIdHigh, traceId, parentId, spanId, "ca", TYPE_BOOLEAN, "service"), newRecord() .values(traceIdHigh, traceId, parentId, spanId, "sa", TYPE_BOOLEAN, "service"), newRecord().values(traceIdHigh, traceId, parentId, spanId, "sr", -1, "service")); Span span = iterator.next(); assertThat(span.kind()).isEqualTo(Span.Kind.SERVER); assertThat(span.localServiceName()).isEqualTo("service"); assertThat(span.remoteEndpoint()).isNull(); }
@Test public void csWithoutSaIsServer() { DependencyLinkV2SpanIterator iterator = iterator(newRecord().values(traceIdHigh, traceId, parentId, spanId, "cs", -1, "s1")); Span span = iterator.next(); assertThat(span.kind()).isEqualTo(Span.Kind.SERVER); assertThat(span.localServiceName()).isEqualTo("s1"); assertThat(span.remoteEndpoint()).isNull(); }