diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionReceiverObservationConvention.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionReceiverObservationConvention.java deleted file mode 100644 index 6e52b62c1..000000000 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionReceiverObservationConvention.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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.function.observability; - -import io.micrometer.common.KeyValues; - -/** - * Default implementation of {@link FunctionReceiverObservationConvention}. - * - * @author Marcin Grzejszczak - * @since 4.0.0 - */ -public class DefaultFunctionReceiverObservationConvention implements FunctionReceiverObservationConvention { - - /** - * Singleton instance of this convention. - */ - public static final FunctionReceiverObservationConvention INSTANCE = new DefaultFunctionReceiverObservationConvention(); - - @Override - public KeyValues getLowCardinalityKeyValues(FunctionReceiverContext context) { - return KeyValues.of(FunctionObservation.FunctionLowCardinalityTags.FUNCTION_NAME.withValue(context.getTargetFunction().getFunctionDefinition())); - } - - @Override - public String getName() { - return "spring.cloud.function.receive"; - } - - @Override - public String getContextualName(FunctionReceiverContext context) { - return context.getTargetFunction().getFunctionDefinition() + " receive"; - } -} diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionSenderObservationConvention.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionSenderObservationConvention.java deleted file mode 100644 index e30cbd044..000000000 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionSenderObservationConvention.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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.function.observability; - -import io.micrometer.common.KeyValues; - -/** - * Default implementation of {@link FunctionSenderObservationConvention}. - * - * @author Marcin Grzejszczak - * @since 4.0.0 - */ -public class DefaultFunctionSenderObservationConvention implements FunctionSenderObservationConvention { - - /** - * Singleton instance of this convention. - */ - public static final FunctionSenderObservationConvention INSTANCE = new DefaultFunctionSenderObservationConvention(); - - @Override - public KeyValues getLowCardinalityKeyValues(FunctionSenderContext context) { - return KeyValues.of(FunctionObservation.FunctionLowCardinalityTags.FUNCTION_NAME.withValue(context.getTargetFunction().getFunctionDefinition())); - } - - @Override - public String getName() { - return "spring.cloud.function.send"; - } - - @Override - public String getContextualName(FunctionSenderContext context) { - return context.getTargetFunction().getFunctionDefinition() + " send"; - } -} diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionObservation.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionObservation.java index ef2a207c2..a50b38814 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionObservation.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionObservation.java @@ -27,45 +27,6 @@ import io.micrometer.observation.docs.ObservationDocumentation; * @since 4.0.0 */ enum FunctionObservation implements ObservationDocumentation { - /** - * Observation created around receiving a message (via consumer or function). - */ - FUNCTION_CONSUMER_OBSERVATION { - @Override - public Class> getDefaultConvention() { - return DefaultFunctionReceiverObservationConvention.class; - } - - @Override - public KeyName[] getLowCardinalityKeyNames() { - return FunctionLowCardinalityTags.values(); - } - - @Override - public String getPrefix() { - return "spring.cloud.function"; - } - }, - - /** - * Observation created around producing a message (via supplier or function). - */ - FUNCTION_PRODUCER_OBSERVATION { - @Override - public Class> getDefaultConvention() { - return DefaultFunctionSenderObservationConvention.class; - } - - @Override - public KeyName[] getLowCardinalityKeyNames() { - return FunctionLowCardinalityTags.values(); - } - - @Override - public String getPrefix() { - return "spring.cloud.function"; - } - }, /** * Observation created around processing a message (functional bean processing). diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverContext.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverContext.java deleted file mode 100644 index bfa615671..000000000 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverContext.java +++ /dev/null @@ -1,44 +0,0 @@ -/* - * 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.function.observability; - -import io.micrometer.observation.transport.ReceiverContext; - -import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry; -import org.springframework.messaging.Message; - -/** - * {@link ReceiverContext} for receiving messages through functional interfaces. - * - * @author Marcin Grzejszczak - * @since 4.0.0 - */ -public class FunctionReceiverContext extends ReceiverContext> { - - private final SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction; - - public FunctionReceiverContext(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, Message carrier) { - super((message, key) -> (String) message.getHeaders().get(key)); - this.targetFunction = targetFunction; - setCarrier(carrier); - } - - public SimpleFunctionRegistry.FunctionInvocationWrapper getTargetFunction() { - return targetFunction; - } - -} diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverObservationConvention.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverObservationConvention.java deleted file mode 100644 index 865bb1dba..000000000 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverObservationConvention.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * Copyright 2022-2022 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.function.observability; - -import io.micrometer.observation.Observation; -import io.micrometer.observation.ObservationConvention; - -/** - * {@link ObservationConvention} for {@link FunctionReceiverContext}. - * - * @author Marcin Grzejszczak - * @author Oleg Zhurakousky - * @since 4.0.0 - */ -public interface FunctionReceiverObservationConvention extends ObservationConvention { - - @Override - default boolean supportsContext(Observation.Context context) { - return context instanceof FunctionReceiverContext; - } -} diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderContext.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderContext.java deleted file mode 100644 index c3bb6df34..000000000 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderContext.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * 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.function.observability; - -import java.util.Objects; - -import io.micrometer.observation.transport.SenderContext; - -import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry; -import org.springframework.messaging.support.MessageBuilder; - -/** - * {@link SenderContext} for sending messages through functional interfaces. - * - * @author Marcin Grzejszczak - * @since 4.0.0 - */ -public class FunctionSenderContext extends SenderContext> { - - private final SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction; - - public FunctionSenderContext(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, MessageBuilder carrier) { - super((messageBuilder, key, value) -> Objects.requireNonNull(messageBuilder).setHeader(key, value)); - this.targetFunction = targetFunction; - setCarrier(carrier); - } - - public SimpleFunctionRegistry.FunctionInvocationWrapper getTargetFunction() { - return targetFunction; - } - -} diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderObservationConvention.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderObservationConvention.java deleted file mode 100644 index fb78c095d..000000000 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderObservationConvention.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Copyright 2022-2022 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.function.observability; - -import io.micrometer.observation.Observation; -import io.micrometer.observation.ObservationConvention; - -/** - * {@link ObservationConvention} for {@link FunctionSenderContext}. - * - * @author Marcin Grzejszczak - * @author Oleg Zhurakousky - */ -public interface FunctionSenderObservationConvention extends ObservationConvention { - - @Override - default boolean supportsContext(Observation.Context context) { - return context instanceof FunctionSenderContext; - } -} diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationAutoConfiguration.java index c8e1b4f8e..3bd61b75c 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationAutoConfiguration.java @@ -37,10 +37,8 @@ public class ObservationAutoConfiguration { @ConditionalOnMissingBean @ConditionalOnBean(ObservationRegistry.class) public FunctionAroundWrapper observationFunctionAroundWrapper(ObservationRegistry registry, - ObjectProvider functionReceiverObservationConvention, - ObjectProvider functionObservationConvention, - ObjectProvider functionSenderObservationConvention) { - return new ObservationFunctionAroundWrapper(registry, functionReceiverObservationConvention.getIfAvailable(() -> null), - functionObservationConvention.getIfAvailable(() -> null), functionSenderObservationConvention.getIfAvailable(() -> null)); + ObjectProvider functionObservationConvention) { + return new ObservationFunctionAroundWrapper(registry, + functionObservationConvention.getIfAvailable(() -> null)); } } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationFunctionAroundWrapper.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationFunctionAroundWrapper.java index 0d6ce5675..c07b93439 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationFunctionAroundWrapper.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationFunctionAroundWrapper.java @@ -18,15 +18,12 @@ package org.springframework.cloud.function.observability; import io.micrometer.observation.Observation; import io.micrometer.observation.ObservationRegistry; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.cloud.function.context.catalog.FunctionAroundWrapper; import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -import org.springframework.messaging.support.MessageBuilder; /** @@ -35,23 +32,13 @@ import org.springframework.messaging.support.MessageBuilder; * @since 4.0.0 */ public class ObservationFunctionAroundWrapper extends FunctionAroundWrapper { - - private static final Log log = LogFactory.getLog(ObservationFunctionAroundWrapper.class); - private final ObservationRegistry observationRegistry; - private final FunctionReceiverObservationConvention functionReceiverObservationConvention; - private final FunctionObservationConvention functionObservationConvention; - private final FunctionSenderObservationConvention functionSenderObservationConvention; - - public ObservationFunctionAroundWrapper(ObservationRegistry observationRegistry, @Nullable FunctionReceiverObservationConvention functionReceiverObservationConvention, - @Nullable FunctionObservationConvention functionObservationConvention, @Nullable FunctionSenderObservationConvention functionSenderObservationConvention) { + public ObservationFunctionAroundWrapper(ObservationRegistry observationRegistry, @Nullable FunctionObservationConvention functionObservationConvention) { this.observationRegistry = observationRegistry; - this.functionReceiverObservationConvention = functionReceiverObservationConvention; this.functionObservationConvention = functionObservationConvention; - this.functionSenderObservationConvention = functionSenderObservationConvention; } @Override @@ -64,67 +51,13 @@ public class ObservationFunctionAroundWrapper extends FunctionAroundWrapper { private Object nonReactorStream(Message message, SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) { - if (targetFunction.isConsumer()) { - Observation observationOfInputMessage = stoppedObservationOfInputMessage(message, targetFunction); - 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, message); - Object outputMessage = consumerObservation.observe(() -> targetFunction.apply(message)); - if (isNonNullMessageType(outputMessage)) { - return outputMessage; // no instrumentation - } - return observeOutputMessage(outputMessage, targetFunction, consumerObservation); - } - else { - Object supplierOutputMessage = functionProcessingObservation(targetFunction, message).observe(targetFunction::get); - if (isNonNullMessageType(supplierOutputMessage)) { - return supplierOutputMessage; // no instrumentation - } - return observeOutputMessage(supplierOutputMessage, targetFunction, null); + if (targetFunction.isConsumer() || targetFunction.isFunction()) { + return functionProcessingObservation(targetFunction, message).observe(() -> targetFunction.apply(message)); } + return functionProcessingObservation(targetFunction, message).observe(targetFunction::get); } 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, Message message) { - return functionProcessingObservation(targetFunction, message) - .parentObservation(observationOfInputMessage); - } - - private boolean isNonNullMessageType(Object outputMessage) { - return outputMessage == null || !(outputMessage instanceof Message); - } - - /** - * Confirmation of getting of message from broker. - * - * @param message message to process - * @param targetFunction target function - * @return stopped observation - */ - private Observation stoppedObservationOfInputMessage(Object message, - SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) { - Observation consumerObservation = FunctionObservation.FUNCTION_CONSUMER_OBSERVATION.observation(this.functionReceiverObservationConvention, DefaultFunctionReceiverObservationConvention.INSTANCE, () -> new FunctionReceiverContext(targetFunction, (Message) message), this.observationRegistry); - consumerObservation.start().stop(); - return consumerObservation; - } - - /** - * Enriching the output message. - * - * @param message message to process - * @param targetFunction target function - * @return enriched output message - */ - private Message observeOutputMessage(Object message, - SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, @Nullable Observation parentObservation) { - FunctionSenderContext context = new FunctionSenderContext(targetFunction, MessageBuilder.fromMessage((Message) message)); - FunctionObservation.FUNCTION_PRODUCER_OBSERVATION.observation(this.functionSenderObservationConvention, DefaultFunctionSenderObservationConvention.INSTANCE, () -> context, this.observationRegistry).parentObservation(parentObservation).start().stop(); - return context.getCarrier().build(); - } }