From a6bf9a328a951fef9b750a430fba5e9d876dfb7f Mon Sep 17 00:00:00 2001 From: Marcin Grzejszczak Date: Thu, 29 Apr 2021 14:08:49 +0200 Subject: [PATCH] Fixed the messaging aspect --- .../instrument/deployer/TraceAppDeployer.java | 516 ++++++------- .../messaging/TraceMessagingAspect.java | 2 +- .../instrument/rsocket/TraceRSocketTests.java | 730 +++++++++--------- 3 files changed, 626 insertions(+), 622 deletions(-) diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/deployer/TraceAppDeployer.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/deployer/TraceAppDeployer.java index b236a6057..b4f2febe5 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/deployer/TraceAppDeployer.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/deployer/TraceAppDeployer.java @@ -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 statusReactive(String id) { - return ReactorSleuth.tracedMono(tracer(), currentTraceContext(), "status", - () -> this.delegate.statusReactive(id), span -> span.tag("deployer.app.id", id)); - } - - @Override - public Flux 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 statusReactive(String id) { + return ReactorSleuth.tracedMono(tracer(), currentTraceContext(), "status", + () -> this.delegate.statusReactive(id), span -> span.tag("deployer.app.id", id)); + } + + @Override + public Flux 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; + } + + } + +} diff --git a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAspect.java b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAspect.java index fda6d71bd..a2b262bc0 100644 --- a/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAspect.java +++ b/spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAspect.java @@ -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 diff --git a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TraceRSocketTests.java b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TraceRSocketTests.java index 556b4ff54..43a1f9f97 100644 --- a/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TraceRSocketTests.java +++ b/tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TraceRSocketTests.java @@ -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 whenRequestFnFIsSent(RSocketRequester requester, String path) { - return requester.route(path).send(); - } - - private Mono whenRequestResponseIsSent(RSocketRequester requester, String path) { - return requester.route(path).retrieveMono(String.class); - } - - private Flux whenRequestStreamIsSent(RSocketRequester requester, String path) { - return requester.route(path).retrieveFlux(String.class); - } - - private Flux 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 receivedFrames = new LinkedBlockingDeque<>(); - - TestController(Tracer tracer) { - this.tracer = tracer; - } - - BlockingQueue getReceivedFrames() { - return this.receivedFrames; - } - - Span getSpan() { - return this.span; - } - - void reset() { - this.span = null; - } - - @MessageMapping("fnf") - Mono testFnf() { - - this.span = this.tracer.currentSpan(); - - return Mono.deferContextual(c -> { - interceptedContext = c; - receivedFrames.offer(FrameType.REQUEST_FNF); - return Mono.empty(); - }); - } - - @MessageMapping("rr") - Mono testRR() { - this.span = this.tracer.currentSpan(); - - return Mono.deferContextual(c -> { - interceptedContext = c; - receivedFrames.offer(FrameType.REQUEST_RESPONSE); - return Mono.just("response"); - }); - } - - @MessageMapping("rs") - Flux testRS() { - this.span = this.tracer.currentSpan(); - - return Flux.deferContextual(c -> { - interceptedContext = c; - receivedFrames.offer(FrameType.REQUEST_STREAM); - return Flux.just("stream"); - }); - } - - @MessageMapping("rc") - Flux testRC(@Payload Flux 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 whenRequestFnFIsSent(RSocketRequester requester, String path) { + return requester.route(path).send(); + } + + private Mono whenRequestResponseIsSent(RSocketRequester requester, String path) { + return requester.route(path).retrieveMono(String.class); + } + + private Flux whenRequestStreamIsSent(RSocketRequester requester, String path) { + return requester.route(path).retrieveFlux(String.class); + } + + private Flux 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 receivedFrames = new LinkedBlockingDeque<>(); + + TestController(Tracer tracer) { + this.tracer = tracer; + } + + BlockingQueue getReceivedFrames() { + return this.receivedFrames; + } + + Span getSpan() { + return this.span; + } + + void reset() { + this.span = null; + } + + @MessageMapping("fnf") + Mono testFnf() { + + this.span = this.tracer.currentSpan(); + + return Mono.deferContextual(c -> { + interceptedContext = c; + receivedFrames.offer(FrameType.REQUEST_FNF); + return Mono.empty(); + }); + } + + @MessageMapping("rr") + Mono testRR() { + this.span = this.tracer.currentSpan(); + + return Mono.deferContextual(c -> { + interceptedContext = c; + receivedFrames.offer(FrameType.REQUEST_RESPONSE); + return Mono.just("response"); + }); + } + + @MessageMapping("rs") + Flux testRS() { + this.span = this.tracer.currentSpan(); + + return Flux.deferContextual(c -> { + interceptedContext = c; + receivedFrames.offer(FrameType.REQUEST_STREAM); + return Flux.just("stream"); + }); + } + + @MessageMapping("rc") + Flux testRC(@Payload Flux inbound) { + this.span = this.tracer.currentSpan(); + + return Flux.deferContextual(c -> { + interceptedContext = c; + receivedFrames.offer(FrameType.REQUEST_CHANNEL); + return inbound; + }); + } + + } + +}