CassandraSpanConsumer implements SpanConsumer { @Override public Call<Void> accept(List<Span> input) { if (input.isEmpty()) return Call.create(null); Set<InsertSpan.Input> spans = new LinkedHashSet<>(); Set<Entry<String, String>> serviceRemoteServices = new LinkedHashSet<>(); Set<Entry<String, String>> serviceSpans = new LinkedHashSet<>(); Set<InsertTraceByServiceRemoteService.Input> traceByServiceRemoteServices = new LinkedHashSet<>(); Set<InsertTraceByServiceSpan.Input> traceByServiceSpans = new LinkedHashSet<>(); Set<Entry<String, String>> autocompleteTags = new LinkedHashSet<>(); for (Span s : input) { long ts_micro = s.timestampAsLong(); if (ts_micro == 0L) ts_micro = guessTimestamp(s); UUID ts_uuid = new UUID( UUIDs.startOf(ts_micro != 0L ? (ts_micro / 1000L) : System.currentTimeMillis()) .getMostSignificantBits(), UUIDs.random().getLeastSignificantBits()); spans.add(insertSpan.newInput(s, ts_uuid)); if (!searchEnabled) continue; String service = s.localServiceName() != null ? s.localServiceName() : ""; String span = null != s.name() ? s.name() : ""; if (null == s.localServiceName()) continue; String remoteService = s.remoteServiceName(); if (insertServiceRemoteService != null && remoteService != null) { serviceRemoteServices.add(new SimpleImmutableEntry<>(service, remoteService)); } serviceSpans.add(new SimpleImmutableEntry<>(service, span)); if (ts_micro == 0L) continue; int bucket = durationIndexBucket(ts_micro); long duration = s.durationAsLong() / 1000L; traceByServiceSpans.add( insertTraceByServiceSpan.newInput(service, span, bucket, ts_uuid, s.traceId(), duration)); if (span.isEmpty()) continue; if (insertServiceRemoteService != null && remoteService != null) { traceByServiceRemoteServices.add( insertTraceByServiceRemoteService.newInput(service, remoteService, bucket, ts_uuid, s.traceId())); } traceByServiceSpans.add( insertTraceByServiceSpan.newInput(service, "", bucket, ts_uuid, s.traceId(), duration)); if (insertAutocompleteValue != null) { for (Entry<String, String> entry : s.tags().entrySet()) { if (autocompleteKeys.contains(entry.getKey())) autocompleteTags.add(entry); } } } List<Call<Void>> calls = new ArrayList<>(); for (InsertSpan.Input span : spans) { calls.add(insertSpan.create(span)); } for (Entry<String, String> serviceSpan : serviceSpans) { insertServiceSpan.maybeAdd(serviceSpan, calls); } for (Entry<String, String> serviceRemoteService : serviceRemoteServices) { insertServiceRemoteService.maybeAdd(serviceRemoteService, calls); } for (InsertTraceByServiceSpan.Input serviceSpan : traceByServiceSpans) { calls.add(insertTraceByServiceSpan.create(serviceSpan)); } for (InsertTraceByServiceRemoteService.Input serviceRemoteService : traceByServiceRemoteServices) { calls.add(insertTraceByServiceRemoteService.create(serviceRemoteService)); } for (Entry<String, String> autocompleteTag : autocompleteTags) { insertAutocompleteValue.maybeAdd(autocompleteTag, calls); } return calls.isEmpty() ? Call.create(null) : AggregateCall.newVoidCall(calls); } CassandraSpanConsumer(CassandraStorage storage); @Override Call<Void> accept(List<Span> input);  }
@Test public void emptyInput_emptyCall() { Call<Void> call = consumer.accept(Collections.emptyList()); assertThat(call).hasSameClassAs(Call.create(null)); }
@Test public void doesntSetTraceIdHigh_128() { Span span = spanWithoutAnnotationsOrTags .toBuilder() .traceId("77fcac3d4c5be8d2a037812820c65f28") .build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertSpan) .extracting("input.trace_id_high", "input.trace_id") .containsExactly(tuple(null, span.traceId())); }
@Test public void doesntSetTraceIdHigh_64() { Span span = spanWithoutAnnotationsOrTags; Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertSpan) .extracting("input.trace_id_high", "input.trace_id") .containsExactly(tuple(null, span.traceId())); }
@Test public void strictTraceIdFalse_setsTraceIdHigh() { consumer = spanConsumer(CassandraStorage.newBuilder().strictTraceId(false)); Span span = spanWithoutAnnotationsOrTags .toBuilder() .traceId("77fcac3d4c5be8d2a037812820c65f28") .build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertSpan) .extracting("input.trace_id_high", "input.trace_id") .containsExactly(tuple("77fcac3d4c5be8d2", "a037812820c65f28")); }
@Test public void serviceSpanKeys() { Span span = spanWithoutAnnotationsOrTags; Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof DeduplicatingVoidCallFactory.InvalidatingVoidCall) .extracting("input") .containsExactly(entry(FRONTEND.serviceName(), span.name())); }
@Test public void serviceRemoteServiceKeys_addsRemoteServiceName() { Span span = spanWithoutAnnotationsOrTags.toBuilder().remoteEndpoint(BACKEND).build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof DeduplicatingVoidCallFactory.InvalidatingVoidCall) .extracting("input") .containsExactly( entry(FRONTEND.serviceName(), span.name()), entry(FRONTEND.serviceName(), BACKEND.serviceName()) ); }
@Test public void serviceRemoteServiceKeys_skipsRemoteServiceNameWhenNoLocalService() { Span span = spanWithoutAnnotationsOrTags.toBuilder() .localEndpoint(null) .remoteEndpoint(BACKEND).build(); Call<Void> call = consumer.accept(singletonList(span)); assertThat(call).isInstanceOf(InsertSpan.class); }
@Test public void serviceSpanKeys_emptyWhenNoEndpoints() { Span span = spanWithoutAnnotationsOrTags.toBuilder().localEndpoint(null).build(); assertThat(consumer.accept(singletonList(span))) .isInstanceOf(ResultSetFutureCall.class); }
@Test public void traceByServiceSpan_indexesLocalServiceNameAndEmptySpanName() { Span span = spanWithoutAnnotationsOrTags; Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertTraceByServiceSpan) .extracting("input.service", "input.span") .containsExactly( tuple(FRONTEND.serviceName(), span.name()), tuple(FRONTEND.serviceName(), "")); }
@Test public void traceByServiceSpan_indexesDurationInMillis() { Span span = spanWithoutAnnotationsOrTags; Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertTraceByServiceSpan) .extracting("input.duration") .containsOnly(span.durationAsLong() / 1000L); }
@Test public void traceByServiceSpan_indexesDurationMinimumZero() { Span span = spanWithoutAnnotationsOrTags.toBuilder().duration(12L).build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertTraceByServiceSpan) .extracting("input.duration") .containsOnly(0L); }
@Test public void traceByServiceSpan_skipsOnNoTimestamp() { Span span = spanWithoutAnnotationsOrTags.toBuilder().timestamp(null).build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertTraceByServiceSpan) .extracting("input.service", "input.span") .isEmpty(); }
@Test public void traceByServiceSpan_doesntIndexRemoteService() { Span span = spanWithoutAnnotationsOrTags.toBuilder().remoteEndpoint(BACKEND).build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertTraceByServiceSpan) .hasSize(2) .extracting("input.service") .doesNotContain(BACKEND.serviceName()); }
@Test public void traceByServiceSpan_appendsEmptyWhenNoName() { Span span = spanWithoutAnnotationsOrTags.toBuilder().name(null).build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedCalls(call) .filteredOn(c -> c instanceof InsertTraceByServiceSpan) .extracting("input.service", "input.span") .containsExactly(tuple(FRONTEND.serviceName(), "")); }
@Test public void traceByServiceSpan_emptyWhenNoEndpoints() { Span span = spanWithoutAnnotationsOrTags.toBuilder().localEndpoint(null).build(); assertThat(consumer.accept(singletonList(span))) .isInstanceOf(ResultSetFutureCall.class); }
@Test public void searchDisabled_doesntIndex() { consumer = spanConsumer(CassandraStorage.newBuilder().searchEnabled(false)); Span span = spanWithoutAnnotationsOrTags .toBuilder() .addAnnotation(TODAY * 1000L, "annotation") .putTag("foo", "bar") .duration(10000L) .build(); assertThat(consumer.accept(singletonList(span))) .extracting("input.annotation_query") .satisfies(q -> assertThat(q).isNull()); }
@Test public void doesntIndexWhenOnlyIncludesTimestamp() { Span span = Span.newBuilder().traceId("a").id("1").timestamp(TODAY * 1000L).build(); assertThat(consumer.accept(singletonList(span))) .isInstanceOf(ResultSetFutureCall.class); }