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

GH-823 Polishing

Browse Source 
Resolves #823
This commit is contained in:
Oleg Zhurakousky
2022-03-04 11:56:26 +01:00
parent ea3be5ef39
commit a364aaf86a
10 changed files with 48 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/observability/DefaultFunctionTagsProvider.java
View File

@@ -16,7 +16,7 @@
package org.springframework.cloud.function.context.catalog.observability;
import io.micrometer.api.instrument.Tags;
import io.micrometer.core.instrument.Tags;
public class DefaultFunctionTagsProvider implements FunctionTagsProvider {
@Override

4
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/observability/FunctionContext.java
View File

@@ -16,8 +16,8 @@
package org.springframework.cloud.function.context.catalog.observability;
import io.micrometer.api.instrument.observation.Observation;
import io.micrometer.api.lang.Nullable;
import io.micrometer.core.instrument.observation.Observation;
import io.micrometer.core.lang.Nullable;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry;

18
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/observability/FunctionObservation.java
View File

@@ -16,18 +16,22 @@
package org.springframework.cloud.function.context.catalog.observability;
import io.micrometer.api.instrument.docs.DocumentedObservation;
import io.micrometer.api.instrument.docs.TagKey;
import io.micrometer.core.instrument.docs.DocumentedObservation;
import io.micrometer.core.instrument.docs.TagKey;
/**
* @author Marcin Grzejszczak
* @author Oleg Zhurakousky
* @since 4.0.0
*/
enum FunctionObservation implements DocumentedObservation {
/**
* Observation created around a function execution
* Observation created around a function execution.
*/
FUNCTION_OBSERVATION {
@Override
public String getName() {
return "spring.function";
return "spring.cloud.function";
}
@Override
@@ -42,7 +46,7 @@ enum FunctionObservation implements DocumentedObservation {
@Override
public String getPrefix() {
return "spring.function";
return "spring.cloud.function";
}
};
@@ -54,7 +58,7 @@ enum FunctionObservation implements DocumentedObservation {
FUNCTION_NAME {
@Override
public String getKey() {
return "spring.function.name";
return "spring.cloud.function.name";
}
}

4
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/observability/FunctionTagsProvider.java
View File

@@ -16,11 +16,11 @@
package org.springframework.cloud.function.context.catalog.observability;
import io.micrometer.api.instrument.observation.Observation;
import io.micrometer.core.instrument.observation.Observation;
/**
* {@link Observation.TagsProvider} for {@link FunctionContext}.
*
*
* @author Marcin Grzejszczak
*/
public interface FunctionTagsProvider extends Observation.TagsProvider<FunctionContext> {

4
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/observability/ObservationFunctionAroundWrapper.java
View File

@@ -16,8 +16,8 @@
package org.springframework.cloud.function.context.catalog.observability;
import io.micrometer.api.instrument.observation.Observation;
import io.micrometer.api.instrument.observation.ObservationRegistry;
import io.micrometer.core.instrument.observation.Observation;
import io.micrometer.core.instrument.observation.ObservationRegistry;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;

33
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/observability/tracing/FunctionTracingObservationHandler.java
View File

@@ -16,11 +16,7 @@
package org.springframework.cloud.function.context.catalog.observability.tracing;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import io.micrometer.api.instrument.observation.Observation;
import io.micrometer.core.instrument.observation.Observation;
import io.micrometer.tracing.Span;
import io.micrometer.tracing.Tracer;
import io.micrometer.tracing.handler.TracingObservationHandler;
@@ -30,6 +26,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry;
import org.springframework.cloud.function.context.catalog.observability.FunctionContext;
import org.springframework.cloud.function.context.message.MessageUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
@@ -37,11 +34,13 @@ import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.StringUtils;
/**
* Function Tracing Observation Handler.
*
* @author Marcin Grzejszczak
* @author Oleg Zhurakousky
* @since 4.0.0
*/
public class FunctionTracingObservationHandler implements TracingObservationHandler<FunctionContext> {
@@ -208,7 +207,7 @@ public class FunctionTracingObservationHandler implements TracingObservationHand
}
private void clearTracingHeaders(MessageHeaderAccessor headers) {
MessageHeaderPropagatorSetter.removeAnyTraceHeaders(headers, this.propagator.fields());
MessageHeaderPropagatorSetter.removeHeaders(headers, this.propagator.fields());
}
/**
@@ -250,22 +249,26 @@ public class FunctionTracingObservationHandler implements TracingObservationHand
}
private String toRemoteServiceName(MessageHeaderAccessor headers) {
for (String key : headers.getMessageHeaders().keySet()) {
if (key.startsWith("kafka_")) {
return "kafka";
}
else if (key.startsWith("amqp_")) {
return "rabbitmq";
}
// for (String key : headers.getMessageHeaders().keySet()) {
// if (key.startsWith("kafka_")) {
// return "kafka";
// }
// else if (key.startsWith("amqp_")) {
// return "rabbitmq";
// }
// }
String serviceName = (String) headers.getHeader(MessageUtils.TARGET_PROTOCOL);
if (!StringUtils.hasLength(serviceName)) {
serviceName = REMOTE_SERVICE_NAME;
}
return REMOTE_SERVICE_NAME;
return serviceName;
}
private Message<?> outputMessage(Message<?> originalMessage, Message<?> retrievedMessage,
MessageHeaderAccessor additionalHeaders) {
MessageHeaderAccessor headers = mutableHeaderAccessor(originalMessage);
if (originalMessage instanceof ErrorMessage errorMessage) {
headers.copyHeaders(MessageHeaderPropagatorSetter.propagationHeaders(additionalHeaders.getMessageHeaders(),
headers.copyHeaders(MessageHeaderPropagatorSetter.copyHeaders(additionalHeaders.getMessageHeaders(),
this.propagator.fields()));
return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders()
: new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage());

16
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/observability/tracing/MessageHeaderPropagatorSetter.java
View File

@@ -32,27 +32,28 @@ import org.springframework.util.LinkedMultiValueMap;
/**
* Setter for Spring Integration based communication.
*
* This always sets native headers in defence of STOMP issues discussed <a href=
* This always sets native headers in defense of STOMP issues discussed <a href=
* "https://github.com/spring-cloud/spring-cloud-sleuth/issues/716#issuecomment-337523705">here</a>.
*
* @author Marcin Grzejszczak
* @author Oleg Zhurakousky
* @since 4.0.0
*/
public class MessageHeaderPropagatorSetter implements Propagator.Setter<MessageHeaderAccessor> {
private static final Log log = LogFactory.getLog(MessageHeaderPropagatorSetter.class);
static Map<String, ?> propagationHeaders(Map<String, ?> headers, List<String> propagationHeaders) {
Map<String, Object> headersToCopy = new HashMap<>();
static Map<String, ?> copyHeaders(Map<String, ?> headers, List<String> headersToCopy) {
Map<String, Object> copiedHeaders = new HashMap<>();
for (Map.Entry<String, ?> entry : headers.entrySet()) {
if (propagationHeaders.contains(entry.getKey())) {
headersToCopy.put(entry.getKey(), entry.getValue());
if (headersToCopy.contains(entry.getKey())) {
copiedHeaders.put(entry.getKey(), entry.getValue());
}
}
return headersToCopy;
return copiedHeaders;
}
static void removeAnyTraceHeaders(MessageHeaderAccessor accessor, List<String> keysToRemove) {
static void removeHeaders(MessageHeaderAccessor accessor, List<String> keysToRemove) {
for (String keyToRemove : keysToRemove) {
accessor.removeHeader(keyToRemove);
if (accessor instanceof NativeMessageHeaderAccessor) {
@@ -85,7 +86,6 @@ public class MessageHeaderPropagatorSetter implements Propagator.Setter<MessageH
Map<String, List<String>> nativeHeaderMap = nativeAccessor.toNativeHeaderMap();
nativeHeaderMap = nativeHeaderMap instanceof LinkedMultiValueMap ? nativeHeaderMap
: new LinkedMultiValueMap<>(nativeHeaderMap);
nativeAccessor.removeHeader(NativeMessageHeaderAccessor.NATIVE_HEADERS);
nativeAccessor.setHeader(NativeMessageHeaderAccessor.NATIVE_HEADERS, nativeHeaderMap);
return nativeAccessor;
}

4
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/observability/ObservationFunctionAroundWrapperIntegrationTests.java
View File

@@ -23,8 +23,8 @@ import java.util.function.BiConsumer;
import java.util.function.Function;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.micrometer.api.instrument.observation.ObservationHandler;
import io.micrometer.api.instrument.simple.SimpleMeterRegistry;
import io.micrometer.core.instrument.observation.ObservationHandler;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import io.micrometer.tracing.Span;
import io.micrometer.tracing.TraceContext;
import io.micrometer.tracing.Tracer;