ElasticsearchSpanConsumer implements SpanConsumer { @Override public Call<Void> accept(List<Span> spans) { if (spans.isEmpty()) return Call.create(null); BulkSpanIndexer indexer = new BulkSpanIndexer(this); indexSpans(indexer, spans); return indexer.newCall(); } ElasticsearchSpanConsumer(ElasticsearchStorage es); @Override Call<Void> accept(List<Span> spans);  }
@Test void addsTimestamp_millisIntoJson() throws Exception { server.enqueue(SUCCESS_RESPONSE); Span span = Span.newBuilder().traceId("20").id("20").name("get").timestamp(TODAY * 1000).build(); accept(span); assertThat(server.takeRequest().request().contentUtf8()) .contains("\n{\"timestamp_millis\":" + TODAY + ",\"traceId\":"); }
@Test void writesSpanNaturallyWhenNoTimestamp() throws Exception { server.enqueue(SUCCESS_RESPONSE); Span span = Span.newBuilder().traceId("1").id("1").name("foo").build(); accept(Span.newBuilder().traceId("1").id("1").name("foo").build()); assertThat(server.takeRequest().request().contentUtf8()) .contains("\n" + new String(SpanBytesEncoder.JSON_V2.encode(span), UTF_8) + "\n"); }
@Test void traceIsSearchableByServerServiceName() throws Exception { server.enqueue(SUCCESS_RESPONSE); Span clientSpan = Span.newBuilder() .traceId("20") .id("22") .name("") .parentId("21") .timestamp(1000L) .kind(Kind.CLIENT) .localEndpoint(WEB_ENDPOINT) .build(); Span serverSpan = Span.newBuilder() .traceId("20") .id("22") .name("get") .parentId("21") .timestamp(2000L) .kind(Kind.SERVER) .localEndpoint(APP_ENDPOINT) .build(); accept(serverSpan, clientSpan); assertThat(server.takeRequest().request().contentUtf8()) .contains("{\"timestamp_millis\":2") .contains("{\"timestamp_millis\":1"); }
@Test void addsPipelineId() throws Exception { storage.close(); storage = ElasticsearchStorage.newBuilder(() -> WebClient.of(server.httpUri())) .pipeline("zipkin") .build(); ensureIndexTemplate(); server.enqueue(SUCCESS_RESPONSE); accept(Span.newBuilder().traceId("1").id("1").name("foo").build()); AggregatedHttpRequest request = server.takeRequest().request(); assertThat(request.path()).isEqualTo("/_bulk?pipeline=zipkin"); }
@Test void choosesTypeSpecificIndex() throws Exception { server.enqueue(SUCCESS_RESPONSE); Span span = Span.newBuilder() .traceId("1") .id("2") .parentId("1") .name("s") .localEndpoint(APP_ENDPOINT) .addAnnotation(TimeUnit.DAYS.toMicros(365) , "foo") .build(); assertThat(span.timestamp()).isNull(); accept(span); assertThat(server.takeRequest().request().contentUtf8()) .startsWith("{\"index\":{\"_index\":\"zipkin:span-1971-01-01\",\"_type\":\"span\""); }
@Test void searchDisabled_doesntAddTimestampMillis() throws Exception { storage.close(); storage = ElasticsearchStorage.newBuilder(() -> WebClient.of(server.httpUri())) .searchEnabled(false) .build(); ensureIndexTemplates(storage); server.enqueue(SUCCESS_RESPONSE); Span span = Span.newBuilder().traceId("20").id("20").name("get").timestamp(TODAY * 1000).build(); storage.spanConsumer().accept(asList(span)).execute(); assertThat(server.takeRequest().request().contentUtf8()).doesNotContain("timestamp_millis"); }
@Test void addsAutocompleteValue() throws Exception { server.enqueue(SUCCESS_RESPONSE); accept(Span.newBuilder().traceId("1").id("1").timestamp(1).putTag("environment", "A").build()); assertThat(server.takeRequest().request().contentUtf8()) .endsWith("" + "{\"index\":{\"_index\":\"zipkin:autocomplete-1970-01-01\",\"_type\":\"autocomplete\",\"_id\":\"environment=A\"}}\n" + "{\"tagKey\":\"environment\",\"tagValue\":\"A\"}\n"); }
@Test void addsAutocompleteValue_suppressesWhenSameDay() throws Exception { server.enqueue(SUCCESS_RESPONSE); server.enqueue(SUCCESS_RESPONSE); Span s = Span.newBuilder().traceId("1").id("1").timestamp(1).putTag("environment", "A").build(); accept(s); accept(s.toBuilder().id(2).build()); server.takeRequest(); assertThat(server.takeRequest().request().contentUtf8()) .doesNotContain("autocomplete"); }
@Test void addsAutocompleteValue_differentDays() throws Exception { server.enqueue(SUCCESS_RESPONSE); server.enqueue(SUCCESS_RESPONSE); Span s = Span.newBuilder().traceId("1").id("1").timestamp(1).putTag("environment", "A").build(); accept(s); accept(s.toBuilder().id(2).timestamp(1 + TimeUnit.DAYS.toMicros(1)).build()); server.takeRequest(); assertThat(server.takeRequest().request().contentUtf8()) .endsWith("" + "{\"index\":{\"_index\":\"zipkin:autocomplete-1970-01-02\",\"_type\":\"autocomplete\",\"_id\":\"environment=A\"}}\n" + "{\"tagKey\":\"environment\",\"tagValue\":\"A\"}\n"); }
@Test void addsAutocompleteValue_revertsSuppressionOnFailure() throws Exception { server.enqueue(AggregatedHttpResponse.of(HttpStatus.INTERNAL_SERVER_ERROR)); server.enqueue(SUCCESS_RESPONSE); Span s = Span.newBuilder().traceId("1").id("1").timestamp(1).putTag("environment", "A").build(); try { accept(s); failBecauseExceptionWasNotThrown(RuntimeException.class); } catch (RuntimeException expected) { } accept(s); assertThat(server.takeRequest().request().contentUtf8()) .isEqualTo(server.takeRequest().request().contentUtf8()); }