SpringCloud/spring-cloud-sleuth
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adds RSocket instrumentation

Browse Source 
coauthor @olegdokuka

fixes gh-1677
This commit is contained in:
Marcin Grzejszczak
2021-04-29 13:32:58 +02:00
parent ba8da46e20
commit 216d681b04
42 changed files with 2494 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
docs/src/main/asciidoc/integrations.adoc
View File

@@ -583,3 +583,12 @@ This feature is available for all tracer implementations.
If you have Spring Cloud Deployer running on the classpath, we wrap the `AppDeployer` in a trace representation. We are polling the application for its status at a default interval. You can change that default by setting the `spring.sleuth.deployer.status-poll-delay` property.
In order to disable this instrumentation set `spring.sleuth.deployer.enabled` to `false`.
[[sleuth-deployer-integration]]
== Spring RSocket
This feature is available for all tracer implementations.
If you have Spring RSocket running on the classpath, we wrap the inbound and outbound communication to propagate the tracing context via the metadata.
In order to disable this instrumentation set `spring.sleuth.rsocket.enabled` to `false`.

5
spring-cloud-sleuth-api/src/main/java/org/springframework/cloud/sleuth/SpanAndScope.java
View File

@@ -41,4 +41,9 @@ public class SpanAndScope {
return this.scope;
}
@Override
public String toString() {
return "SpanAndScope{" + "span=" + this.span + '}';
}
}

43
spring-cloud-sleuth-api/src/main/java/org/springframework/cloud/sleuth/TraceContext.java
View File

@@ -51,4 +51,47 @@ public interface TraceContext {
*/
Boolean sampled();
/**
* Builder for {@link TraceContext}.
*
* @since 3.1.0
*/
interface Builder {
/**
* Sets trace id on the trace context.
* @param traceId trace id
* @return this
*/
TraceContext.Builder traceId(String traceId);
/**
* Sets parent id on the trace context.
* @param parentId parent trace id
* @return this
*/
TraceContext.Builder parentId(String parentId);
/**
* Sets span id on the trace context.
* @param spanId span id
* @return this
*/
TraceContext.Builder spanId(String spanId);
/**
* Sets sampled on the trace context.
* @param sampled if span is sampled
* @return this
*/
TraceContext.Builder sampled(Boolean sampled);
/**
* Builds the trace context.
* @return trace context
*/
TraceContext build();
}
}

6
spring-cloud-sleuth-api/src/main/java/org/springframework/cloud/sleuth/Tracer.java
View File

@@ -134,6 +134,12 @@ public interface Tracer extends BaggageManager {
*/
Span.Builder spanBuilder();
/**
* Builder for {@link TraceContext}.
* @return a trace context builder
*/
TraceContext.Builder traceContextBuilder();
/**
* Allows to customize the current span in scope.
* @return current span customizer

91
spring-cloud-sleuth-api/src/main/java/org/springframework/cloud/sleuth/WithThreadLocalSpan.java Normal file
View File

@@ -0,0 +1,91 @@
/*
* 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;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.lang.Nullable;
/**
* Represents a {@link Span} stored in thread local.
*
* @author Marcin Grzejszczak
* @since 3.1.0
*/
public interface WithThreadLocalSpan {
/**
* Logger.
*/
Log log = LogFactory.getLog(WithThreadLocalSpan.class);
/**
* Sets the span in thread local scope.
* @param span span to put in thread local
*/
default void setSpanInScope(Span span) {
getThreadLocalSpan().set(span);
if (log.isDebugEnabled()) {
log.debug("Put span in scope " + span);
}
}
/**
* Finishes the thread local span.
* @param error potential error to be stored in span
*/
default void finishSpan(@Nullable Throwable error) {
SpanAndScope spanAndScope = takeSpanFromThreadLocal();
if (spanAndScope == null) {
return;
}
Span span = spanAndScope.getSpan();
Tracer.SpanInScope scope = spanAndScope.getScope();
if (span.isNoop()) {
if (log.isDebugEnabled()) {
log.debug("Span " + span + " is noop - will stope the scope");
}
scope.close();
return;
}
if (error != null) { // an error occurred, adding error to span
span.error(error);
}
if (log.isDebugEnabled()) {
log.debug("Will finish the span and its corresponding scope " + span);
}
span.end();
scope.close();
}
/**
* Takes a span from thread local and restores the previous one if present.
* @return span from a thread local span
*/
default SpanAndScope takeSpanFromThreadLocal() {
SpanAndScope span = getThreadLocalSpan().get();
if (log.isDebugEnabled()) {
log.debug("Took span [" + span + "] from thread local");
}
getThreadLocalSpan().remove();
return span;
}
ThreadLocalSpan getThreadLocalSpan();
}

5
spring-cloud-sleuth-autoconfigure/pom.xml
View File

@@ -324,6 +324,11 @@
<artifactId>spring-boot-starter-data-mongodb</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-rsocket</artifactId>
<optional>true</optional>
</dependency>
<!-- Zipkin -->
<dependency>

34
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/SleuthMessagingProperties.java
View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.sleuth.autoconfig.instrument.messaging;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.messaging.handler.annotation.MessageMapping;
/**
* Properties for messaging.
@@ -32,6 +33,11 @@ public class SleuthMessagingProperties {
*/
private boolean enabled;
/**
* Aspect related properties.
*/
private Aspect aspect = new Aspect();
/**
* Rabbit related properties.
*/
@@ -55,6 +61,14 @@ public class SleuthMessagingProperties {
this.enabled = enabled;
}
public Aspect getAspect() {
return this.aspect;
}
public void setAspect(Aspect aspect) {
this.aspect = aspect;
}
public Rabbit getRabbit() {
return this.rabbit;
}
@@ -79,6 +93,26 @@ public class SleuthMessagingProperties {
this.jms = jms;
}
/**
* Aspect configuration.
*/
public static class Aspect {
/**
* Should {@link MessageMapping} wrapping be enabled.
*/
private boolean enabled;
public boolean isEnabled() {
return this.enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
}
/**
* RabbitMQ configuration.
*/

11
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceSpringMessagingAutoConfiguration.java
View File

@@ -16,12 +16,16 @@
package org.springframework.cloud.sleuth.autoconfig.instrument.messaging;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.instrument.messaging.MessageHeaderPropagatorGetter;
import org.springframework.cloud.sleuth.instrument.messaging.MessageHeaderPropagatorSetter;
import org.springframework.cloud.sleuth.instrument.messaging.TraceMessagingAspect;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -29,10 +33,17 @@ import org.springframework.messaging.support.MessageHeaderAccessor;
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(MessageHeaderAccessor.class)
@ConditionalOnBean(Tracer.class)
@ConditionalOnProperty(value = "spring.sleuth.messaging.enabled", matchIfMissing = true)
@EnableConfigurationProperties({ SleuthIntegrationMessagingProperties.class, SleuthMessagingProperties.class })
class TraceSpringMessagingAutoConfiguration {
@Bean
@ConditionalOnProperty(value = "spring.sleuth.messaging.aspect.enabled", matchIfMissing = true)
TraceMessagingAspect traceMessagingAspect(Tracer tracer, SpanNamer spanNamer) {
return new TraceMessagingAspect(tracer, spanNamer);
}
@Bean
@ConditionalOnMissingBean
Propagator.Setter<MessageHeaderAccessor> traceMessagePropagationSetter() {

43
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/rsocket/SleuthRSocketProperties.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* 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.autoconfig.instrument.rsocket;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* Sleuth RSocket settings.
*
* @author Oleh Dokuka
* @since 3.1.0
*/
@ConfigurationProperties("spring.sleuth.rsocket")
public class SleuthRSocketProperties {
/**
* When true enables instrumentation for rsocket.
*/
private boolean enabled = true;
public boolean isEnabled() {
return this.enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
}

89
spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/rsocket/TraceRSocketAutoConfiguration.java Normal file
View File

@@ -0,0 +1,89 @@
/*
* 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.autoconfig.instrument.rsocket;
import java.util.List;
import io.rsocket.RSocket;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.rsocket.RSocketRequesterAutoConfiguration;
import org.springframework.boot.autoconfigure.rsocket.RSocketServerAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration;
import org.springframework.cloud.sleuth.brave.propagation.PropagationType;
import org.springframework.cloud.sleuth.instrument.rsocket.TracingRSocketConnectorConfigurer;
import org.springframework.cloud.sleuth.instrument.rsocket.TracingRSocketServerCustomizer;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.messaging.rsocket.RSocketConnectorConfigurer;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketRequester.Builder;
import org.springframework.messaging.rsocket.RSocketStrategies;
@Configuration(proxyBeanMethods = false)
@ConditionalOnBean(Tracer.class)
@ConditionalOnProperty(value = "spring.sleuth.rsocket.enabled", matchIfMissing = true)
@ConditionalOnClass({ RSocket.class, RSocketStrategies.class })
@AutoConfigureAfter(BraveAutoConfiguration.class)
@AutoConfigureBefore({ RSocketRequesterAutoConfiguration.class, RSocketServerAutoConfiguration.class })
@EnableConfigurationProperties(SleuthRSocketProperties.class)
public class TraceRSocketAutoConfiguration {
// We're using text instead of objects cause we can have same properties from Brave /
// OTel
@Bean
@Scope("prototype")
@ConditionalOnMissingBean
Builder rSocketRequesterBuilder(RSocketStrategies strategies,
ObjectProvider<RSocketConnectorConfigurer> connectorConfigurerProvider) {
// TODO: should be in spring boot
final Builder builder = RSocketRequester.builder().rsocketStrategies(strategies);
connectorConfigurerProvider.forEach(builder::rsocketConnector);
return builder;
}
private boolean containsZipkinPropagationType(List<PropagationType> types) {
return types.contains(PropagationType.B3);
}
@Bean
RSocketConnectorConfigurer tracingRSocketConnectorConfigurer(Propagator propagator, Tracer tracer,
@Value("${spring.sleuth.propagation.type:B3}") List<PropagationType> types) {
return new TracingRSocketConnectorConfigurer(propagator, tracer, containsZipkinPropagationType(types));
}
// We're using text instead of objects cause we can have same properties from Brave /
// OTel
@Bean
RSocketServerCustomizer tracingRSocketServerCustomizer(Propagator propagator, Tracer tracer,
@Value("${spring.sleuth.propagation.type:B3}") List<PropagationType> types) {
return new TracingRSocketServerCustomizer(propagator, tracer, containsZipkinPropagationType(types));
}
}

1
spring-cloud-sleuth-autoconfigure/src/main/resources/META-INF/spring.factories
View File

@@ -19,6 +19,7 @@ org.springframework.cloud.sleuth.autoconfig.instrument.messaging.TraceFunctionAu
org.springframework.cloud.sleuth.autoconfig.instrument.messaging.TraceSpringIntegrationAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.instrument.messaging.TraceSpringMessagingAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.instrument.messaging.TraceWebSocketAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.instrument.rsocket.TraceRSocketAutoConfiguration, \
org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.brave.instrument.web.client.BraveWebClientAutoConfiguration,\
org.springframework.cloud.sleuth.autoconfig.brave.instrument.rpc.BraveRpcAutoConfiguration,\

54
spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/NoOpTraceContextBuilder.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* 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.autoconfig;
import org.springframework.cloud.sleuth.TraceContext;
/**
* A noop implementation. Does nothing.
*
* @author Marcin Grzejszczak
* @since 3.1.0
*/
class NoOpTraceContextBuilder implements TraceContext.Builder {
@Override
public TraceContext.Builder traceId(String traceId) {
return this;
}
@Override
public TraceContext.Builder parentId(String traceId) {
return this;
}
@Override
public TraceContext.Builder spanId(String spanId) {
return null;
}
@Override
public TraceContext.Builder sampled(Boolean sampled) {
return this;
}
@Override
public TraceContext build() {
return new NoOpTraceContext();
}
}

5
spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/NoOpTracer.java
View File

@@ -69,6 +69,11 @@ class NoOpTracer implements Tracer {
return new NoOpSpanBuilder();
}
@Override
public TraceContext.Builder traceContextBuilder() {
return new NoOpTraceContextBuilder();
}
@Override
public Map<String, String> getAllBaggage() {
return new HashMap<>();

71
spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/bridge/BraveTraceContextBuilder.java Normal file
View File

@@ -0,0 +1,71 @@
/*
* 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.brave.bridge;
import org.springframework.cloud.sleuth.TraceContext;
import org.springframework.cloud.sleuth.internal.EncodingUtils;
/**
* Brave implementation of a {@link TraceContext.Builder}.
*
* @author Marcin Grzejszczak
* @since 3.1.0
*/
class BraveTraceContextBuilder implements TraceContext.Builder {
brave.propagation.TraceContext.Builder delegate = brave.propagation.TraceContext.newBuilder();
@Override
public TraceContext.Builder traceId(String traceId) {
long[] fromString = EncodingUtils.fromString(traceId);
if (fromString.length == 2) {
this.delegate.traceIdHigh(fromString[0]);
this.delegate.traceId(fromString[1]);
}
else {
this.delegate.traceId(fromString[0]);
}
return this;
}
@Override
public TraceContext.Builder parentId(String traceId) {
long[] fromString = EncodingUtils.fromString(traceId);
this.delegate.parentId(fromString[fromString.length == 2 ? 1 : 0]);
return this;
}
@Override
public TraceContext.Builder spanId(String spanId) {
long[] fromString = EncodingUtils.fromString(spanId);
this.delegate.spanId(fromString[fromString.length == 2 ? 1 : 0]);
return this;
}
@Override
public TraceContext.Builder sampled(Boolean sampled) {
this.delegate.sampled(sampled);
return this;
}
@Override
public TraceContext build() {
brave.propagation.TraceContext context = this.delegate.build();
return BraveTraceContext.fromBrave(context);
}
}

5
spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/bridge/BraveTracer.java
View File

@@ -90,6 +90,11 @@ public class BraveTracer implements Tracer {
return new BraveSpanBuilder(this.tracer);
}
@Override
public TraceContext.Builder traceContextBuilder() {
return new BraveTraceContextBuilder();
}
@Override
public Map<String, String> getAllBaggage() {
return this.braveBaggageManager.getAllBaggage();

174
spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/bridge/W3CPropagation.java
View File

@@ -37,6 +37,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.sleuth.BaggageInScope;
import org.springframework.cloud.sleuth.internal.EncodingUtils;
import static java.util.Collections.singletonList;
@@ -219,12 +220,12 @@ class W3CPropagation extends Propagation.Factory implements Propagation<String>
private static boolean isTraceIdValid(CharSequence traceId) {
return (traceId.length() == TRACE_ID_HEX_SIZE) && !INVALID_TRACE_ID.contentEquals(traceId)
&& BigendianEncoding.isValidBase16String(traceId);
&& EncodingUtils.isValidBase16String(traceId);
}
private static boolean isSpanIdValid(String spanId) {
return (spanId.length() == SPAN_ID_HEX_SIZE) && !INVALID_SPAN_ID.equals(spanId)
&& BigendianEncoding.isValidBase16String(spanId);
&& EncodingUtils.isValidBase16String(spanId);
}
private static TraceContext extractContextFromTraceParent(String traceparent) {
@@ -257,10 +258,10 @@ class W3CPropagation extends Propagation.Factory implements Propagation<String>
String traceIdLow = traceId.substring(traceId.length() / 2);
byte isSampled = TraceFlags.byteFromHex(traceparent, TRACE_OPTION_OFFSET);
return TraceContext.newBuilder().shared(true)
.traceIdHigh(BigendianEncoding.longFromBase16String(traceIdHigh))
.traceId(BigendianEncoding.longFromBase16String(traceIdLow))
.spanId(BigendianEncoding.longFromBase16String(spanId))
.sampled(isSampled == TraceFlags.IS_SAMPLED).build();
.traceIdHigh(EncodingUtils.longFromBase16String(traceIdHigh))
.traceId(EncodingUtils.longFromBase16String(traceIdLow))
.spanId(EncodingUtils.longFromBase16String(spanId)).sampled(isSampled == TraceFlags.IS_SAMPLED)
.build();
}
return null;
}
@@ -421,41 +422,6 @@ final class TemporaryBuffers {
}
/**
* Taken from OpenTelemetry API.
*/
final class Utils {
private Utils() {
}
/**
* Throws an {@link IllegalArgumentException} if the argument is false. This method is
* similar to {@code Preconditions.checkArgument(boolean, Object)} from Guava.
* @param isValid whether the argument check passed.
* @param errorMessage the message to use for the exception.
*/
static void checkArgument(boolean isValid, String errorMessage) {
if (!isValid) {
throw new IllegalArgumentException(errorMessage);
}
}
/**
* Throws an {@link IllegalStateException} if the argument is false. This method is
* similar to {@code Preconditions.checkState(boolean, Object)} from Guava.
* @param isValid whether the state check passed.
* @param errorMessage the message to use for the exception.
*/
static void checkState(boolean isValid, String errorMessage) {
if (!isValid) {
throw new IllegalStateException(String.valueOf(errorMessage));
}
}
}
/**
* Taken from OpenTelemetry API.
*/
@@ -469,131 +435,7 @@ final class TraceFlags {
/** Extract the byte representation of the flags from a hex-representation. */
static byte byteFromHex(CharSequence src, int srcOffset) {
return BigendianEncoding.byteFromBase16String(src, srcOffset);
}
}
/**
* Taken from OpenTelemetry API.
*/
final class BigendianEncoding {
private BigendianEncoding() {
}
static final int LONG_BYTES = Long.SIZE / Byte.SIZE;
static final int BYTE_BASE16 = 2;
static final int LONG_BASE16 = BYTE_BASE16 * LONG_BYTES;
private static final String ALPHABET = "0123456789abcdef";
private static final int ASCII_CHARACTERS = 128;
private static final char[] ENCODING = buildEncodingArray();
private static final byte[] DECODING = buildDecodingArray();
private static char[] buildEncodingArray() {
char[] encoding = new char[512];
for (int i = 0; i < 256; ++i) {
encoding[i] = ALPHABET.charAt(i >>> 4);
encoding[i | 0x100] = ALPHABET.charAt(i & 0xF);
}
return encoding;
}
private static byte[] buildDecodingArray() {
byte[] decoding = new byte[ASCII_CHARACTERS];
Arrays.fill(decoding, (byte) -1);
for (int i = 0; i < ALPHABET.length(); i++) {
char c = ALPHABET.charAt(i);
decoding[c] = (byte) i;
}
return decoding;
}
/**
* Returns the {@code long} value whose base16 representation is stored in the first
* 16 chars of {@code chars} starting from the {@code offset}.
* @param chars the base16 representation of the {@code long}.
*/
static long longFromBase16String(CharSequence chars) {
return longFromBase16String(chars, 0);
}
/**
* Returns the {@code long} value whose base16 representation is stored in the first
* 16 chars of {@code chars} starting from the {@code offset}.
* @param chars the base16 representation of the {@code long}.
*/
static long longFromBase16String(CharSequence chars, int offset) {
Utils.checkArgument(chars.length() >= offset + LONG_BASE16, "chars too small");
return (decodeByte(chars.charAt(offset), chars.charAt(offset + 1)) & 0xFFL) << 56
| (decodeByte(chars.charAt(offset + 2), chars.charAt(offset + 3)) & 0xFFL) << 48
| (decodeByte(chars.charAt(offset + 4), chars.charAt(offset + 5)) & 0xFFL) << 40
| (decodeByte(chars.charAt(offset + 6), chars.charAt(offset + 7)) & 0xFFL) << 32
| (decodeByte(chars.charAt(offset + 8), chars.charAt(offset + 9)) & 0xFFL) << 24
| (decodeByte(chars.charAt(offset + 10), chars.charAt(offset + 11)) & 0xFFL) << 16
| (decodeByte(chars.charAt(offset + 12), chars.charAt(offset + 13)) & 0xFFL) << 8
| (decodeByte(chars.charAt(offset + 14), chars.charAt(offset + 15)) & 0xFFL);
}
/**
* Decodes the specified two character sequence, and returns the resulting
* {@code byte}.
* @param chars the character sequence to be decoded.
* @param offset the starting offset in the {@code CharSequence}.
* @return the resulting {@code byte}
* @throws IllegalArgumentException if the input is not a valid encoded string
* according to this encoding.
*/
static byte byteFromBase16String(CharSequence chars, int offset) {
Utils.checkArgument(chars.length() >= offset + 2, "chars too small");
return decodeByte(chars.charAt(offset), chars.charAt(offset + 1));
}
private static byte decodeByte(char hi, char lo) {
Utils.checkArgument(lo < ASCII_CHARACTERS && DECODING[lo] != -1, "invalid character " + lo);
Utils.checkArgument(hi < ASCII_CHARACTERS && DECODING[hi] != -1, "invalid character " + hi);
int decoded = DECODING[hi] << 4 | DECODING[lo];
return (byte) decoded;
}
/**
* Returns the {@code long} value whose big-endian representation is stored in the
* first 8 bytes of {@code bytes} starting from the {@code offset}.
* @param bytes the byte array representation of the {@code long}.
* @param offset the starting offset in the byte array.
* @return the {@code long} value whose big-endian representation is given.
* @throws IllegalArgumentException if {@code bytes} has fewer than 8 elements.
*/
static long longFromByteArray(byte[] bytes, int offset) {
Utils.checkArgument(bytes.length >= offset + LONG_BYTES, "array too small");
return (bytes[offset] & 0xFFL) << 56 | (bytes[offset + 1] & 0xFFL) << 48 | (bytes[offset + 2] & 0xFFL) << 40
| (bytes[offset + 3] & 0xFFL) << 32 | (bytes[offset + 4] & 0xFFL) << 24
| (bytes[offset + 5] & 0xFFL) << 16 | (bytes[offset + 6] & 0xFFL) << 8 | (bytes[offset + 7] & 0xFFL);
}
static boolean isValidBase16String(CharSequence value) {
for (int i = 0; i < value.length(); i++) {
char b = value.charAt(i);
// 48..57 && 97..102 are valid
if (!isDigit(b) && !isLowercaseHexCharacter(b)) {
return false;
}
}
return true;
}
private static boolean isLowercaseHexCharacter(char b) {
return 97 <= b && b <= 102;
}
private static boolean isDigit(char b) {
return 48 <= b && b <= 57;
return EncodingUtils.byteFromBase16String(src, srcOffset);
}
}

54
spring-cloud-sleuth-brave/src/test/java/org/springframework/cloud/sleuth/brave/bridge/BraveTraceContextBuilderTests.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* 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.brave.bridge;
import org.junit.jupiter.api.Test;
import org.springframework.cloud.sleuth.TraceContext;
import static org.assertj.core.api.BDDAssertions.then;
class BraveTraceContextBuilderTests {
@Test
void should_set_trace_context_for_64_bit() {
BraveTraceContextBuilder builder = new BraveTraceContextBuilder();
TraceContext traceContext = builder.parentId("7c6239a5ad0a4287").spanId("caff89f7f0f229dd")
.traceId("596e1787feb11040").sampled(true).build();
then(traceContext.parentId()).isEqualTo("7c6239a5ad0a4287");
then(traceContext.spanId()).isEqualTo("caff89f7f0f229dd");
then(traceContext.traceId()).isEqualTo("596e1787feb11040");
then(traceContext.sampled()).isTrue();
}
@Test
void should_set_trace_context_for_128_bit() {
BraveTraceContextBuilder builder = new BraveTraceContextBuilder();
TraceContext traceContext = builder.parentId("00000000000000007c6239a5ad0a4287")
.spanId("0000000000000000caff89f7f0f229dd").traceId("596e1787feb11040caff89f7f0f229dd").sampled(true)
.build();
then(traceContext.parentId()).isEqualTo("7c6239a5ad0a4287");
then(traceContext.spanId()).isEqualTo("caff89f7f0f229dd");
then(traceContext.traceId()).isEqualTo("596e1787feb11040caff89f7f0f229dd");
then(traceContext.sampled()).isTrue();
}
}

8
spring-cloud-sleuth-brave/src/test/java/org/springframework/cloud/sleuth/brave/bridge/W3CPropagationTest.java
View File

@@ -27,6 +27,8 @@ import brave.propagation.TraceContext;
import brave.propagation.TraceContextOrSamplingFlags;
import org.junit.jupiter.api.Test;
import org.springframework.cloud.sleuth.internal.EncodingUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
import static org.springframework.cloud.sleuth.brave.bridge.W3CPropagation.TRACE_PARENT;
@@ -68,9 +70,9 @@ class W3CPropagationTest {
private TraceContext.Builder sampledTraceContext(String traceIdHigh, String traceId, String spanId) {
return TraceContext.newBuilder().sampled(SAMPLED_TRACE_OPTIONS)
.traceIdHigh(BigendianEncoding.longFromBase16String(traceIdHigh))
.traceId(BigendianEncoding.longFromBase16String(traceId))
.spanId(BigendianEncoding.longFromBase16String(spanId));
.traceIdHigh(EncodingUtils.longFromBase16String(traceIdHigh))
.traceId(EncodingUtils.longFromBase16String(traceId))
.spanId(EncodingUtils.longFromBase16String(spanId));
}
@Test

10
spring-cloud-sleuth-instrumentation/pom.xml
View File

@@ -42,6 +42,11 @@
<artifactId>spring-boot-starter-web</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-rsocket</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-core</artifactId>
@@ -52,6 +57,11 @@
<artifactId>reactor-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>

1
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/async/TraceAsyncAspect.java
View File

@@ -62,6 +62,7 @@ public class TraceAsyncAspect {
}
span = span.name(spanName);
try (Tracer.SpanInScope ws = this.tracer.withSpan(span.start())) {
// TODO: Make this less generic
span.tag(CLASS_KEY, pjp.getTarget().getClass().getSimpleName());
span.tag(METHOD_KEY, pjp.getSignature().getName());
return pjp.proceed();

6
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessageHandler.java
View File

@@ -191,12 +191,6 @@ class TraceMessageHandler {
}
}
private void addTags(Span result, String destinationName) {
if (StringUtils.hasText(destinationName)) {
result.tag("channel", SpanNameUtil.shorten(destinationName));
}
}
/**
* Called either when message got received and processed or message got sent.
* @param span - span that corresponds to the given operation

100
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessagingAspect.java Normal file
View File

@@ -0,0 +1,100 @@
/*
* 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.messaging;
import java.lang.reflect.Method;
import org.apache.commons.logging.Log;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.SpanNamer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.SpanNameUtil;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.util.ReflectionUtils;
/**
* Aspect that wraps {@link MessageMapping} annotated methods in a tracing representation.
*
* TODO: Document that for client side responders declare them as beans
*
* @author Marcin Grzejszczak
* @since 3.1.0
*/
@SuppressWarnings("ArgNamesWarningsInspection")
@Aspect
public class TraceMessagingAspect {
private static final Log log = org.apache.commons.logging.LogFactory.getLog(TraceMessagingAspect.class);
static final String MESSAGING_CONTROLLER_CLASS_KEY = "messaging.controller.class";
static final String MESSAGING_CONTROLLER_METHOD_KEY = "messaging.controller.method";
private final Tracer tracer;
private final SpanNamer spanNamer;
public TraceMessagingAspect(Tracer tracer, SpanNamer spanNamer) {
this.tracer = tracer;
this.spanNamer = spanNamer;
}
@Pointcut("@within(org.springframework.messaging.handler.annotation.MessageMapping)")
private void anyMessageMappingAnnotated() {
} // NOSONAR
@Around("anyMessageMappingAnnotated()")
@SuppressWarnings("unchecked")
public Object addTags(ProceedingJoinPoint pjp) throws Throwable {
Object object = pjp.proceed();
String methodName = pjp.getSignature().getName();
String className = pjp.getTarget().getClass().getName();
Span currentSpan = currentSpan(pjp);
currentSpan.tag(MESSAGING_CONTROLLER_CLASS_KEY, className);
currentSpan.tag(MESSAGING_CONTROLLER_METHOD_KEY, methodName);
return object;
}
private Span currentSpan(ProceedingJoinPoint pjp) {
Span currentSpan = this.tracer.currentSpan();
if (currentSpan == null) {
if (log.isDebugEnabled()) {
log.debug("No span found - will create a new one");
}
currentSpan = this.tracer.nextSpan().name(name(pjp)).start();
}
return currentSpan;
}
private String name(ProceedingJoinPoint pjp) {
return this.spanNamer.name(getMethod(pjp, pjp.getTarget()),
SpanNameUtil.toLowerHyphen(pjp.getSignature().getName()));
}
private Method getMethod(ProceedingJoinPoint pjp, Object object) {
MethodSignature signature = (MethodSignature) pjp.getSignature();
Method method = signature.getMethod();
return ReflectionUtils.findMethod(object.getClass(), method.getName(), method.getParameterTypes());
}
}

49
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java
View File

@@ -26,9 +26,9 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.SpanAndScope;
import org.springframework.cloud.sleuth.ThreadLocalSpan;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.WithThreadLocalSpan;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.cloud.stream.binder.BinderType;
import org.springframework.cloud.stream.binder.BinderTypeRegistry;
@@ -58,7 +58,7 @@ import org.springframework.util.StringUtils;
* @since 3.0.0
*/
public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
implements ExecutorChannelInterceptor, ApplicationContextAware {
implements ExecutorChannelInterceptor, ApplicationContextAware, WithThreadLocalSpan {
/**
* Name of the class in Spring Cloud Stream that is a direct channel.
@@ -163,13 +163,6 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
return outputMessage;
}
private void setSpanInScope(Span span) {
this.threadLocalSpan.set(span);
if (log.isDebugEnabled()) {
log.debug("Put span in scope " + span);
}
}
private String toRemoteServiceName(MessageHeaderAccessor headers) {
for (String key : headers.getMessageHeaders().keySet()) {
String remoteServiceName = this.remoteServiceNameMapper.apply(key);
@@ -357,41 +350,9 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
finishSpan(ex);
}
void finishSpan(Exception error) {
SpanAndScope spanAndScope = getSpanFromThreadLocal();
if (spanAndScope == null) {
return;
}
Span span = spanAndScope.getSpan();
Tracer.SpanInScope scope = spanAndScope.getScope();
if (span.isNoop()) {
if (log.isDebugEnabled()) {
log.debug("Span " + span + " is noop - will stope the scope");
}
scope.close();
return;
}
if (error != null) { // an error occurred, adding error to span
String message = error.getMessage();
if (message == null) {
message = error.getClass().getSimpleName();
}
span.tag("error", message);
}
if (log.isDebugEnabled()) {
log.debug("Will finish the and its corresponding scope " + span);
}
span.end();
scope.close();
}
private SpanAndScope getSpanFromThreadLocal() {
SpanAndScope span = this.threadLocalSpan.get();
if (log.isDebugEnabled()) {
log.debug("Took span [" + span + "] from thread local");
}
this.threadLocalSpan.remove();
return span;
@Override
public ThreadLocalSpan getThreadLocalSpan() {
return this.threadLocalSpan;
}
private MessageHeaderAccessor mutableHeaderAccessor(Message<?> message) {

75
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ReactorSleuth.java
View File

@@ -334,15 +334,26 @@ public abstract class ReactorSleuth {
public static <T> Mono<T> tracedMono(@NonNull Tracer tracer, @NonNull CurrentTraceContext currentTraceContext,
@NonNull String childSpanName, @NonNull Supplier<Mono<T>> supplier,
@NonNull Consumer<Span> spanCustomizer) {
return runMonoSupplierInScope(supplier, spanCustomizer).contextWrite(
context -> ReactorSleuth.enhanceContext(tracer, currentTraceContext, context, childSpanName));
}
private static <T> Mono<T> runMonoSupplierInScope(Supplier<Mono<T>> supplier, Consumer<Span> spanCustomizer) {
return Mono.deferContextual(contextView -> {
Span span = contextView.get(Span.class);
spanCustomizer.accept(span);
Tracer.SpanInScope scope = contextView.get(Tracer.SpanInScope.class);
return supplier.get().doOnError(span::error).doFinally(signalType -> {
span.end();
scope.close();
});
}).contextWrite(context -> ReactorSleuth.enhanceContext(tracer, currentTraceContext, context, childSpanName));
// @formatter:off
return supplier.get()
// TODO: Fix me when this is resolved in Reactor
// .doOnSubscribe(__ -> scope.close())
.doOnError(span::error)
.doFinally(signalType -> {
span.end();
scope.close();
});
// @formatter:on
});
}
/**
@@ -361,6 +372,20 @@ public abstract class ReactorSleuth {
});
}
/**
* Wraps the given Mono in a trace representation. Puts the provided span to context.
* @param tracer - Tracer bean
* @param span - span to put in context
* @param supplier - supplier of a {@link Mono} to be wrapped in tracing
* @param <T> - type returned by the Mono
* @return traced Mono
*/
public static <T> Mono<T> tracedMono(@NonNull Tracer tracer, @NonNull Span span,
@NonNull Supplier<Mono<T>> supplier) {
return runMonoSupplierInScope(supplier, span1 -> {
}).contextWrite(context -> ReactorSleuth.putSpanInScope(tracer, context, span));
}
/**
* Wraps the given Flux in a trace representation. Retrieves the span from context,
* creates a child span with the given name.
@@ -375,15 +400,41 @@ public abstract class ReactorSleuth {
public static <T> Flux<T> tracedFlux(@NonNull Tracer tracer, @NonNull CurrentTraceContext currentTraceContext,
@NonNull String childSpanName, @NonNull Supplier<Flux<T>> supplier,
@NonNull Consumer<Span> spanCustomizer) {
return runFluxSupplierInScope(supplier, spanCustomizer).contextWrite(
context -> ReactorSleuth.enhanceContext(tracer, currentTraceContext, context, childSpanName));
}
/**
* Wraps the given Flux in a trace representation. Retrieves the span from context,
* creates a child span with the given name.
* @param tracer - Tracer bean
* @param span - span to put in context
* @param supplier - supplier of a {@link Flux} to be wrapped in tracing
* @param <T> - type returned by the Flux
* @return traced Flux
*/
public static <T> Flux<T> tracedFlux(@NonNull Tracer tracer, @NonNull Span span,
@NonNull Supplier<Flux<T>> supplier) {
return runFluxSupplierInScope(supplier, span1 -> {
}).contextWrite(context -> ReactorSleuth.putSpanInScope(tracer, context, span));
}
private static <T> Flux<T> runFluxSupplierInScope(Supplier<Flux<T>> supplier, Consumer<Span> spanCustomizer) {
return Flux.deferContextual(contextView -> {
Span span = contextView.get(Span.class);
spanCustomizer.accept(span);
Tracer.SpanInScope scope = contextView.get(Tracer.SpanInScope.class);
return supplier.get().doOnError(span::error).doFinally(signalType -> {
span.end();
scope.close();
});
}).contextWrite(context -> ReactorSleuth.enhanceContext(tracer, currentTraceContext, context, childSpanName));
// @formatter:off
return supplier.get()
// TODO: Fix me when this is resolved in Reactor
// .doOnSubscribe(__ -> scope.close())
.doOnError(span::error)
.doFinally(signalType -> {
span.end();
scope.close();
});
// @formatter:on
});
}
/**
@@ -442,6 +493,10 @@ public abstract class ReactorSleuth {
private static Context enhanceContext(Tracer tracer, CurrentTraceContext currentTraceContext,
reactor.util.context.Context context, String childSpanName) {
Span span = spanFromContext(tracer, currentTraceContext, context, childSpanName);
return putSpanInScope(tracer, context, span);
}
private static Context putSpanInScope(Tracer tracer, Context context, Span span) {
return context.put(Span.class, span).put(TraceContext.class, span.context()).put(Tracer.SpanInScope.class,
tracer.withSpan(span));
}

38
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/ByteBufGetter.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* 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 io.netty.buffer.ByteBuf;
import io.netty.util.CharsetUtil;
import io.rsocket.metadata.CompositeMetadata;
import org.springframework.cloud.sleuth.propagation.Propagator;
class ByteBufGetter implements Propagator.Getter<ByteBuf> {
@Override
public String get(ByteBuf carrier, String key) {
final CompositeMetadata compositeMetadata = new CompositeMetadata(carrier, false);
for (CompositeMetadata.Entry entry : compositeMetadata) {
if (key.equals(entry.getMimeType())) {
return entry.getContent().toString(CharsetUtil.UTF_8);
}
}
return null;
}
}

35
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/ByteBufSetter.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* 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 io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.CompositeByteBuf;
import io.rsocket.metadata.CompositeMetadataCodec;
import org.springframework.cloud.sleuth.propagation.Propagator;
class ByteBufSetter implements Propagator.Setter<CompositeByteBuf> {
@Override
public void set(CompositeByteBuf carrier, String key, String value) {
final ByteBufAllocator alloc = carrier.alloc();
CompositeMetadataCodec.encodeAndAddMetadataWithCompression(carrier, alloc, key,
ByteBufUtil.writeUtf8(alloc, value));
}
}

42
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/CompositeMetadataUtils.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* 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 io.netty.buffer.ByteBuf;
import io.rsocket.metadata.CompositeMetadata;
import org.springframework.lang.Nullable;
final class CompositeMetadataUtils {
private CompositeMetadataUtils() {
throw new IllegalStateException("Can't instantiate a utility class");
}
@Nullable
static ByteBuf extract(ByteBuf metadata, String key) {
final CompositeMetadata compositeMetadata = new CompositeMetadata(metadata, false);
for (CompositeMetadata.Entry entry : compositeMetadata) {
final String entryKey = entry.getMimeType();
if (key.equals(entryKey)) {
return entry.getContent();
}
}
return null;
}
}

68
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/PayloadUtils.java Normal file
View File

@@ -0,0 +1,68 @@
/*
* 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.util.HashSet;
import java.util.Set;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;
import io.rsocket.Payload;
import io.rsocket.metadata.CompositeMetadata;
import io.rsocket.metadata.CompositeMetadata.Entry;
import io.rsocket.metadata.CompositeMetadataCodec;
import io.rsocket.metadata.WellKnownMimeType;
import io.rsocket.util.ByteBufPayload;
import io.rsocket.util.DefaultPayload;
final class PayloadUtils {
private PayloadUtils() {
throw new IllegalStateException("Can't instantiate a utility class");
}
static Payload cleanTracingMetadata(Payload payload, Set<String> fields) {
Set<String> fieldsWithDefaultZipkin = new HashSet<>(fields);
fieldsWithDefaultZipkin.add(WellKnownMimeType.MESSAGE_RSOCKET_TRACING_ZIPKIN.getString());
final CompositeMetadata entries = new CompositeMetadata(payload.metadata(), true);
final CompositeByteBuf metadata = ByteBufAllocator.DEFAULT.compositeBuffer();
for (Entry entry : entries) {
if (!fieldsWithDefaultZipkin.contains(entry.getMimeType())) {
CompositeMetadataCodec.encodeAndAddMetadataWithCompression(metadata, ByteBufAllocator.DEFAULT,
entry.getMimeType(), entry.getContent());
}
}
return payload(payload, metadata);
}
private static Payload payload(Payload payload, CompositeByteBuf metadata) {
final Payload newPayload;
try {
if (payload instanceof ByteBufPayload) {
newPayload = ByteBufPayload.create(payload.data().retain(), metadata.retain());
}
else {
newPayload = DefaultPayload.create(payload.data().retain(), metadata.retain());
}
}
finally {
payload.release();
}
return newPayload;
}
}

49
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TracingRSocketConnectorConfigurer.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* 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 io.rsocket.core.RSocketConnector;
import io.rsocket.plugins.RSocketInterceptor;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.propagation.Propagator;
import org.springframework.messaging.rsocket.RSocketConnectorConfigurer;
public class TracingRSocketConnectorConfigurer implements RSocketConnectorConfigurer {
private final Propagator propagator;
private final Tracer tracer;
private final boolean isZipkinPropagationEnabled;
public TracingRSocketConnectorConfigurer(Propagator propagator, Tracer tracer, boolean isZipkinPropagationEnabled) {
this.propagator = propagator;
this.tracer = tracer;
this.isZipkinPropagationEnabled = isZipkinPropagationEnabled;
}
@Override
public void configure(RSocketConnector rSocketConnector) {
rSocketConnector.interceptors(ir -> ir
.forResponder((RSocketInterceptor) rSocket -> new TracingResponderRSocketProxy(rSocket, this.propagator,
new ByteBufGetter(), this.tracer, this.isZipkinPropagationEnabled))
.forRequester((RSocketInterceptor) rSocket -> new TracingRequesterRSocketProxy(rSocket, this.propagator,
new ByteBufSetter(), this.tracer, this.isZipkinPropagationEnabled)));
}
}

49
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TracingRSocketServerCustomizer.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* 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 io.rsocket.core.RSocketServer;
import io.rsocket.plugins.RSocketInterceptor;
import org.springframework.boot.rsocket.server.RSocketServerCustomizer;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.propagation.Propagator;
public class TracingRSocketServerCustomizer implements RSocketServerCustomizer {
private final Propagator propagator;
private final Tracer tracer;
private final boolean isZipkinPropagationEnabled;
public TracingRSocketServerCustomizer(Propagator propagator, Tracer tracer, boolean isZipkinPropagationEnabled) {
this.propagator = propagator;
this.tracer = tracer;
this.isZipkinPropagationEnabled = isZipkinPropagationEnabled;
}
@Override
public void customize(RSocketServer rSocketServer) {
rSocketServer.interceptors(ir -> ir
.forResponder((RSocketInterceptor) rSocket -> new TracingResponderRSocketProxy(rSocket, propagator,
new ByteBufGetter(), this.tracer, this.isZipkinPropagationEnabled))
.forRequester((RSocketInterceptor) rSocket -> new TracingRequesterRSocketProxy(rSocket, propagator,
new ByteBufSetter(), tracer, isZipkinPropagationEnabled)));
}
}

168
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TracingRequesterRSocketProxy.java Normal file
View File

@@ -0,0 +1,168 @@
/*
* 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.util.HashSet;
import java.util.Iterator;
import java.util.function.Function;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.CompositeByteBuf;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.frame.FrameType;
import io.rsocket.metadata.RoutingMetadata;
import io.rsocket.metadata.TracingMetadataCodec;
import io.rsocket.metadata.WellKnownMimeType;
import io.rsocket.util.RSocketProxy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.context.ContextView;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.TraceContext;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.internal.EncodingUtils;
import org.springframework.cloud.sleuth.propagation.Propagator;
/**
* Tracing representation of a {@link RSocketProxy} for the requester.
*
* @author Marcin Grzejszczak
* @author Oleh Dokuka
* @since 3.1.0
*/
public class TracingRequesterRSocketProxy extends RSocketProxy {
private static final Log log = LogFactory.getLog(TracingRequesterRSocketProxy.class);
private final Propagator propagator;
private final Propagator.Setter<CompositeByteBuf> setter;
private final Tracer tracer;
private final boolean isZipkinPropagationEnabled;
public TracingRequesterRSocketProxy(RSocket source, Propagator propagator,
Propagator.Setter<CompositeByteBuf> setter, Tracer tracer, boolean isZipkinPropagationEnabled) {
super(source);
this.propagator = propagator;
this.setter = setter;
this.tracer = tracer;
this.isZipkinPropagationEnabled = isZipkinPropagationEnabled;
}
@Override
public Mono<Void> fireAndForget(Payload payload) {
return setSpan(super::fireAndForget, payload, FrameType.REQUEST_FNF);
}
@Override
public Mono<Payload> requestResponse(Payload payload) {
return setSpan(super::requestResponse, payload, FrameType.REQUEST_RESPONSE);
}
<T> Mono<T> setSpan(Function<Payload, Mono<T>> input, Payload payload, FrameType frameType) {
return Mono.deferContextual(contextView -> {
Span.Builder spanBuilder = spanBuilder(contextView);
ByteBuf extracted = CompositeMetadataUtils.extract(payload.sliceMetadata(),
WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString());
// TODO: do sth about extracted == null, log that tracing can't be used or sth
final RoutingMetadata routingMetadata = new RoutingMetadata(extracted);
final Iterator<String> iterator = routingMetadata.iterator();
String route = iterator.next();
Span span = spanBuilder.kind(Span.Kind.PRODUCER).name(frameType.name() + " " + route).start();
span.tag("rsocket.route", route);
span.tag("rsocket.request-type", frameType.name());
if (log.isDebugEnabled()) {
log.debug("Extracted result from context or thread local " + span);
}
final Payload newPayload = PayloadUtils.cleanTracingMetadata(payload, new HashSet<>(propagator.fields()));
TraceContext traceContext = span.context();
if (this.isZipkinPropagationEnabled) {
injectDefaultZipkinRSocketHeaders(newPayload, traceContext);
}
this.propagator.inject(traceContext, (CompositeByteBuf) newPayload.metadata(), this.setter);
return input.apply(newPayload).doOnError(span::error).doFinally(signalType -> span.end());
});
}
private void injectDefaultZipkinRSocketHeaders(Payload newPayload, TraceContext traceContext) {
TracingMetadataCodec.Flags flags = traceContext.sampled() == null ? TracingMetadataCodec.Flags.UNDECIDED
: traceContext.sampled() ? TracingMetadataCodec.Flags.SAMPLE : TracingMetadataCodec.Flags.NOT_SAMPLE;
String traceId = traceContext.traceId();
long[] traceIds = EncodingUtils.fromString(traceId);
long[] spanId = EncodingUtils.fromString(traceContext.spanId());
long[] parentSpanId = EncodingUtils.fromString(traceContext.parentId());
boolean isTraceId128Bit = traceIds.length == 2;
if (isTraceId128Bit) {
TracingMetadataCodec.encode128(newPayload.metadata().alloc(), traceIds[0], traceIds[1], spanId[0],
EncodingUtils.fromString(traceContext.parentId())[0], flags);
}
else {
TracingMetadataCodec.encode64(newPayload.metadata().alloc(), traceIds[0], spanId[0], parentSpanId[0],
flags);
}
}
private Span.Builder spanBuilder(ContextView contextView) {
Span.Builder spanBuilder = this.tracer.spanBuilder();
if (contextView.hasKey(TraceContext.class)) {
spanBuilder = spanBuilder.setParent(contextView.get(TraceContext.class));
}
else if (this.tracer.currentSpan() != null) {
spanBuilder = spanBuilder.setParent(this.tracer.currentSpan().context());
}
return spanBuilder;
}
@Override
public Flux<Payload> requestStream(Payload payload) {
return Flux.deferContextual(contextView -> setSpan(super::requestStream, payload, contextView));
}
@Override
public Flux<Payload> requestChannel(Publisher<Payload> inbound) {
return Flux.from(inbound).switchOnFirst((firstSignal, flux) -> {
final Payload firstPayload = firstSignal.get();
if (firstPayload != null) {
return setSpan(p -> super.requestChannel(flux.skip(1).startWith(p)), firstPayload,
firstSignal.getContextView());
}
return flux;
});
}
<T> Flux<Payload> setSpan(Function<Payload, Flux<Payload>> input, Payload payload, ContextView contextView) {
Span.Builder spanBuilder = spanBuilder(contextView);
final RoutingMetadata routingMetadata = new RoutingMetadata(CompositeMetadataUtils
.extract(payload.sliceMetadata(), WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString()));
final Iterator<String> iterator = routingMetadata.iterator();
Span span = spanBuilder.kind(Span.Kind.PRODUCER).name(iterator.next()).start();
if (log.isDebugEnabled()) {
log.debug("Extracted result from context or thread local " + span);
}
final Payload newPayload = PayloadUtils.cleanTracingMetadata(payload, new HashSet<>(propagator.fields()));
this.propagator.inject(span.context(), (CompositeByteBuf) newPayload.metadata(), this.setter);
return input.apply(newPayload).doOnError(span::error).doFinally(signalType -> span.end());
}
}

170
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TracingResponderRSocketProxy.java Normal file
View File

@@ -0,0 +1,170 @@
/*
* 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.util.HashSet;
import java.util.Iterator;
import io.netty.buffer.ByteBuf;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.frame.FrameType;
import io.rsocket.metadata.RoutingMetadata;
import io.rsocket.metadata.TracingMetadata;
import io.rsocket.metadata.TracingMetadataCodec;
import io.rsocket.metadata.WellKnownMimeType;
import io.rsocket.util.RSocketProxy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.ThreadLocalSpan;
import org.springframework.cloud.sleuth.TraceContext;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.cloud.sleuth.WithThreadLocalSpan;
import org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth;
import org.springframework.cloud.sleuth.internal.EncodingUtils;
import org.springframework.cloud.sleuth.propagation.Propagator;
/**
* Tracing representation of a {@link RSocketProxy} for the responder.
*
* @author Marcin Grzejszczak
* @author Oleh Dokuka
* @since 3.1.0
*/
public class TracingResponderRSocketProxy extends RSocketProxy implements WithThreadLocalSpan {
private static final Log log = LogFactory.getLog(TracingResponderRSocketProxy.class);
private final Propagator propagator;
private final Propagator.Getter<ByteBuf> getter;
private final Tracer tracer;
private final ThreadLocalSpan threadLocalSpan;
private final boolean isZipkinPropagationEnabled;
public TracingResponderRSocketProxy(RSocket source, Propagator propagator, Propagator.Getter<ByteBuf> getter,
Tracer tracer, boolean isZipkinPropagationEnabled) {
super(source);
this.propagator = propagator;
this.getter = getter;
this.tracer = tracer;
this.threadLocalSpan = new ThreadLocalSpan(tracer);
this.isZipkinPropagationEnabled = isZipkinPropagationEnabled;
}
@Override
public Mono<Void> fireAndForget(Payload payload) {
// called on Netty EventLoop
// there can't be trace context in thread local here
Span handle = consumerSpanBuilder(payload.sliceMetadata(), FrameType.REQUEST_FNF);
if (log.isDebugEnabled()) {
log.debug("Created consumer span " + handle);
}
final Payload newPayload = PayloadUtils.cleanTracingMetadata(payload, new HashSet<>(propagator.fields()));
return ReactorSleuth.tracedMono(this.tracer, handle, () -> super.fireAndForget(newPayload));
}
@Override
public Mono<Payload> requestResponse(Payload payload) {
Span handle = consumerSpanBuilder(payload.sliceMetadata(), FrameType.REQUEST_RESPONSE);
if (log.isDebugEnabled()) {
log.debug("Created consumer span " + handle);
}
final Payload newPayload = PayloadUtils.cleanTracingMetadata(payload, new HashSet<>(propagator.fields()));
return ReactorSleuth.tracedMono(this.tracer, handle, () -> super.requestResponse(newPayload));
}
@Override
public Flux<Payload> requestStream(Payload payload) {
Span handle = consumerSpanBuilder(payload.sliceMetadata(), FrameType.REQUEST_STREAM);
if (log.isDebugEnabled()) {
log.debug("Created consumer span " + handle);
}
final Payload newPayload = PayloadUtils.cleanTracingMetadata(payload, new HashSet<>(propagator.fields()));
return ReactorSleuth.tracedFlux(this.tracer, handle, () -> super.requestStream(newPayload));
}
@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
return Flux.from(payloads).switchOnFirst((firstSignal, flux) -> {
final Payload firstPayload = firstSignal.get();
if (firstPayload != null) {
Span handle = consumerSpanBuilder(firstPayload.sliceMetadata(), FrameType.REQUEST_CHANNEL);
if (handle == null) {
return super.requestChannel(flux);
}
if (log.isDebugEnabled()) {
log.debug("Created consumer span " + handle);
}
final Payload newPayload = PayloadUtils.cleanTracingMetadata(firstPayload,
new HashSet<>(propagator.fields()));
return ReactorSleuth.tracedFlux(this.tracer, handle,
() -> super.requestChannel(flux.skip(1).startWith(newPayload)));
}
return flux;
});
}
private Span consumerSpanBuilder(ByteBuf headers, FrameType requestType) {
Span.Builder consumerSpanBuilder = consumerSpanBuilder(headers);
if (log.isDebugEnabled()) {
log.debug("Extracted result from headers " + consumerSpanBuilder);
}
final ByteBuf extract = CompositeMetadataUtils.extract(headers,
WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString());
String name = "handle";
if (extract != null) {
final RoutingMetadata routingMetadata = new RoutingMetadata(extract);
final Iterator<String> iterator = routingMetadata.iterator();
name = requestType.name() + " " + iterator.next();
}
return consumerSpanBuilder.kind(Span.Kind.CONSUMER).name(name).start();
}
private Span.Builder consumerSpanBuilder(ByteBuf headers) {
if (this.isZipkinPropagationEnabled) {
ByteBuf extract = CompositeMetadataUtils.extract(headers,
WellKnownMimeType.MESSAGE_RSOCKET_TRACING_ZIPKIN.getString());
if (extract != null) {
TracingMetadata tracingMetadata = TracingMetadataCodec.decode(extract);
Span.Builder builder = this.tracer.spanBuilder();
TraceContext.Builder parentBuilder = this.tracer.traceContextBuilder()
.sampled(tracingMetadata.isSampled()).traceId(EncodingUtils.fromLong(tracingMetadata.traceId()))
.parentId(EncodingUtils.fromLong(tracingMetadata.parentId()));
return builder.setParent(parentBuilder.build());
}
else {
return this.propagator.extract(headers, this.getter);
}
}
return this.propagator.extract(headers, this.getter);
}
@Override
public ThreadLocalSpan getThreadLocalSpan() {
return this.threadLocalSpan;
}
}

322
spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/internal/EncodingUtils.java Normal file
View File

@@ -0,0 +1,322 @@
/*
* 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.internal;
import java.util.Arrays;
import org.springframework.util.Assert;
/**
* Adopted from OpenTelemetry API.
*
* @since 3.1.0
*/
public final class EncodingUtils {
private static final ThreadLocal<char[]> charBuffer = new ThreadLocal();
private EncodingUtils() {
}
static final int LONG_BYTES = Long.SIZE / Byte.SIZE;
static final int BYTE_BASE16 = 2;
static final int LONG_BASE16 = BYTE_BASE16 * LONG_BYTES;
private static final String ALPHABET = "0123456789abcdef";
private static final int ASCII_CHARACTERS = 128;
private static final char[] ENCODING = buildEncodingArray();
private static final byte[] DECODING = buildDecodingArray();
private static char[] buildEncodingArray() {
char[] encoding = new char[512];
for (int i = 0; i < 256; ++i) {
encoding[i] = ALPHABET.charAt(i >>> 4);
encoding[i | 0x100] = ALPHABET.charAt(i & 0xF);
}
return encoding;
}
private static byte[] buildDecodingArray() {
byte[] decoding = new byte[ASCII_CHARACTERS];
Arrays.fill(decoding, (byte) -1);
for (int i = 0; i < ALPHABET.length(); i++) {
char c = ALPHABET.charAt(i);
decoding[c] = (byte) i;
}
return decoding;
}
/**
* Returns the {@code long} value.
* @param chars the base8 or base16 representation of the {@code long}.
* @return long array from string. Either contains high and low or just low trace id
*/
public static long[] fromString(CharSequence chars) {
if (chars == null || chars.length() == 0) {
return new long[] { 0 };
}
if (chars.length() == 32) {
long high = HexCodec.lenientLowerHexToUnsignedLong(chars, 0, 16);
long low = HexCodec.lenientLowerHexToUnsignedLong(chars, 16, 32);
return new long[] { high, low };
}
return new long[] { HexCodec.lenientLowerHexToUnsignedLong(chars, 0, 16) };
}
/**
* Returns the {@code long} value whose base16 representation is stored in the first
* 16 chars of {@code chars} starting from the {@code offset}.
* @param chars the base16 representation of the {@code long}.
* @return long value from string
*/
public static long longFromBase16String(CharSequence chars) {
return longFromBase16String(chars, 0);
}
/**
* Returns the {@code long} value whose base16 representation is stored in the first
* 16 chars of {@code chars} starting from the {@code offset}.
* @param chars the base16 representation of the {@code long}.
*/
static long longFromBase16String(CharSequence chars, int offset) {
Assert.isTrue(chars.length() >= offset + LONG_BASE16, "chars too small");
return (decodeByte(chars.charAt(offset), chars.charAt(offset + 1)) & 0xFFL) << 56
| (decodeByte(chars.charAt(offset + 2), chars.charAt(offset + 3)) & 0xFFL) << 48
| (decodeByte(chars.charAt(offset + 4), chars.charAt(offset + 5)) & 0xFFL) << 40
| (decodeByte(chars.charAt(offset + 6), chars.charAt(offset + 7)) & 0xFFL) << 32
| (decodeByte(chars.charAt(offset + 8), chars.charAt(offset + 9)) & 0xFFL) << 24
| (decodeByte(chars.charAt(offset + 10), chars.charAt(offset + 11)) & 0xFFL) << 16
| (decodeByte(chars.charAt(offset + 12), chars.charAt(offset + 13)) & 0xFFL) << 8
| (decodeByte(chars.charAt(offset + 14), chars.charAt(offset + 15)) & 0xFFL);
}
/**
* Decodes the specified two character sequence, and returns the resulting
* {@code byte}.
* @param chars the character sequence to be decoded.
* @param offset the starting offset in the {@code CharSequence}.
* @return the resulting {@code byte}
* @throws IllegalArgumentException if the input is not a valid encoded string
* according to this encoding.
*/
public static byte byteFromBase16String(CharSequence chars, int offset) {
Assert.isTrue(chars.length() >= offset + 2, "chars too small");
return decodeByte(chars.charAt(offset), chars.charAt(offset + 1));
}
private static byte decodeByte(char hi, char lo) {
Assert.isTrue(lo < ASCII_CHARACTERS && DECODING[lo] != -1, "invalid character " + lo);
Assert.isTrue(hi < ASCII_CHARACTERS && DECODING[hi] != -1, "invalid character " + hi);
int decoded = DECODING[hi] << 4 | DECODING[lo];
return (byte) decoded;
}
/**
* Checks if string is valid base16.
* @param value to check
* @return {@code true} if valid base16 string
*/
public static boolean isValidBase16String(CharSequence value) {
for (int i = 0; i < value.length(); i++) {
char b = value.charAt(i);
// 48..57 && 97..102 are valid
if (!isDigit(b) && !isLowercaseHexCharacter(b)) {
return false;
}
}
return true;
}
/**
* Converts long into string.
* @param id 64 bit
* @return string representation of the long
*/
public static String fromLong(long id) {
return fromLongs(0, id);
}
/**
* Converts longs into string.
* @param idHigh - trace id high part
* @param idLow - trace id low part
* @return string representation of the long
*/
public static String fromLongs(long idHigh, long idLow) {
if (idHigh == 0L) {
return HexCodec.toLowerHex(idLow);
}
else {
char[] chars = getTemporaryBuffer();
longToBase16String(idHigh, chars, 0);
longToBase16String(idLow, chars, 16);
return new String(chars);
}
}
public static void longToBase16String(long value, char[] dest, int destOffset) {
byteToBase16((byte) ((int) (value >> 56 & 255L)), dest, destOffset);
byteToBase16((byte) ((int) (value >> 48 & 255L)), dest, destOffset + 2);
byteToBase16((byte) ((int) (value >> 40 & 255L)), dest, destOffset + 4);
byteToBase16((byte) ((int) (value >> 32 & 255L)), dest, destOffset + 6);
byteToBase16((byte) ((int) (value >> 24 & 255L)), dest, destOffset + 8);
byteToBase16((byte) ((int) (value >> 16 & 255L)), dest, destOffset + 10);
byteToBase16((byte) ((int) (value >> 8 & 255L)), dest, destOffset + 12);
byteToBase16((byte) ((int) (value & 255L)), dest, destOffset + 14);
}
public static void byteToBase16(byte value, char[] dest, int destOffset) {
int b = value & 255;
dest[destOffset] = ENCODING[b];
dest[destOffset + 1] = ENCODING[b | 256];
}
private static char[] getTemporaryBuffer() {
char[] chars = charBuffer.get();
if (chars == null) {
chars = new char[32];
charBuffer.set(chars);
}
return chars;
}
private static boolean isLowercaseHexCharacter(char b) {
return 97 <= b && b <= 102;
}
private static boolean isDigit(char b) {
return 48 <= b && b <= 57;
}
}
// taken from brave.internal.codec.HexCodec
final class HexCodec {
private HexCodec() {
throw new IllegalStateException("Can't instantiate a utility class");
}
/**
* Parses a 16 character lower-hex string with no prefix into an unsigned long,
* starting at the specified index.
*
* This reads a trace context a sequence potentially larger than the format. The
* use-case is reducing garbage, by re-using the input {@code value} across multiple
* parse operations.
* @param value the sequence that contains a lower-hex encoded unsigned long.
* @param beginIndex the inclusive begin index: {@linkplain CharSequence#charAt(int)
* index} of the first lower-hex character representing the unsigned long.
*/
static long lowerHexToUnsignedLong(CharSequence value, int beginIndex) {
int endIndex = Math.min(beginIndex + 16, value.length());
long result = lenientLowerHexToUnsignedLong(value, beginIndex, endIndex);
if (result == 0) {
throw isntLowerHexLong(value);
}
return result;
}
/**
* Like {@link #lowerHexToUnsignedLong(CharSequence, int)}, but returns zero on
* invalid input.
* @param value the sequence that contains a lower-hex encoded unsigned long.
* @param beginIndex the inclusive begin index: {@linkplain CharSequence#charAt(int)
* index} of the first lower-hex character representing the unsigned long.
* @param endIndex the exclusive end index: {@linkplain CharSequence#charAt(int)
* index} after the last lower-hex character representing the unsigned long.
*/
static long lenientLowerHexToUnsignedLong(CharSequence value, int beginIndex, int endIndex) {
long result = 0;
int pos = beginIndex;
while (pos < endIndex) {
char c = value.charAt(pos++);
result <<= 4;
if (c >= '0' && c <= '9') {
result |= c - '0';
}
else if (c >= 'a' && c <= 'f') {
result |= c - 'a' + 10;
}
else {
return 0;
}
}
return result;
}
static NumberFormatException isntLowerHexLong(CharSequence lowerHex) {
throw new NumberFormatException(lowerHex + " should be a 1 to 32 character lower-hex string with no prefix");
}
/** Inspired by {@code okio.Buffer.writeLong}. */
static String toLowerHex(long v) {
char[] data = RecyclableBuffers.parseBuffer();
writeHexLong(data, 0, v);
return new String(data, 0, 16);
}
/** Inspired by {@code okio.Buffer.writeLong}. */
static void writeHexLong(char[] data, int pos, long v) {
writeHexByte(data, pos + 0, (byte) ((v >>> 56L) & 0xff));
writeHexByte(data, pos + 2, (byte) ((v >>> 48L) & 0xff));
writeHexByte(data, pos + 4, (byte) ((v >>> 40L) & 0xff));
writeHexByte(data, pos + 6, (byte) ((v >>> 32L) & 0xff));
writeHexByte(data, pos + 8, (byte) ((v >>> 24L) & 0xff));
writeHexByte(data, pos + 10, (byte) ((v >>> 16L) & 0xff));
writeHexByte(data, pos + 12, (byte) ((v >>> 8L) & 0xff));
writeHexByte(data, pos + 14, (byte) (v & 0xff));
}
static final char[] HEX_DIGITS = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
static void writeHexByte(char[] data, int pos, byte b) {
data[pos + 0] = HEX_DIGITS[(b >> 4) & 0xf];
data[pos + 1] = HEX_DIGITS[b & 0xf];
}
}
// taken from brave
final class RecyclableBuffers {
private RecyclableBuffers() {
throw new IllegalStateException("Can't instantiate a utility class");
}
private static final ThreadLocal<char[]> PARSE_BUFFER = new ThreadLocal<>();
/**
* Returns a {@link ThreadLocal} reused {@code char[]} for use when decoding bytes
* into an ID hex string. The buffer should be immediately copied into a
* {@link String} after decoding within the same method.
*/
static char[] parseBuffer() {
char[] idBuffer = PARSE_BUFFER.get();
if (idBuffer == null) {
idBuffer = new char[32 + 1 + 16 + 3 + 16]; // traceid128-spanid-1-parentid
PARSE_BUFFER.set(idBuffer);
}
return idBuffer;
}
}

5
spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/instrument/deployer/SimpleTracer.java
View File

@@ -85,6 +85,11 @@ class SimpleTracer implements Tracer {
return null;
}
@Override
public TraceContext.Builder traceContextBuilder() {
return null;
}
@Override
public Map<String, String> getAllBaggage() {
return new HashMap<>();

43
spring-cloud-sleuth-instrumentation/src/test/java/org/springframework/cloud/sleuth/internal/EncodingUtilsTests.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* 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.internal;
import org.assertj.core.api.BDDAssertions;
import org.junit.jupiter.api.Test;
class EncodingUtilsTests {
@Test
void should_convert_back_and_forth_with_64bits() {
long[] fromString = EncodingUtils.fromString("7c6239a5ad0a4287");
BDDAssertions.then(fromString).hasSize(1);
String fromLong = EncodingUtils.fromLong(fromString[0]);
BDDAssertions.then(fromLong).isEqualTo("7c6239a5ad0a4287");
}
@Test
void should_convert_back_and_forth_with_128bits() {
long[] fromString = EncodingUtils.fromString("596e1787feb110407c6239a5ad0a4287");
BDDAssertions.then(fromString).hasSize(2);
String fromLong = EncodingUtils.fromLongs(fromString[0], fromString[1]);
BDDAssertions.then(fromLong).isEqualTo("596e1787feb110407c6239a5ad0a4287");
}
}

1
tests/brave/pom.xml
View File

@@ -54,6 +54,7 @@
<module>spring-cloud-sleuth-instrumentation-scheduling-tests</module>
<module>spring-cloud-sleuth-instrumentation-task-tests</module>
<module>spring-cloud-sleuth-instrumentation-webflux-tests</module>
<module>spring-cloud-sleuth-instrumentation-rsocket-tests</module>
<module>spring-cloud-sleuth-zipkin-tests</module>
</modules>

100
tests/brave/spring-cloud-sleuth-instrumentation-rsocket-tests/pom.xml Normal file
View File

@@ -0,0 +1,100 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
~
~
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-sleuth-instrumentation-rsocket-tests</artifactId>
<packaging>jar</packaging>
<name>Spring Cloud Sleuth Brave RSocket Instrumentation Tests</name>
<description>Spring Cloud Sleuth Brave RSocket Instrumentation Tests</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-tests-brave</artifactId>
<version>3.1.0-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<properties>
<sonar.skip>true</sonar.skip>
</properties>
<build>
<plugins>
<plugin>
<!--skip deploy -->
<artifactId>maven-deploy-plugin</artifactId>
<configuration>
<skip>true</skip>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>spring-cloud-sleuth-tests-common</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-loadbalancer</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-sleuth</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.brave</groupId>
<artifactId>brave-tests</artifactId>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
</dependency>
</dependencies>
</project>

53
tests/brave/spring-cloud-sleuth-instrumentation-rsocket-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/rsocket/TraceRSocketTests.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* 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.brave.instrument.rsocket;
import brave.sampler.Sampler;
import org.springframework.cloud.sleuth.brave.BraveTestSpanHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
public class TraceRSocketTests extends org.springframework.cloud.sleuth.instrument.rsocket.TraceRSocketTests {
@Override
protected Class testConfiguration() {
return Config.class;
}
@Configuration(proxyBeanMethods = false)
static class Config {
@Bean
org.springframework.cloud.sleuth.test.TestSpanHandler testSpanHandlerSupplier(
brave.test.TestSpanHandler testSpanHandler) {
return new BraveTestSpanHandler(testSpanHandler);
}
@Bean
Sampler alwaysSampler() {
return Sampler.ALWAYS_SAMPLE;
}
@Bean
brave.test.TestSpanHandler braveTestSpanHandler() {
return new brave.test.TestSpanHandler();
}
}
}

32
tests/brave/spring-cloud-sleuth-instrumentation-rsocket-tests/src/test/resources/logback.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2013-2018 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.
-->
<configuration>
<include resource="org/springframework/boot/logging/logback/base.xml"/>
<logger name="feign" level="DEBUG"/>
<logger name="com.netflix.discovery.InstanceInfoReplicator" level="ERROR"/>
<logger name="org.springframework" level="INFO"/>
<logger name="org.springframework.cloud.sleuth" level="DEBUG"/>
<logger name="org.springframework.boot.autoconfigure.logging" level="INFO"/>
<logger name="org.springframework.cloud.sleuth.log" level="DEBUG"/>
<logger name="org.springframework.cloud.sleuth.trace" level="DEBUG"/>
<logger name="org.springframework.cloud.sleuth.instrument.rxjava" level="DEBUG"/>
<root level="INFO">
<appender-ref ref="CONSOLE"/>
<appender-ref ref="FILE"/>
</root>
</configuration>

465
tests/common/pom.xml
View File

@@ -1,230 +1,235 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
~
~
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-sleuth-tests-common</artifactId>
<packaging>jar</packaging>
<name>Spring Cloud Sleuth Tests Common</name>
<description>Spring Cloud Sleuth Tests Common</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-tests</artifactId>
<version>3.1.0-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-instrumentation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>mockwebserver</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-gateway</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-task</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-circuitbreaker-resilience4j</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.github.openfeign</groupId>
<artifactId>feign-okhttp</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-loadbalancer</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-config-server</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-autoconfigure</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-brave</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.brave</groupId>
<artifactId>brave-tests</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-zipkin</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.zipkin2</groupId>
<artifactId>zipkin</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter-brave</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-kafka</artifactId>
<optional>true</optional>
<exclusions>
<!-- assigned with spring-kafka -->
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-activemq-client</artifactId>
<optional>true</optional>
<exclusions>
<exclusion>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-amqp-client</artifactId>
<optional>true</optional>
<exclusions>
<!-- assigned with spring-rabbit -->
<exclusion>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
</project>
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
~
~
-->
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-sleuth-tests-common</artifactId>
<packaging>jar</packaging>
<name>Spring Cloud Sleuth Tests Common</name>
<description>Spring Cloud Sleuth Tests Common</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-tests</artifactId>
<version>3.1.0-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-instrumentation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>mockwebserver</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-gateway</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-task</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-circuitbreaker-resilience4j</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.github.openfeign</groupId>
<artifactId>feign-okhttp</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-loadbalancer</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-config-server</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-autoconfigure</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-brave</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.brave</groupId>
<artifactId>brave-tests</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.projectreactor.kafka</groupId>
<artifactId>reactor-kafka</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>kafka</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-rsocket</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-sleuth-zipkin</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.zipkin2</groupId>
<artifactId>zipkin</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter-brave</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-kafka</artifactId>
<optional>true</optional>
<exclusions>
<!-- assigned with spring-kafka -->
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-activemq-client</artifactId>
<optional>true</optional>
<exclusions>
<exclusion>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-sender-amqp-client</artifactId>
<optional>true</optional>
<exclusions>
<!-- assigned with spring-rabbit -->
<exclusion>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
</project>

2
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/config/ConfigServerIntegrationTests.java
View File

@@ -88,7 +88,7 @@ public abstract class ConfigServerIntegrationTests {
void call(int port) {
log.info("Sending request");
String result = new RestTemplate().getForObject("http://localhost:" + port + "/master/application.yml",
String result = new RestTemplate().getForObject("http://localhost:" + port + "/foo/default/main",
String.class);
log.info("Got [\n" + result + "\n]");
}

363
tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TraceRSocketTests.java Normal file
View File

@@ -0,0 +1,363 @@
/*
* 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;
});
}
}
}