@Test public void testZipkinDurationSampler() throws Exception { ZipkinPortUnificationHandler handler = new ZipkinPortUnificationHandler("9411", new NoopHealthCheckManager(), mockTraceHandler, mockTraceSpanLogsHandler, null, () -> false, () -> false, null, new SpanSampler(new DurationSampler(5), false), null, null); Endpoint localEndpoint1 = Endpoint.newBuilder().serviceName("frontend").ip("10.0.0.1").build(); zipkin2.Span spanServer1 = zipkin2.Span.newBuilder(). traceId("2822889fe47043bd"). id("2822889fe47043bd"). kind(zipkin2.Span.Kind.SERVER). name("getservice"). timestamp(startTime * 1000). duration(4 * 1000). localEndpoint(localEndpoint1). putTag("http.method", "GET"). putTag("http.url", "none+h1c: putTag("http.status_code", "200"). addAnnotation(startTime * 1000, "start processing"). build(); zipkin2.Span spanServer2 = zipkin2.Span.newBuilder(). traceId("3822889fe47043bd"). id("3822889fe47043bd"). kind(zipkin2.Span.Kind.SERVER). name("getservice"). timestamp(startTime * 1000). duration(9 * 1000). localEndpoint(localEndpoint1). putTag("http.method", "GET"). putTag("http.url", "none+h1c: putTag("http.status_code", "200"). addAnnotation(startTime * 1000, "start processing"). build(); List<zipkin2.Span> zipkinSpanList = ImmutableList.of(spanServer1, spanServer2); SpanBytesEncoder encoder = SpanBytesEncoder.values()[1]; ByteBuf content = Unpooled.copiedBuffer(encoder.encodeList(zipkinSpanList)); reset(mockTraceHandler, mockTraceSpanLogsHandler); Span expectedSpan2 = Span.newBuilder().setCustomer("dummy").setStartMillis(startTime). setDuration(9). setName("getservice"). setSource(DEFAULT_SOURCE). setSpanId("00000000-0000-0000-3822-889fe47043bd"). setTraceId("00000000-0000-0000-3822-889fe47043bd"). setAnnotations(ImmutableList.of( new Annotation("zipkinSpanId", "3822889fe47043bd"), new Annotation("zipkinTraceId", "3822889fe47043bd"), new Annotation("span.kind", "server"), new Annotation("_spanSecondaryId", "server"), new Annotation("service", "frontend"), new Annotation("http.method", "GET"), new Annotation("http.status_code", "200"), new Annotation("http.url", "none+h1c: new Annotation("application", "Zipkin"), new Annotation("cluster", "none"), new Annotation("shard", "none"), new Annotation("ipv4", "10.0.0.1"), new Annotation("_spanLogs", "true"))). build(); mockTraceHandler.report(expectedSpan2); expectLastCall(); mockTraceSpanLogsHandler.report(SpanLogs.newBuilder(). setCustomer("default"). setTraceId("00000000-0000-0000-3822-889fe47043bd"). setSpanId("00000000-0000-0000-3822-889fe47043bd"). setSpanSecondaryId("server"). setLogs(ImmutableList.of( SpanLog.newBuilder(). setTimestamp(startTime * 1000). setFields(ImmutableMap.of("annotation", "start processing")). build() )). build()); expectLastCall(); replay(mockTraceHandler, mockTraceSpanLogsHandler); ChannelHandlerContext mockCtx = createNiceMock(ChannelHandlerContext.class); doMockLifecycle(mockCtx); FullHttpRequest httpRequest = new DefaultFullHttpRequest( HttpVersion.HTTP_1_1, HttpMethod.POST, "http: content, true ); handler.handleHttpMessage(mockCtx, httpRequest); verify(mockTraceHandler, mockTraceSpanLogsHandler); }
@Override protected void handleHttpMessage(final ChannelHandlerContext ctx, final FullHttpRequest request) throws URISyntaxException { URI uri = new URI(request.uri()); String path = uri.getPath().endsWith("/") ? uri.getPath() : uri.getPath() + "/"; if (!ZIPKIN_VALID_PATHS.contains(path)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported URL path.", request); logWarning("Requested URI path '" + path + "' is not supported.", null, ctx); return; } if (!request.method().toString().equalsIgnoreCase(ZIPKIN_VALID_HTTP_METHOD)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported Http method.", request); logWarning("Requested http method '" + request.method().toString() + "' is not supported.", null, ctx); return; } HttpResponseStatus status; StringBuilder output = new StringBuilder(); try { byte[] bytesArray = new byte[request.content().nioBuffer().remaining()]; request.content().nioBuffer().get(bytesArray, 0, bytesArray.length); BytesDecoder<zipkin2.Span> decoder = SpanBytesDecoderDetector.decoderForListMessage(bytesArray); List<zipkin2.Span> zipkinSpanSink = new ArrayList<>(); decoder.decodeList(bytesArray, zipkinSpanSink); if (isFeatureDisabled(traceDisabled, SPAN_DISABLED, discardedBatches, output)) { status = HttpResponseStatus.ACCEPTED; writeHttpResponse(ctx, status, output, request); discardedTraces.inc(zipkinSpanSink.size()); receivedSpansTotal.inc(zipkinSpanSink.size()); processedBatches.inc(); return; } receivedSpansTotal.inc(zipkinSpanSink.size()); processZipkinSpans(zipkinSpanSink); status = HttpResponseStatus.ACCEPTED; processedBatches.inc(); } catch (Exception e) { failedBatches.inc(); output.append(errorMessageWithRootCause(e)); status = HttpResponseStatus.BAD_REQUEST; logger.log(Level.WARNING, "Zipkin batch processing failed", Throwables.getRootCause(e)); } writeHttpResponse(ctx, status, output, request); }
ZipkinPortUnificationHandler extends AbstractHttpOnlyHandler implements Runnable, Closeable { @Override protected void handleHttpMessage(final ChannelHandlerContext ctx, final FullHttpRequest request) throws URISyntaxException { URI uri = new URI(request.uri()); String path = uri.getPath().endsWith("/") ? uri.getPath() : uri.getPath() + "/"; if (!ZIPKIN_VALID_PATHS.contains(path)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported URL path.", request); logWarning("Requested URI path '" + path + "' is not supported.", null, ctx); return; } if (!request.method().toString().equalsIgnoreCase(ZIPKIN_VALID_HTTP_METHOD)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported Http method.", request); logWarning("Requested http method '" + request.method().toString() + "' is not supported.", null, ctx); return; } HttpResponseStatus status; StringBuilder output = new StringBuilder(); try { byte[] bytesArray = new byte[request.content().nioBuffer().remaining()]; request.content().nioBuffer().get(bytesArray, 0, bytesArray.length); BytesDecoder<zipkin2.Span> decoder = SpanBytesDecoderDetector.decoderForListMessage(bytesArray); List<zipkin2.Span> zipkinSpanSink = new ArrayList<>(); decoder.decodeList(bytesArray, zipkinSpanSink); if (isFeatureDisabled(traceDisabled, SPAN_DISABLED, discardedBatches, output)) { status = HttpResponseStatus.ACCEPTED; writeHttpResponse(ctx, status, output, request); discardedTraces.inc(zipkinSpanSink.size()); receivedSpansTotal.inc(zipkinSpanSink.size()); processedBatches.inc(); return; } receivedSpansTotal.inc(zipkinSpanSink.size()); processZipkinSpans(zipkinSpanSink); status = HttpResponseStatus.ACCEPTED; processedBatches.inc(); } catch (Exception e) { failedBatches.inc(); output.append(errorMessageWithRootCause(e)); status = HttpResponseStatus.BAD_REQUEST; logger.log(Level.WARNING, "Zipkin batch processing failed", Throwables.getRootCause(e)); } writeHttpResponse(ctx, status, output, request); } }
ZipkinPortUnificationHandler extends AbstractHttpOnlyHandler implements Runnable, Closeable { @Override protected void handleHttpMessage(final ChannelHandlerContext ctx, final FullHttpRequest request) throws URISyntaxException { URI uri = new URI(request.uri()); String path = uri.getPath().endsWith("/") ? uri.getPath() : uri.getPath() + "/"; if (!ZIPKIN_VALID_PATHS.contains(path)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported URL path.", request); logWarning("Requested URI path '" + path + "' is not supported.", null, ctx); return; } if (!request.method().toString().equalsIgnoreCase(ZIPKIN_VALID_HTTP_METHOD)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported Http method.", request); logWarning("Requested http method '" + request.method().toString() + "' is not supported.", null, ctx); return; } HttpResponseStatus status; StringBuilder output = new StringBuilder(); try { byte[] bytesArray = new byte[request.content().nioBuffer().remaining()]; request.content().nioBuffer().get(bytesArray, 0, bytesArray.length); BytesDecoder<zipkin2.Span> decoder = SpanBytesDecoderDetector.decoderForListMessage(bytesArray); List<zipkin2.Span> zipkinSpanSink = new ArrayList<>(); decoder.decodeList(bytesArray, zipkinSpanSink); if (isFeatureDisabled(traceDisabled, SPAN_DISABLED, discardedBatches, output)) { status = HttpResponseStatus.ACCEPTED; writeHttpResponse(ctx, status, output, request); discardedTraces.inc(zipkinSpanSink.size()); receivedSpansTotal.inc(zipkinSpanSink.size()); processedBatches.inc(); return; } receivedSpansTotal.inc(zipkinSpanSink.size()); processZipkinSpans(zipkinSpanSink); status = HttpResponseStatus.ACCEPTED; processedBatches.inc(); } catch (Exception e) { failedBatches.inc(); output.append(errorMessageWithRootCause(e)); status = HttpResponseStatus.BAD_REQUEST; logger.log(Level.WARNING, "Zipkin batch processing failed", Throwables.getRootCause(e)); } writeHttpResponse(ctx, status, output, request); } ZipkinPortUnificationHandler(String handle,
                                      final HealthCheckManager healthCheckManager,
                                      ReportableEntityHandlerFactory handlerFactory,
                                      @Nullable WavefrontSender wfSender,
                                      Supplier<Boolean> traceDisabled,
                                      Supplier<Boolean> spanLogsDisabled,
                                      @Nullable Supplier<ReportableEntityPreprocessor> preprocessor,
                                      SpanSampler sampler,
                                      @Nullable String traceZipkinApplicationName,
                                      Set<String> traceDerivedCustomTagKeys); @VisibleForTesting  ZipkinPortUnificationHandler(final String handle,
                               final HealthCheckManager healthCheckManager,
                               ReportableEntityHandler<Span, String> spanHandler,
                               ReportableEntityHandler<SpanLogs, String> spanLogsHandler,
                               @Nullable WavefrontSender wfSender,
                               Supplier<Boolean> traceDisabled,
                               Supplier<Boolean> spanLogsDisabled,
                               @Nullable Supplier<ReportableEntityPreprocessor> preprocessor,
                               SpanSampler sampler,
                               @Nullable String traceZipkinApplicationName,
                               Set<String> traceDerivedCustomTagKeys); }
ZipkinPortUnificationHandler extends AbstractHttpOnlyHandler implements Runnable, Closeable { @Override protected void handleHttpMessage(final ChannelHandlerContext ctx, final FullHttpRequest request) throws URISyntaxException { URI uri = new URI(request.uri()); String path = uri.getPath().endsWith("/") ? uri.getPath() : uri.getPath() + "/"; if (!ZIPKIN_VALID_PATHS.contains(path)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported URL path.", request); logWarning("Requested URI path '" + path + "' is not supported.", null, ctx); return; } if (!request.method().toString().equalsIgnoreCase(ZIPKIN_VALID_HTTP_METHOD)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported Http method.", request); logWarning("Requested http method '" + request.method().toString() + "' is not supported.", null, ctx); return; } HttpResponseStatus status; StringBuilder output = new StringBuilder(); try { byte[] bytesArray = new byte[request.content().nioBuffer().remaining()]; request.content().nioBuffer().get(bytesArray, 0, bytesArray.length); BytesDecoder<zipkin2.Span> decoder = SpanBytesDecoderDetector.decoderForListMessage(bytesArray); List<zipkin2.Span> zipkinSpanSink = new ArrayList<>(); decoder.decodeList(bytesArray, zipkinSpanSink); if (isFeatureDisabled(traceDisabled, SPAN_DISABLED, discardedBatches, output)) { status = HttpResponseStatus.ACCEPTED; writeHttpResponse(ctx, status, output, request); discardedTraces.inc(zipkinSpanSink.size()); receivedSpansTotal.inc(zipkinSpanSink.size()); processedBatches.inc(); return; } receivedSpansTotal.inc(zipkinSpanSink.size()); processZipkinSpans(zipkinSpanSink); status = HttpResponseStatus.ACCEPTED; processedBatches.inc(); } catch (Exception e) { failedBatches.inc(); output.append(errorMessageWithRootCause(e)); status = HttpResponseStatus.BAD_REQUEST; logger.log(Level.WARNING, "Zipkin batch processing failed", Throwables.getRootCause(e)); } writeHttpResponse(ctx, status, output, request); } ZipkinPortUnificationHandler(String handle,
                                      final HealthCheckManager healthCheckManager,
                                      ReportableEntityHandlerFactory handlerFactory,
                                      @Nullable WavefrontSender wfSender,
                                      Supplier<Boolean> traceDisabled,
                                      Supplier<Boolean> spanLogsDisabled,
                                      @Nullable Supplier<ReportableEntityPreprocessor> preprocessor,
                                      SpanSampler sampler,
                                      @Nullable String traceZipkinApplicationName,
                                      Set<String> traceDerivedCustomTagKeys); @VisibleForTesting  ZipkinPortUnificationHandler(final String handle,
                               final HealthCheckManager healthCheckManager,
                               ReportableEntityHandler<Span, String> spanHandler,
                               ReportableEntityHandler<SpanLogs, String> spanLogsHandler,
                               @Nullable WavefrontSender wfSender,
                               Supplier<Boolean> traceDisabled,
                               Supplier<Boolean> spanLogsDisabled,
                               @Nullable Supplier<ReportableEntityPreprocessor> preprocessor,
                               SpanSampler sampler,
                               @Nullable String traceZipkinApplicationName,
                               Set<String> traceDerivedCustomTagKeys); @Override void run(); @Override void close(); }
ZipkinPortUnificationHandler extends AbstractHttpOnlyHandler implements Runnable, Closeable { @Override protected void handleHttpMessage(final ChannelHandlerContext ctx, final FullHttpRequest request) throws URISyntaxException { URI uri = new URI(request.uri()); String path = uri.getPath().endsWith("/") ? uri.getPath() : uri.getPath() + "/"; if (!ZIPKIN_VALID_PATHS.contains(path)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported URL path.", request); logWarning("Requested URI path '" + path + "' is not supported.", null, ctx); return; } if (!request.method().toString().equalsIgnoreCase(ZIPKIN_VALID_HTTP_METHOD)) { writeHttpResponse(ctx, HttpResponseStatus.BAD_REQUEST, "Unsupported Http method.", request); logWarning("Requested http method '" + request.method().toString() + "' is not supported.", null, ctx); return; } HttpResponseStatus status; StringBuilder output = new StringBuilder(); try { byte[] bytesArray = new byte[request.content().nioBuffer().remaining()]; request.content().nioBuffer().get(bytesArray, 0, bytesArray.length); BytesDecoder<zipkin2.Span> decoder = SpanBytesDecoderDetector.decoderForListMessage(bytesArray); List<zipkin2.Span> zipkinSpanSink = new ArrayList<>(); decoder.decodeList(bytesArray, zipkinSpanSink); if (isFeatureDisabled(traceDisabled, SPAN_DISABLED, discardedBatches, output)) { status = HttpResponseStatus.ACCEPTED; writeHttpResponse(ctx, status, output, request); discardedTraces.inc(zipkinSpanSink.size()); receivedSpansTotal.inc(zipkinSpanSink.size()); processedBatches.inc(); return; } receivedSpansTotal.inc(zipkinSpanSink.size()); processZipkinSpans(zipkinSpanSink); status = HttpResponseStatus.ACCEPTED; processedBatches.inc(); } catch (Exception e) { failedBatches.inc(); output.append(errorMessageWithRootCause(e)); status = HttpResponseStatus.BAD_REQUEST; logger.log(Level.WARNING, "Zipkin batch processing failed", Throwables.getRootCause(e)); } writeHttpResponse(ctx, status, output, request); } ZipkinPortUnificationHandler(String handle,
                                      final HealthCheckManager healthCheckManager,
                                      ReportableEntityHandlerFactory handlerFactory,
                                      @Nullable WavefrontSender wfSender,
                                      Supplier<Boolean> traceDisabled,
                                      Supplier<Boolean> spanLogsDisabled,
                                      @Nullable Supplier<ReportableEntityPreprocessor> preprocessor,
                                      SpanSampler sampler,
                                      @Nullable String traceZipkinApplicationName,
                                      Set<String> traceDerivedCustomTagKeys); @VisibleForTesting  ZipkinPortUnificationHandler(final String handle,
                               final HealthCheckManager healthCheckManager,
                               ReportableEntityHandler<Span, String> spanHandler,
                               ReportableEntityHandler<SpanLogs, String> spanLogsHandler,
                               @Nullable WavefrontSender wfSender,
                               Supplier<Boolean> traceDisabled,
                               Supplier<Boolean> spanLogsDisabled,
                               @Nullable Supplier<ReportableEntityPreprocessor> preprocessor,
                               SpanSampler sampler,
                               @Nullable String traceZipkinApplicationName,
                               Set<String> traceDerivedCustomTagKeys); @Override void run(); @Override void close();  }