CassandraSpanConsumer implements SpanConsumer { @Override public Call<Void> accept(List<Span> spans) { int spanCount = spans.size(); if (spanCount == 0) return Call.create(null); V2SpanConverter converter = V2SpanConverter.create(); V1ThriftSpanWriter encoder = new V1ThriftSpanWriter(); Set<InsertTrace.Input> insertTraces = new LinkedHashSet<>(); List<Call<Void>> calls = new ArrayList<>(); for (int i = 0; i < spanCount; i++) { Span span = spans.get(i); V1Span v1Span = converter.convert(span); long ts_micro = span.timestampAsLong(); if (ts_micro == 0L) ts_micro = guessTimestamp(span); insertTraces.add( insertTrace.newInput(v1Span, ByteBuffer.wrap(encoder.write(span)), ts_micro)); } for (InsertTrace.Input insert : insertTraces) { calls.add(insertTrace.create(insert)); } if (!searchEnabled) return AggregateCall.newVoidCall(calls); Set<String> insertServiceNames = new LinkedHashSet<>(); Set<Entry<String, String>> insertRemoteServiceNames = new LinkedHashSet<>(); Set<Entry<String, String>> insertSpanNames = new LinkedHashSet<>(); Set<Entry<String, String>> insertAutocompleteTags = new LinkedHashSet<>(); TraceIdIndexer indexTraceIdByServiceNames = indexTraceIdByServiceName.newIndexer(); TraceIdIndexer indexTraceIdByRemoteServiceNames = indexTraceIdByRemoteServiceName != null ? indexTraceIdByRemoteServiceName.newIndexer() : TraceIdIndexer.NOOP; TraceIdIndexer indexTraceIdBySpanNames = indexTraceIdBySpanName.newIndexer(); TraceIdIndexer indexTraceIdByAnnotations = indexTraceIdByAnnotation.newIndexer(); for (int i = 0; i < spanCount; i++) { Span span = spans.get(i); String serviceName = span.localServiceName(); if (serviceName == null) continue; long timestamp = 1000L * (span.timestampAsLong() / 1000L); if (timestamp == 0L) continue; insertServiceNames.add(serviceName); String remoteServiceName = insertRemoteServiceName != null ? span.remoteServiceName() : null; if (remoteServiceName != null) { insertRemoteServiceNames.add(new SimpleImmutableEntry<>(serviceName, remoteServiceName)); } String spanName = span.name(); if (spanName != null) { insertSpanNames.add(new SimpleImmutableEntry<>(serviceName, spanName)); } if (insertAutocompleteValue != null) { for (Entry<String, String> entry : span.tags().entrySet()) { if (autocompleteKeys.contains(entry.getKey())) insertAutocompleteTags.add(entry); } } long traceId = HexCodec.lowerHexToUnsignedLong(span.traceId()); indexTraceIdByServiceNames.add(IndexTraceId.Input.create(serviceName, timestamp, traceId)); if (remoteServiceName != null) { String partitionKey = serviceName + "." + remoteServiceName; indexTraceIdByRemoteServiceNames.add( IndexTraceId.Input.create(partitionKey, timestamp, traceId)); } if (spanName != null) { String partitionKey = serviceName + "." + spanName; indexTraceIdBySpanNames.add(IndexTraceId.Input.create(partitionKey, timestamp, traceId)); } for (String partitionKey : annotationKeys(span)) { indexTraceIdByAnnotations.add( IndexTraceId.Input.create(partitionKey, timestamp, traceId)); } } for (String insert : insertServiceNames) { insertServiceName.maybeAdd(insert, calls); } for (Entry<String, String> insert : insertRemoteServiceNames) { insertRemoteServiceName.maybeAdd(insert, calls); } for (Entry<String, String> insert : insertSpanNames) { insertSpanName.maybeAdd(insert, calls); } for (Entry<String, String> insert : insertAutocompleteTags) { insertAutocompleteValue.maybeAdd(insert, calls); } for (IndexTraceId.Input insert : indexTraceIdByServiceNames) { indexTraceIdByServiceName.maybeAdd(insert, calls); } for (IndexTraceId.Input insert : indexTraceIdByRemoteServiceNames) { indexTraceIdByRemoteServiceName.maybeAdd(insert, calls); } for (IndexTraceId.Input insert : indexTraceIdBySpanNames) { indexTraceIdBySpanName.maybeAdd(insert, calls); } for (IndexTraceId.Input insert : indexTraceIdByAnnotations) { indexTraceIdByAnnotation.maybeAdd(insert, calls); } return AggregateCall.newVoidCall(calls); } CassandraSpanConsumer(CassandraStorage storage); @Override Call<Void> accept(List<Span> spans);  }
@Test public void emptyInput_emptyCall() { Call<Void> call = consumer.accept(Collections.emptyList()); assertThat(call).hasSameClassAs(Call.create(null)); }
@Test public void doesntIndexWhenOnlyIncludesTimestamp() { Span span = Span.newBuilder().traceId("a").id("1").timestamp(TODAY * 1000L).build(); assertThat(consumer.accept(singletonList(span))) .isInstanceOf(ResultSetFutureCall.class); }
@Test public void doesntIndexWhenMissingLocalServiceName() { Span span = spanWithoutAnnotationsOrTags.toBuilder().localEndpoint(null).build(); assertThat(consumer.accept(singletonList(span))) .isInstanceOf(ResultSetFutureCall.class); }
@Test public void indexesLocalServiceNameAndSpanName() { Span span = spanWithoutAnnotationsOrTags; Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedIndexCalls(call) .extracting("input.partitionKey") .containsExactly("frontend", "frontend.get"); }
@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))) .isInstanceOf(InsertTrace.class); }
@Test public void doesntIndexWhenMissingTimestamp() { Span span = spanWithoutAnnotationsOrTags.toBuilder().timestamp(null).build(); assertThat(consumer.accept(singletonList(span))) .isInstanceOf(InsertTrace.class); }
@Test public void indexKeysBasedOnLocalServiceNotRemote() { Span span = spanWithoutAnnotationsOrTags.toBuilder().remoteEndpoint(BACKEND).build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedIndexCalls(call) .extracting("input.partitionKey") .containsExactly("frontend", "frontend.backend", "frontend.get"); }
@Test public void indexesServiceNameWhenNoSpanName() { Span span = spanWithoutAnnotationsOrTags.toBuilder().name(null).build(); Call<Void> call = consumer.accept(singletonList(span)); assertEnclosedIndexCalls(call) .extracting("input.partitionKey") .containsExactly(FRONTEND.serviceName()); }
@Test public void treatsIndexesSeparately() { Span span1 = Span.newBuilder() .traceId("1") .id("2") .name("foo") .timestamp(TODAY * 1000L) .localEndpoint(Endpoint.newBuilder().serviceName("app").build()) .remoteEndpoint(Endpoint.newBuilder().serviceName("foo").build()) .build(); Span span2 = Span.newBuilder() .traceId("1") .id("3") .timestamp(TODAY * 1000L) .localEndpoint(Endpoint.newBuilder().serviceName("app.foo").build()) .build(); Call<Void> call = consumer.accept(asList(span1, span2)); assertEnclosedIndexCalls(call) .extracting("factory.indexerFactory.table", "input.partitionKey") .containsExactly( tuple(Tables.SERVICE_NAME_INDEX, "app"), tuple(Tables.SERVICE_NAME_INDEX, "app.foo"), tuple(Tables.SERVICE_REMOTE_SERVICE_NAME_INDEX, "app.foo"), tuple(Tables.SERVICE_SPAN_NAME_INDEX, "app.foo") ); assertThat(consumer.accept(singletonList(span2))) .isInstanceOf(InsertTrace.class); }