From 547a124d41e589d7c53ec3d9f577eb166babfcc0 Mon Sep 17 00:00:00 2001 From: Adrian Cole Date: Mon, 24 Feb 2020 15:46:35 +0800 Subject: [PATCH] Adds Brave tests for reactor-netty (#1554) --- spring-cloud-sleuth-core/pom.xml | 5 + .../ReactorNettyHttpClientBraveTests.java | 187 ++++++++++++++++++ 2 files changed, 192 insertions(+) create mode 100644 spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java diff --git a/spring-cloud-sleuth-core/pom.xml b/spring-cloud-sleuth-core/pom.xml index 9d7157d31..af72b90f3 100644 --- a/spring-cloud-sleuth-core/pom.xml +++ b/spring-cloud-sleuth-core/pom.xml @@ -335,6 +335,11 @@ spring-boot-starter-test test + + io.zipkin.brave + brave-instrumentation-http-tests + test + com.netflix.archaius archaius-core diff --git a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java new file mode 100644 index 000000000..fe84c6529 --- /dev/null +++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/ReactorNettyHttpClientBraveTests.java @@ -0,0 +1,187 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.sleuth.instrument.web.client; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import brave.http.HttpTracing; +import brave.test.http.ITHttpAsyncClient; +import io.netty.channel.ChannelOption; +import io.netty.handler.timeout.ReadTimeoutHandler; +import org.junit.After; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.reactivestreams.Subscription; +import reactor.core.CoreSubscriber; +import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Operators; +import reactor.core.scheduler.Schedulers; +import reactor.netty.ByteBufFlux; +import reactor.netty.http.client.HttpClient; +import reactor.netty.http.client.HttpClientResponse; +import reactor.util.context.Context; +import zipkin2.Callback; + +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; + +/** + * This runs Brave's integration tests, ensuring common instrumentation bugs aren't + * present. + */ +public class ReactorNettyHttpClientBraveTests extends ITHttpAsyncClient { + @Before + @After + public void resetHooks() { + // There's an assumption some other test is leaking hooks, so we clear them all to + // prevent should_not_scope_scalar_subscribe from being interfered with. + Hooks.resetOnEachOperator(); + Hooks.resetOnLastOperator(); + Schedulers.removeExecutorServiceDecorator("sleuth"); + } + + /** + * This uses Spring to instrument the {@link HttpClient} using a + * {@link BeanPostProcessor}. + */ + @Override + protected HttpClient newClient(int port) { + AnnotationConfigApplicationContext result = new AnnotationConfigApplicationContext(); + result.registerBean(HttpTracing.class, () -> httpTracing); + result.registerBean(HttpClient.class, + ReactorNettyHttpClientBraveTests::testHttpClient); + result.register(HttpClientBeanPostProcessor.class); + result.refresh(); + return result.getBean(HttpClient.class).baseUrl("http://127.0.0.1:" + port); + } + + static HttpClient testHttpClient() { + return HttpClient.create() + .tcpConfiguration(tcpClient -> tcpClient + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000) + .doOnConnected(conn -> conn + .addHandler(new ReadTimeoutHandler(1, TimeUnit.SECONDS)))) + .followRedirect(true); + } + + @Override + protected void closeClient(HttpClient client) { + // HttpClient is not Closeable + } + + @Override + protected void get(HttpClient client, String pathIncludingQuery) { + client.get().uri(pathIncludingQuery).response().block(); + } + + @Test + @Ignore("TODO: consider integrating TracingMapConnect with ScopePassingSpanSubscriber") + @Override + public void callbackContextIsFromInvocationTime() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void redirect() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void supportsPortableCustomization() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void post() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void customSampler() { + } + + @Test + @Ignore("TODO: reactor/reactor-netty#1000") + @Override + public void httpPathTagExcludesQueryParams() { + } + + @Override + protected void post(HttpClient client, String pathIncludingQuery, String body) { + client.post().send(ByteBufFlux.fromString(Mono.just(body))) + .uri(pathIncludingQuery).response().block(); + } + + @Override + protected void getAsync(HttpClient client, String path, Callback callback) { + Mono request = client.get().uri(path).response(); + + request.subscribe(new CoreSubscriber() { + + final AtomicReference ref = new AtomicReference<>(); + + @Override + public void onSubscribe(Subscription s) { + if (Operators.validate(ref.getAndSet(s), s)) { + s.request(Long.MAX_VALUE); + } + else { + s.cancel(); + } + } + + @Override + public void onNext(HttpClientResponse t) { + Subscription s = ref.getAndSet(null); + if (s != null) { + callback.onSuccess(null); + s.cancel(); + } + else { + Operators.onNextDropped(t, currentContext()); + } + } + + @Override + public void onError(Throwable t) { + if (ref.getAndSet(null) != null) { + callback.onError(t); + } + } + + @Override + public void onComplete() { + if (ref.getAndSet(null) != null) { + callback.onSuccess(null); + } + } + + @Override + public Context currentContext() { + return Context.empty(); + } + }); + } + +}