Polishing Observability changes
This commit is contained in:
@@ -147,6 +147,11 @@
|
||||
<artifactId>micrometer-observation-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.micrometer</groupId>
|
||||
<artifactId>micrometer-tracing-bridge-otel</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
|
||||
@@ -43,6 +43,6 @@ public class DefaultFunctionObservationConvention implements FunctionObservation
|
||||
|
||||
@Override
|
||||
public String getContextualName(FunctionContext context) {
|
||||
return context.getTargetFunction().getFunctionDefinition();
|
||||
return context.getTargetFunction().getFunctionDefinition() + " process";
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,19 +19,24 @@ package org.springframework.cloud.function.observability;
|
||||
import io.micrometer.observation.Observation;
|
||||
|
||||
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry;
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* {@link Observation.Context} for function processing.
|
||||
*
|
||||
* @author Marcin Grzejszczak
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 4.0.0
|
||||
*/
|
||||
public class FunctionContext extends Observation.Context {
|
||||
|
||||
private final SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction;
|
||||
|
||||
public FunctionContext(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) {
|
||||
private final Message<?> message;
|
||||
|
||||
public FunctionContext(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, Message<?> message) {
|
||||
this.targetFunction = targetFunction;
|
||||
this.message = message;
|
||||
}
|
||||
|
||||
public SimpleFunctionRegistry.FunctionInvocationWrapper getTargetFunction() {
|
||||
|
||||
@@ -36,7 +36,11 @@ public class ObservationAutoConfiguration {
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
@ConditionalOnBean(ObservationRegistry.class)
|
||||
public FunctionAroundWrapper observationFunctionAroundWrapper(ObservationRegistry registry, ObjectProvider<FunctionReceiverObservationConvention> functionReceiverObservationConvention, ObjectProvider<FunctionObservationConvention> functionObservationConvention, ObjectProvider<FunctionSenderObservationConvention> functionSenderObservationConvention) {
|
||||
return new ObservationFunctionAroundWrapper(registry, functionReceiverObservationConvention.getIfAvailable(() -> null), functionObservationConvention.getIfAvailable(() -> null), functionSenderObservationConvention.getIfAvailable(() -> null));
|
||||
public FunctionAroundWrapper observationFunctionAroundWrapper(ObservationRegistry registry,
|
||||
ObjectProvider<FunctionReceiverObservationConvention> functionReceiverObservationConvention,
|
||||
ObjectProvider<FunctionObservationConvention> functionObservationConvention,
|
||||
ObjectProvider<FunctionSenderObservationConvention> functionSenderObservationConvention) {
|
||||
return new ObservationFunctionAroundWrapper(registry, functionReceiverObservationConvention.getIfAvailable(() -> null),
|
||||
functionObservationConvention.getIfAvailable(() -> null), functionSenderObservationConvention.getIfAvailable(() -> null));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,7 +46,8 @@ public class ObservationFunctionAroundWrapper extends FunctionAroundWrapper {
|
||||
|
||||
private final FunctionSenderObservationConvention functionSenderObservationConvention;
|
||||
|
||||
public ObservationFunctionAroundWrapper(ObservationRegistry observationRegistry, @Nullable FunctionReceiverObservationConvention functionReceiverObservationConvention, @Nullable FunctionObservationConvention functionObservationConvention, @Nullable FunctionSenderObservationConvention functionSenderObservationConvention) {
|
||||
public ObservationFunctionAroundWrapper(ObservationRegistry observationRegistry, @Nullable FunctionReceiverObservationConvention functionReceiverObservationConvention,
|
||||
@Nullable FunctionObservationConvention functionObservationConvention, @Nullable FunctionSenderObservationConvention functionSenderObservationConvention) {
|
||||
this.observationRegistry = observationRegistry;
|
||||
this.functionReceiverObservationConvention = functionReceiverObservationConvention;
|
||||
this.functionObservationConvention = functionObservationConvention;
|
||||
@@ -58,19 +59,19 @@ public class ObservationFunctionAroundWrapper extends FunctionAroundWrapper {
|
||||
if (FunctionTypeUtils.isCollectionOfMessage(targetFunction.getOutputType())) {
|
||||
return targetFunction.apply(message); // no instrumentation
|
||||
}
|
||||
return nonReactorStream(message, targetFunction);
|
||||
return nonReactorStream((Message<?>) message, targetFunction);
|
||||
}
|
||||
|
||||
private Object nonReactorStream(Object message,
|
||||
private Object nonReactorStream(Message<?> message,
|
||||
SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) {
|
||||
if (targetFunction.isConsumer()) {
|
||||
Observation observationOfInputMessage = stoppedObservationOfInputMessage(message, targetFunction);
|
||||
Observation consumerObservation = consumerObservation(targetFunction, observationOfInputMessage);
|
||||
Observation consumerObservation = consumerObservation(targetFunction, observationOfInputMessage, message);
|
||||
return consumerObservation.observe(() -> targetFunction.apply(message));
|
||||
}
|
||||
else if (targetFunction.isFunction()) {
|
||||
Observation observationOfInputMessage = stoppedObservationOfInputMessage(message, targetFunction);
|
||||
Observation consumerObservation = consumerObservation(targetFunction, observationOfInputMessage);
|
||||
Observation consumerObservation = consumerObservation(targetFunction, observationOfInputMessage, message);
|
||||
Object outputMessage = consumerObservation.observe(() -> targetFunction.apply(message));
|
||||
if (isNonNullMessageType(outputMessage)) {
|
||||
return outputMessage; // no instrumentation
|
||||
@@ -78,7 +79,7 @@ public class ObservationFunctionAroundWrapper extends FunctionAroundWrapper {
|
||||
return observeOutputMessage(outputMessage, targetFunction, consumerObservation);
|
||||
}
|
||||
else {
|
||||
Object supplierOutputMessage = functionProcessingObservation(targetFunction).observe(targetFunction::get);
|
||||
Object supplierOutputMessage = functionProcessingObservation(targetFunction, message).observe(targetFunction::get);
|
||||
if (isNonNullMessageType(supplierOutputMessage)) {
|
||||
return supplierOutputMessage; // no instrumentation
|
||||
}
|
||||
@@ -86,12 +87,12 @@ public class ObservationFunctionAroundWrapper extends FunctionAroundWrapper {
|
||||
}
|
||||
}
|
||||
|
||||
private Observation functionProcessingObservation(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) {
|
||||
return FunctionObservation.FUNCTION_PROCESSING_OBSERVATION.observation(this.functionObservationConvention, DefaultFunctionObservationConvention.INSTANCE, () -> new FunctionContext(targetFunction), this.observationRegistry);
|
||||
private Observation functionProcessingObservation(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, Message<?> message) {
|
||||
return FunctionObservation.FUNCTION_PROCESSING_OBSERVATION.observation(this.functionObservationConvention, DefaultFunctionObservationConvention.INSTANCE, () -> new FunctionContext(targetFunction, message), this.observationRegistry);
|
||||
}
|
||||
|
||||
private Observation consumerObservation(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, Observation observationOfInputMessage) {
|
||||
return functionProcessingObservation(targetFunction)
|
||||
private Observation consumerObservation(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, Observation observationOfInputMessage, Message<?> message) {
|
||||
return functionProcessingObservation(targetFunction, message)
|
||||
.parentObservation(observationOfInputMessage);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user