Fixed the messaging aspect
This commit is contained in:
Marcin Grzejszczak
@@ -1,92 +1,92 @@
|
||||
/*
|
||||
* Copyright 2018-2021 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.deployer;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.cloud.deployer.spi.app.AppDeployer;
|
||||
import org.springframework.cloud.deployer.spi.app.AppScaleRequest;
|
||||
import org.springframework.cloud.deployer.spi.app.AppStatus;
|
||||
import org.springframework.cloud.deployer.spi.app.DeploymentState;
|
||||
import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
|
||||
import org.springframework.cloud.deployer.spi.core.RuntimeEnvironmentInfo;
|
||||
import org.springframework.cloud.sleuth.CurrentTraceContext;
|
||||
import org.springframework.cloud.sleuth.Span;
|
||||
import org.springframework.cloud.sleuth.Tracer;
|
||||
import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth;
|
||||
import org.springframework.core.env.Environment;
|
||||
|
||||
/**
|
||||
* Trace representation of an {@link AppDeployer}.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.1.0
|
||||
*/
|
||||
public class TraceAppDeployer implements AppDeployer {
|
||||
|
||||
private static final Log log = LogFactory.getLog(TraceAppDeployer.class);
|
||||
|
||||
private final AppDeployer delegate;
|
||||
|
||||
private final BeanFactory beanFactory;
|
||||
|
||||
private final Environment environment;
|
||||
|
||||
private Tracer tracer;
|
||||
|
||||
private CurrentTraceContext currentTraceContext;
|
||||
|
||||
private Long pollDelay;
|
||||
|
||||
public TraceAppDeployer(AppDeployer delegate, BeanFactory beanFactory, Environment environment) {
|
||||
this.delegate = delegate;
|
||||
this.beanFactory = beanFactory;
|
||||
this.environment = environment;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String deploy(AppDeploymentRequest request) {
|
||||
Span span = tracer().nextSpan().name("deploy");
|
||||
// TODO: Is this secure to pass?
|
||||
// TODO: Does it make sense?
|
||||
// if (!request.getCommandlineArguments().isEmpty()) {
|
||||
// span.tag("commandlineArguments", request.getCommandlineArguments().toString());
|
||||
// }
|
||||
// if (!request.getDeploymentProperties().isEmpty()) {
|
||||
// span.tag("deploymentProperties", request.getDeploymentProperties().toString());
|
||||
// }
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
span.event("start");
|
||||
String id = this.delegate.deploy(request);
|
||||
span.tag("deployer.app.id", id);
|
||||
registerListener(span, id);
|
||||
return id;
|
||||
}
|
||||
}
|
||||
|
||||
private void registerListener(Span span, String id) {
|
||||
PreviousAndCurrentStatus previousAndCurrentStatus = new PreviousAndCurrentStatus(span);
|
||||
/*
|
||||
* Copyright 2018-2021 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.deployer;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.cloud.deployer.spi.app.AppDeployer;
|
||||
import org.springframework.cloud.deployer.spi.app.AppScaleRequest;
|
||||
import org.springframework.cloud.deployer.spi.app.AppStatus;
|
||||
import org.springframework.cloud.deployer.spi.app.DeploymentState;
|
||||
import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
|
||||
import org.springframework.cloud.deployer.spi.core.RuntimeEnvironmentInfo;
|
||||
import org.springframework.cloud.sleuth.CurrentTraceContext;
|
||||
import org.springframework.cloud.sleuth.Span;
|
||||
import org.springframework.cloud.sleuth.Tracer;
|
||||
import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth;
|
||||
import org.springframework.core.env.Environment;
|
||||
|
||||
/**
|
||||
* Trace representation of an {@link AppDeployer}.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @since 3.1.0
|
||||
*/
|
||||
public class TraceAppDeployer implements AppDeployer {
|
||||
|
||||
private static final Log log = LogFactory.getLog(TraceAppDeployer.class);
|
||||
|
||||
private final AppDeployer delegate;
|
||||
|
||||
private final BeanFactory beanFactory;
|
||||
|
||||
private final Environment environment;
|
||||
|
||||
private Tracer tracer;
|
||||
|
||||
private CurrentTraceContext currentTraceContext;
|
||||
|
||||
private Long pollDelay;
|
||||
|
||||
public TraceAppDeployer(AppDeployer delegate, BeanFactory beanFactory, Environment environment) {
|
||||
this.delegate = delegate;
|
||||
this.beanFactory = beanFactory;
|
||||
this.environment = environment;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String deploy(AppDeploymentRequest request) {
|
||||
Span span = tracer().nextSpan().name("deploy");
|
||||
// TODO: Is this secure to pass?
|
||||
// TODO: Does it make sense?
|
||||
// if (!request.getCommandlineArguments().isEmpty()) {
|
||||
// span.tag("commandlineArguments", request.getCommandlineArguments().toString());
|
||||
// }
|
||||
// if (!request.getDeploymentProperties().isEmpty()) {
|
||||
// span.tag("deploymentProperties", request.getDeploymentProperties().toString());
|
||||
// }
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
span.event("start");
|
||||
String id = this.delegate.deploy(request);
|
||||
span.tag("deployer.app.id", id);
|
||||
registerListener(span, id);
|
||||
return id;
|
||||
}
|
||||
}
|
||||
|
||||
private void registerListener(Span span, String id) {
|
||||
PreviousAndCurrentStatus previousAndCurrentStatus = new PreviousAndCurrentStatus(span);
|
||||
// @formatter:off
|
||||
this.delegate.statusReactive(id)
|
||||
.map(previousAndCurrentStatus::updateCurrent)
|
||||
@@ -97,172 +97,172 @@ public class TraceAppDeployer implements AppDeployer {
|
||||
.doOnError(span::error)
|
||||
// we will close the span in the reactive part
|
||||
.doFinally(signalType -> span.end()).subscribe();
|
||||
// @formatter:on
|
||||
}
|
||||
|
||||
@Override
|
||||
public void undeploy(String id) {
|
||||
Span span = tracer().nextSpan().name("undeploy");
|
||||
span.tag("deployer.app.id", id);
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
span.event("start");
|
||||
this.delegate.undeploy(id);
|
||||
registerListener(span, id);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public AppStatus status(String id) {
|
||||
Span span = tracer().nextSpan().name("status");
|
||||
span.tag("deployer.app.id", id);
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
return this.delegate.status(id);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<AppStatus> statusReactive(String id) {
|
||||
return ReactorSleuth.tracedMono(tracer(), currentTraceContext(), "status",
|
||||
() -> this.delegate.statusReactive(id), span -> span.tag("deployer.app.id", id));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<AppStatus> statusesReactive(String... ids) {
|
||||
return ReactorSleuth.tracedFlux(tracer(), currentTraceContext(), "statuses",
|
||||
() -> this.delegate.statusesReactive(ids), span -> span.tag("deployer.app.ids", Arrays.toString(ids)));
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeEnvironmentInfo environmentInfo() {
|
||||
return this.delegate.environmentInfo();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getLog(String id) {
|
||||
Span span = tracer().nextSpan().name("getLog");
|
||||
span.tag("deployer.app.id", id);
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
return this.delegate.getLog(id);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void scale(AppScaleRequest appScaleRequest) {
|
||||
Span span = tracer().nextSpan().name("scale");
|
||||
span.tag("deployer.scale.deploymentId", appScaleRequest.getDeploymentId());
|
||||
span.tag("deployer.scale.count", String.valueOf(appScaleRequest.getCount()));
|
||||
// TODO: Is this secure to pass?
|
||||
// TODO: Does it make sense?
|
||||
// if (appScaleRequest.getProperties().isPresent() &&
|
||||
// !appScaleRequest.getProperties().get().isEmpty()) {
|
||||
// span.tag("properties", appScaleRequest.getProperties().get().toString());
|
||||
// }
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
this.delegate.scale(appScaleRequest);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
private Tracer tracer() {
|
||||
if (this.tracer == null) {
|
||||
this.tracer = this.beanFactory.getBean(Tracer.class);
|
||||
}
|
||||
return this.tracer;
|
||||
}
|
||||
|
||||
private CurrentTraceContext currentTraceContext() {
|
||||
if (this.currentTraceContext == null) {
|
||||
this.currentTraceContext = this.beanFactory.getBean(CurrentTraceContext.class);
|
||||
}
|
||||
return this.currentTraceContext;
|
||||
}
|
||||
|
||||
private long pollDelay() {
|
||||
if (this.pollDelay == null) {
|
||||
this.pollDelay = this.environment.getProperty("spring.sleuth.deployer.status-poll-delay", Long.class, 500L);
|
||||
}
|
||||
return this.pollDelay;
|
||||
}
|
||||
|
||||
private static final class PreviousAndCurrentStatus {
|
||||
|
||||
private final Span span;
|
||||
|
||||
private AppStatus current;
|
||||
|
||||
private AppStatus previous;
|
||||
|
||||
private PreviousAndCurrentStatus(Span span) {
|
||||
this.span = span;
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Current span is [" + span + "]");
|
||||
}
|
||||
}
|
||||
|
||||
private PreviousAndCurrentStatus updateCurrent(AppStatus current) {
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("State before change: current [" + this.current + "], previous [" + this.previous + "]");
|
||||
}
|
||||
this.previous = this.current;
|
||||
this.current = current;
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("State after change: current [" + this.current + "], previous [" + this.previous + "]");
|
||||
}
|
||||
if (statusChanged()) {
|
||||
annotateSpan();
|
||||
}
|
||||
else if (log.isTraceEnabled()) {
|
||||
log.trace("State has not changed, will not annotate the span");
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
private void annotateSpan() {
|
||||
String name = this.current.getState().name();
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Will annotate its state with [" + name + "]");
|
||||
}
|
||||
this.span.event(name);
|
||||
}
|
||||
|
||||
private boolean statusChanged() {
|
||||
if (this.previous == null && this.current != null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Previous is null, current is not null");
|
||||
}
|
||||
return true;
|
||||
}
|
||||
else if (this.current == null) {
|
||||
throw new IllegalStateException("Current state can't be null");
|
||||
}
|
||||
DeploymentState currentState = this.current.getState();
|
||||
DeploymentState previousState = this.previous.getState();
|
||||
return currentState != previousState;
|
||||
}
|
||||
|
||||
private boolean isFinished() {
|
||||
boolean finished = this.current.getState() == DeploymentState.deployed
|
||||
|| this.current.getState() == DeploymentState.undeployed
|
||||
|| this.current.getState() == DeploymentState.failed
|
||||
|| this.current.getState() == DeploymentState.error
|
||||
|| this.current.getState() == DeploymentState.unknown;
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("Status is finished [" + finished + "]");
|
||||
}
|
||||
return finished;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
// @formatter:on
|
||||
}
|
||||
|
||||
@Override
|
||||
public void undeploy(String id) {
|
||||
Span span = tracer().nextSpan().name("undeploy");
|
||||
span.tag("deployer.app.id", id);
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
span.event("start");
|
||||
this.delegate.undeploy(id);
|
||||
registerListener(span, id);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public AppStatus status(String id) {
|
||||
Span span = tracer().nextSpan().name("status");
|
||||
span.tag("deployer.app.id", id);
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
return this.delegate.status(id);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Mono<AppStatus> statusReactive(String id) {
|
||||
return ReactorSleuth.tracedMono(tracer(), currentTraceContext(), "status",
|
||||
() -> this.delegate.statusReactive(id), span -> span.tag("deployer.app.id", id));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<AppStatus> statusesReactive(String... ids) {
|
||||
return ReactorSleuth.tracedFlux(tracer(), currentTraceContext(), "statuses",
|
||||
() -> this.delegate.statusesReactive(ids), span -> span.tag("deployer.app.ids", Arrays.toString(ids)));
|
||||
}
|
||||
|
||||
@Override
|
||||
public RuntimeEnvironmentInfo environmentInfo() {
|
||||
return this.delegate.environmentInfo();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getLog(String id) {
|
||||
Span span = tracer().nextSpan().name("getLog");
|
||||
span.tag("deployer.app.id", id);
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
return this.delegate.getLog(id);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void scale(AppScaleRequest appScaleRequest) {
|
||||
Span span = tracer().nextSpan().name("scale");
|
||||
span.tag("deployer.scale.deploymentId", appScaleRequest.getDeploymentId());
|
||||
span.tag("deployer.scale.count", String.valueOf(appScaleRequest.getCount()));
|
||||
// TODO: Is this secure to pass?
|
||||
// TODO: Does it make sense?
|
||||
// if (appScaleRequest.getProperties().isPresent() &&
|
||||
// !appScaleRequest.getProperties().get().isEmpty()) {
|
||||
// span.tag("properties", appScaleRequest.getProperties().get().toString());
|
||||
// }
|
||||
try (Tracer.SpanInScope spanInScope = tracer().withSpan(span.start())) {
|
||||
this.delegate.scale(appScaleRequest);
|
||||
}
|
||||
finally {
|
||||
span.end();
|
||||
}
|
||||
}
|
||||
|
||||
private Tracer tracer() {
|
||||
if (this.tracer == null) {
|
||||
this.tracer = this.beanFactory.getBean(Tracer.class);
|
||||
}
|
||||
return this.tracer;
|
||||
}
|
||||
|
||||
private CurrentTraceContext currentTraceContext() {
|
||||
if (this.currentTraceContext == null) {
|
||||
this.currentTraceContext = this.beanFactory.getBean(CurrentTraceContext.class);
|
||||
}
|
||||
return this.currentTraceContext;
|
||||
}
|
||||
|
||||
private long pollDelay() {
|
||||
if (this.pollDelay == null) {
|
||||
this.pollDelay = this.environment.getProperty("spring.sleuth.deployer.status-poll-delay", Long.class, 500L);
|
||||
}
|
||||
return this.pollDelay;
|
||||
}
|
||||
|
||||
private static final class PreviousAndCurrentStatus {
|
||||
|
||||
private final Span span;
|
||||
|
||||
private AppStatus current;
|
||||
|
||||
private AppStatus previous;
|
||||
|
||||
private PreviousAndCurrentStatus(Span span) {
|
||||
this.span = span;
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Current span is [" + span + "]");
|
||||
}
|
||||
}
|
||||
|
||||
private PreviousAndCurrentStatus updateCurrent(AppStatus current) {
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("State before change: current [" + this.current + "], previous [" + this.previous + "]");
|
||||
}
|
||||
this.previous = this.current;
|
||||
this.current = current;
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("State after change: current [" + this.current + "], previous [" + this.previous + "]");
|
||||
}
|
||||
if (statusChanged()) {
|
||||
annotateSpan();
|
||||
}
|
||||
else if (log.isTraceEnabled()) {
|
||||
log.trace("State has not changed, will not annotate the span");
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
private void annotateSpan() {
|
||||
String name = this.current.getState().name();
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Will annotate its state with [" + name + "]");
|
||||
}
|
||||
this.span.event(name);
|
||||
}
|
||||
|
||||
private boolean statusChanged() {
|
||||
if (this.previous == null && this.current != null) {
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Previous is null, current is not null");
|
||||
}
|
||||
return true;
|
||||
}
|
||||
else if (this.current == null) {
|
||||
throw new IllegalStateException("Current state can't be null");
|
||||
}
|
||||
DeploymentState currentState = this.current.getState();
|
||||
DeploymentState previousState = this.previous.getState();
|
||||
return currentState != previousState;
|
||||
}
|
||||
|
||||
private boolean isFinished() {
|
||||
boolean finished = this.current.getState() == DeploymentState.deployed
|
||||
|| this.current.getState() == DeploymentState.undeployed
|
||||
|| this.current.getState() == DeploymentState.failed
|
||||
|| this.current.getState() == DeploymentState.error
|
||||
|| this.current.getState() == DeploymentState.unknown;
|
||||
if (log.isTraceEnabled()) {
|
||||
log.trace("Status is finished [" + finished + "]");
|
||||
}
|
||||
return finished;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -59,7 +59,7 @@ public class TraceMessagingAspect {
|
||||
this.spanNamer = spanNamer;
|
||||
}
|
||||
|
||||
@Pointcut("@within(org.springframework.messaging.handler.annotation.MessageMapping)")
|
||||
@Pointcut("@annotation(org.springframework.messaging.handler.annotation.MessageMapping)")
|
||||
private void anyMessageMappingAnnotated() {
|
||||
} // NOSONAR
|
||||
|
||||
|
||||
@@ -1,363 +1,367 @@
|
||||
/*
|
||||
* Copyright 2013-2021 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.rsocket;
|
||||
|
||||
import java.net.URI;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
|
||||
import brave.Span;
|
||||
import brave.Tracer;
|
||||
import brave.test.TestSpanHandler;
|
||||
import io.rsocket.frame.FrameType;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.util.context.ContextView;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.sleuth.TraceContext;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.env.Environment;
|
||||
import org.springframework.messaging.handler.annotation.MessageMapping;
|
||||
import org.springframework.messaging.handler.annotation.Payload;
|
||||
import org.springframework.messaging.rsocket.RSocketRequester;
|
||||
import org.springframework.messaging.rsocket.RSocketRequester.Builder;
|
||||
import org.springframework.messaging.rsocket.RSocketStrategies;
|
||||
import org.springframework.stereotype.Controller;
|
||||
import org.springframework.util.MimeType;
|
||||
|
||||
import static org.assertj.core.api.BDDAssertions.then;
|
||||
|
||||
public abstract class TraceRSocketTests {
|
||||
|
||||
public static final String EXPECTED_TRACE_ID = "b919095138aa4c6e";
|
||||
|
||||
@Test
|
||||
public void should_instrument_responder() throws Exception {
|
||||
// setup
|
||||
ConfigurableApplicationContext context = new SpringApplicationBuilder(MyConfig.class, testConfiguration())
|
||||
.web(WebApplicationType.REACTIVE)
|
||||
.properties("server.port=0", "spring.rsocket.server.transport=websocket",
|
||||
"spring.rsocket.server.mapping-path=/rsocket", "spring.jmx.enabled=false",
|
||||
"spring.application.name=TraceRSocketTests", "security.basic.enabled=false",
|
||||
"management.security.enabled=false")
|
||||
.run();
|
||||
final TestSpanHandler spans = context.getBean(TestSpanHandler.class);
|
||||
final int port = context.getBean(Environment.class).getProperty("local.server.port", Integer.class);
|
||||
final TestController controller2 = context.getBean(TestController.class);
|
||||
final RSocketStrategies strategies = context.getBean(RSocketStrategies.class);
|
||||
|
||||
final Builder rsocketRequesterBuilder = RSocketRequester.builder().rsocketStrategies(strategies);
|
||||
|
||||
final RSocketRequester rSocketRequester = rsocketRequesterBuilder
|
||||
.websocket(URI.create("ws://localhost:" + port + "/rsocket"));
|
||||
|
||||
// REQUEST FNF
|
||||
whenRequestFnFIsSent(rSocketRequester, "api.c2.fnf").block();
|
||||
|
||||
FrameType receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.fnf", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST RESPONSE
|
||||
whenRequestResponseIsSent(rSocketRequester, "api.c2.rr").block();
|
||||
|
||||
receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.rr", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST STREAM
|
||||
whenRequestStreamIsSent(rSocketRequester, "api.c2.rs").blockLast();
|
||||
|
||||
receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.rs", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST CHANNEL
|
||||
whenRequestChannelIsSent(rSocketRequester, "api.c2.rc").blockLast();
|
||||
|
||||
receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.rc", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST FNF
|
||||
whenNonSampledRequestFnfIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST RESPONSE
|
||||
whenNonSampledRequestResponseIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST STREAM
|
||||
whenNonSampledRequestStreamIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST CHANNEL
|
||||
whenNonSampledRequestChannelIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// cleanup
|
||||
context.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void should_instrument_requester_and_responder() throws Exception {
|
||||
// setup
|
||||
ConfigurableApplicationContext context = new SpringApplicationBuilder(MyConfig.class, testConfiguration())
|
||||
.web(WebApplicationType.REACTIVE)
|
||||
.properties("server.port=0", "spring.rsocket.server.transport=websocket",
|
||||
"spring.rsocket.server.mapping-path=/rsocket", "spring.jmx.enabled=false",
|
||||
"spring.application.name=TraceRSocketTests", "security.basic.enabled=false",
|
||||
"management.security.enabled=false")
|
||||
.run();
|
||||
|
||||
final org.springframework.cloud.sleuth.Tracer tracer = context
|
||||
.getBean(org.springframework.cloud.sleuth.Tracer.class);
|
||||
final TestSpanHandler spans = context.getBean(TestSpanHandler.class);
|
||||
final int port = context.getBean(Environment.class).getProperty("local.server.port", Integer.class);
|
||||
final TestController controller2 = context.getBean(TestController.class);
|
||||
|
||||
final Builder rsocketRequesterBuilder = context.getBean(Builder.class);
|
||||
|
||||
final RSocketRequester rSocketRequester = rsocketRequesterBuilder
|
||||
.websocket(URI.create("ws://localhost:" + port + "/rsocket"));
|
||||
|
||||
// REQUEST FNF
|
||||
final org.springframework.cloud.sleuth.Span nextSpanFnf = tracer.nextSpan().start();
|
||||
whenRequestFnFIsSent(rSocketRequester, "api.c2.fnf")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanFnf.context()))
|
||||
.doFinally(signalType -> nextSpanFnf.end()).block();
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanFnf.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST RESPONSE
|
||||
final org.springframework.cloud.sleuth.Span nextSpanRR = tracer.nextSpan().start();
|
||||
whenRequestResponseIsSent(rSocketRequester, "api.c2.rr")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanRR.context()))
|
||||
.doFinally(signalType -> nextSpanRR.end()).block();
|
||||
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanRR.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST STREAM
|
||||
final org.springframework.cloud.sleuth.Span nextSpanRS = tracer.nextSpan().start();
|
||||
whenRequestStreamIsSent(rSocketRequester, "api.c2.rs")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanRS.context()))
|
||||
.doFinally(signalType -> nextSpanRS.end()).blockLast();
|
||||
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanRS.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST CHANNEL
|
||||
final org.springframework.cloud.sleuth.Span nextSpanRC = tracer.nextSpan().start();
|
||||
whenRequestChannelIsSent(rSocketRequester, "api.c2.rc")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanRC.context()))
|
||||
.doFinally(signalType -> nextSpanRC.end()).blockLast();
|
||||
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanRC.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// cleanup
|
||||
context.close();
|
||||
}
|
||||
|
||||
protected abstract Class testConfiguration();
|
||||
|
||||
private void thenSpanWasReportedWithTags(TestSpanHandler spans, String path, FrameType frameType) {
|
||||
then(spans).hasSize(1);
|
||||
// TODO: Preferred option would be : [api.c2.{name}]
|
||||
then(spans.get(0).name()).isEqualTo(frameType.name() + " " + path);
|
||||
}
|
||||
|
||||
private Mono<Void> whenRequestFnFIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).send();
|
||||
}
|
||||
|
||||
private Mono<String> whenRequestResponseIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).retrieveMono(String.class);
|
||||
}
|
||||
|
||||
private Flux<String> whenRequestStreamIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).retrieveFlux(String.class);
|
||||
}
|
||||
|
||||
private Flux<String> whenRequestChannelIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).data(Flux.fromArray(new String[] { "test1", "test2" })).retrieveFlux(String.class);
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestFnfIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.fnf").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).send().block();
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestResponseIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.rr").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).retrieveMono(String.class).block();
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestStreamIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.rs").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).retrieveFlux(String.class).blockLast();
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestChannelIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.rc").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).data(Flux.fromArray(new String[] { "test1", "test2" })).retrieveFlux(String.class).blockLast();
|
||||
}
|
||||
|
||||
private void thenNoSpanWasReported(TestSpanHandler spans, TestController controller2, String expectedTraceId) {
|
||||
// then(spans).isEmpty(); // FIXME: does not work for request case
|
||||
then(controller2.getSpan()).isNotNull();
|
||||
then(controller2.getSpan().context().traceIdString()).isEqualTo(expectedTraceId);
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableAutoConfiguration
|
||||
static class MyConfig {
|
||||
|
||||
@Bean
|
||||
TestController controller(Tracer tracer) {
|
||||
return new TestController(tracer);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Controller
|
||||
@MessageMapping("api.c2")
|
||||
static class TestController {
|
||||
|
||||
final Tracer tracer;
|
||||
|
||||
Span span;
|
||||
|
||||
ContextView interceptedContext;
|
||||
|
||||
BlockingQueue<FrameType> receivedFrames = new LinkedBlockingDeque<>();
|
||||
|
||||
TestController(Tracer tracer) {
|
||||
this.tracer = tracer;
|
||||
}
|
||||
|
||||
BlockingQueue<FrameType> getReceivedFrames() {
|
||||
return this.receivedFrames;
|
||||
}
|
||||
|
||||
Span getSpan() {
|
||||
return this.span;
|
||||
}
|
||||
|
||||
void reset() {
|
||||
this.span = null;
|
||||
}
|
||||
|
||||
@MessageMapping("fnf")
|
||||
Mono<Void> testFnf() {
|
||||
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Mono.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_FNF);
|
||||
return Mono.empty();
|
||||
});
|
||||
}
|
||||
|
||||
@MessageMapping("rr")
|
||||
Mono<String> testRR() {
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Mono.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_RESPONSE);
|
||||
return Mono.just("response");
|
||||
});
|
||||
}
|
||||
|
||||
@MessageMapping("rs")
|
||||
Flux<String> testRS() {
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Flux.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_STREAM);
|
||||
return Flux.just("stream");
|
||||
});
|
||||
}
|
||||
|
||||
@MessageMapping("rc")
|
||||
Flux<String> testRC(@Payload Flux<String> inbound) {
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Flux.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_CHANNEL);
|
||||
return inbound;
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2013-2021 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.rsocket;
|
||||
|
||||
import java.net.URI;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
|
||||
import io.rsocket.frame.FrameType;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.util.context.ContextView;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.sleuth.Span;
|
||||
import org.springframework.cloud.sleuth.TraceContext;
|
||||
import org.springframework.cloud.sleuth.Tracer;
|
||||
import org.springframework.cloud.sleuth.exporter.FinishedSpan;
|
||||
import org.springframework.cloud.sleuth.test.TestSpanHandler;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.env.Environment;
|
||||
import org.springframework.messaging.handler.annotation.MessageMapping;
|
||||
import org.springframework.messaging.handler.annotation.Payload;
|
||||
import org.springframework.messaging.rsocket.RSocketRequester;
|
||||
import org.springframework.messaging.rsocket.RSocketRequester.Builder;
|
||||
import org.springframework.messaging.rsocket.RSocketStrategies;
|
||||
import org.springframework.stereotype.Controller;
|
||||
import org.springframework.util.MimeType;
|
||||
|
||||
import static org.assertj.core.api.BDDAssertions.then;
|
||||
|
||||
public abstract class TraceRSocketTests {
|
||||
|
||||
public static final String EXPECTED_TRACE_ID = "b919095138aa4c6e";
|
||||
|
||||
@Test
|
||||
public void should_instrument_responder() throws Exception {
|
||||
// setup
|
||||
ConfigurableApplicationContext context = new SpringApplicationBuilder(MyConfig.class, testConfiguration())
|
||||
.web(WebApplicationType.REACTIVE)
|
||||
.properties("server.port=0", "spring.rsocket.server.transport=websocket",
|
||||
"spring.rsocket.server.mapping-path=/rsocket", "spring.jmx.enabled=false",
|
||||
"spring.application.name=TraceRSocketTests", "security.basic.enabled=false",
|
||||
"management.security.enabled=false")
|
||||
.run();
|
||||
final TestSpanHandler spans = context.getBean(TestSpanHandler.class);
|
||||
final int port = context.getBean(Environment.class).getProperty("local.server.port", Integer.class);
|
||||
final TestController controller2 = context.getBean(TestController.class);
|
||||
final RSocketStrategies strategies = context.getBean(RSocketStrategies.class);
|
||||
|
||||
final Builder rsocketRequesterBuilder = RSocketRequester.builder().rsocketStrategies(strategies);
|
||||
|
||||
final RSocketRequester rSocketRequester = rsocketRequesterBuilder
|
||||
.websocket(URI.create("ws://localhost:" + port + "/rsocket"));
|
||||
|
||||
// REQUEST FNF
|
||||
whenRequestFnFIsSent(rSocketRequester, "api.c2.fnf").block();
|
||||
|
||||
FrameType receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.fnf", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST RESPONSE
|
||||
whenRequestResponseIsSent(rSocketRequester, "api.c2.rr").block();
|
||||
|
||||
receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.rr", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST STREAM
|
||||
whenRequestStreamIsSent(rSocketRequester, "api.c2.rs").blockLast();
|
||||
|
||||
receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.rs", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST CHANNEL
|
||||
whenRequestChannelIsSent(rSocketRequester, "api.c2.rc").blockLast();
|
||||
|
||||
receivedFrame = controller2.getReceivedFrames().take();
|
||||
thenSpanWasReportedWithTags(spans, "api.c2.rc", receivedFrame);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST FNF
|
||||
whenNonSampledRequestFnfIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST RESPONSE
|
||||
whenNonSampledRequestResponseIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST STREAM
|
||||
whenNonSampledRequestStreamIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST CHANNEL
|
||||
whenNonSampledRequestChannelIsSent(rSocketRequester);
|
||||
controller2.getReceivedFrames().take();
|
||||
// then
|
||||
thenNoSpanWasReported(spans, controller2, EXPECTED_TRACE_ID);
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// cleanup
|
||||
context.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void should_instrument_requester_and_responder() throws Exception {
|
||||
// setup
|
||||
ConfigurableApplicationContext context = new SpringApplicationBuilder(MyConfig.class, testConfiguration())
|
||||
.web(WebApplicationType.REACTIVE)
|
||||
.properties("server.port=0", "spring.rsocket.server.transport=websocket",
|
||||
"spring.rsocket.server.mapping-path=/rsocket", "spring.jmx.enabled=false",
|
||||
"spring.application.name=TraceRSocketTests", "security.basic.enabled=false",
|
||||
"management.security.enabled=false")
|
||||
.run();
|
||||
|
||||
final org.springframework.cloud.sleuth.Tracer tracer = context
|
||||
.getBean(org.springframework.cloud.sleuth.Tracer.class);
|
||||
final TestSpanHandler spans = context.getBean(TestSpanHandler.class);
|
||||
final int port = context.getBean(Environment.class).getProperty("local.server.port", Integer.class);
|
||||
final TestController controller2 = context.getBean(TestController.class);
|
||||
|
||||
final Builder rsocketRequesterBuilder = context.getBean(Builder.class);
|
||||
|
||||
final RSocketRequester rSocketRequester = rsocketRequesterBuilder
|
||||
.websocket(URI.create("ws://localhost:" + port + "/rsocket"));
|
||||
|
||||
// REQUEST FNF
|
||||
final org.springframework.cloud.sleuth.Span nextSpanFnf = tracer.nextSpan().start();
|
||||
whenRequestFnFIsSent(rSocketRequester, "api.c2.fnf")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanFnf.context()))
|
||||
.doFinally(signalType -> nextSpanFnf.end()).block();
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanFnf.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST RESPONSE
|
||||
final org.springframework.cloud.sleuth.Span nextSpanRR = tracer.nextSpan().start();
|
||||
whenRequestResponseIsSent(rSocketRequester, "api.c2.rr")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanRR.context()))
|
||||
.doFinally(signalType -> nextSpanRR.end()).block();
|
||||
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanRR.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST STREAM
|
||||
final org.springframework.cloud.sleuth.Span nextSpanRS = tracer.nextSpan().start();
|
||||
whenRequestStreamIsSent(rSocketRequester, "api.c2.rs")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanRS.context()))
|
||||
.doFinally(signalType -> nextSpanRS.end()).blockLast();
|
||||
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanRS.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// REQUEST CHANNEL
|
||||
final org.springframework.cloud.sleuth.Span nextSpanRC = tracer.nextSpan().start();
|
||||
whenRequestChannelIsSent(rSocketRequester, "api.c2.rc")
|
||||
.contextWrite(ctx -> ctx.put(TraceContext.class, nextSpanRC.context()))
|
||||
.doFinally(signalType -> nextSpanRC.end()).blockLast();
|
||||
|
||||
controller2.getReceivedFrames().take();
|
||||
thenNoSpanWasReported(spans, controller2, nextSpanRC.context().traceId());
|
||||
spans.clear();
|
||||
controller2.reset();
|
||||
|
||||
// cleanup
|
||||
context.close();
|
||||
}
|
||||
|
||||
protected abstract Class testConfiguration();
|
||||
|
||||
private void thenSpanWasReportedWithTags(TestSpanHandler spans, String path, FrameType frameType) {
|
||||
then(spans).hasSize(1);
|
||||
// TODO: Preferred option would be : [api.c2.{name}]
|
||||
FinishedSpan span = spans.get(0);
|
||||
then(span.getName()).isEqualTo(frameType.name() + " " + path);
|
||||
then(span.getTags()).containsEntry("messaging.controller.class", "org.springframework.cloud.sleuth.instrument.rsocket.TraceRSocketTests$TestController");
|
||||
then(span.getTags()).containsKey("messaging.controller.method");
|
||||
}
|
||||
|
||||
private Mono<Void> whenRequestFnFIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).send();
|
||||
}
|
||||
|
||||
private Mono<String> whenRequestResponseIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).retrieveMono(String.class);
|
||||
}
|
||||
|
||||
private Flux<String> whenRequestStreamIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).retrieveFlux(String.class);
|
||||
}
|
||||
|
||||
private Flux<String> whenRequestChannelIsSent(RSocketRequester requester, String path) {
|
||||
return requester.route(path).data(Flux.fromArray(new String[] { "test1", "test2" })).retrieveFlux(String.class);
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestFnfIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.fnf").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).send().block();
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestResponseIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.rr").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).retrieveMono(String.class).block();
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestStreamIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.rs").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).retrieveFlux(String.class).blockLast();
|
||||
}
|
||||
|
||||
private void whenNonSampledRequestChannelIsSent(RSocketRequester requester) {
|
||||
requester.route("api.c2.rc").metadata(EXPECTED_TRACE_ID + "-" + EXPECTED_TRACE_ID + "-0", new MimeType("b3") {
|
||||
@Override
|
||||
public String toString() {
|
||||
return "b3";
|
||||
}
|
||||
}).data(Flux.fromArray(new String[] { "test1", "test2" })).retrieveFlux(String.class).blockLast();
|
||||
}
|
||||
|
||||
private void thenNoSpanWasReported(TestSpanHandler spans, TestController controller2, String expectedTraceId) {
|
||||
// then(spans).isEmpty(); // FIXME: does not work for request case
|
||||
then(controller2.getSpan()).isNotNull();
|
||||
then(controller2.getSpan().context().traceId()).isEqualTo(expectedTraceId);
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableAutoConfiguration
|
||||
static class MyConfig {
|
||||
|
||||
@Bean
|
||||
TestController controller(Tracer tracer) {
|
||||
return new TestController(tracer);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Controller
|
||||
@MessageMapping("api.c2")
|
||||
static class TestController {
|
||||
|
||||
final Tracer tracer;
|
||||
|
||||
Span span;
|
||||
|
||||
ContextView interceptedContext;
|
||||
|
||||
BlockingQueue<FrameType> receivedFrames = new LinkedBlockingDeque<>();
|
||||
|
||||
TestController(Tracer tracer) {
|
||||
this.tracer = tracer;
|
||||
}
|
||||
|
||||
BlockingQueue<FrameType> getReceivedFrames() {
|
||||
return this.receivedFrames;
|
||||
}
|
||||
|
||||
Span getSpan() {
|
||||
return this.span;
|
||||
}
|
||||
|
||||
void reset() {
|
||||
this.span = null;
|
||||
}
|
||||
|
||||
@MessageMapping("fnf")
|
||||
Mono<Void> testFnf() {
|
||||
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Mono.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_FNF);
|
||||
return Mono.empty();
|
||||
});
|
||||
}
|
||||
|
||||
@MessageMapping("rr")
|
||||
Mono<String> testRR() {
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Mono.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_RESPONSE);
|
||||
return Mono.just("response");
|
||||
});
|
||||
}
|
||||
|
||||
@MessageMapping("rs")
|
||||
Flux<String> testRS() {
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Flux.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_STREAM);
|
||||
return Flux.just("stream");
|
||||
});
|
||||
}
|
||||
|
||||
@MessageMapping("rc")
|
||||
Flux<String> testRC(@Payload Flux<String> inbound) {
|
||||
this.span = this.tracer.currentSpan();
|
||||
|
||||
return Flux.deferContextual(c -> {
|
||||
interceptedContext = c;
|
||||
receivedFrames.offer(FrameType.REQUEST_CHANNEL);
|
||||
return inbound;
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user