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

Removes the producer and sender observations

Browse Source
This commit is contained in:
Marcin Grzejszczak
2022-10-21 13:28:12 +02:00
parent deabd62093
commit d627742772
9 changed files with 7 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionReceiverObservationConvention.java
View File

@@ -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";
}
}

48
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/DefaultFunctionSenderObservationConvention.java
View File

@@ -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";
}
}

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

@@ -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<? extends ObservationConvention<? extends Observation.Context>> 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<? extends ObservationConvention<? extends Observation.Context>> 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).

44
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverContext.java
View File

@@ -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<Message<?>> {
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;
}
}

35
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionReceiverObservationConvention.java
View File

@@ -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<FunctionReceiverContext> {
@Override
default boolean supportsContext(Observation.Context context) {
return context instanceof FunctionReceiverContext;
}
}

46
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderContext.java
View File

@@ -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<MessageBuilder<?>> {
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;
}
}

34
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/FunctionSenderObservationConvention.java
View File

@@ -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<FunctionSenderContext> {
@Override
default boolean supportsContext(Observation.Context context) {
return context instanceof FunctionSenderContext;
}
}

8
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationAutoConfiguration.java
View File

@@ -37,10 +37,8 @@ public class ObservationAutoConfiguration {
@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));
ObjectProvider<FunctionObservationConvention> functionObservationConvention) {
return new ObservationFunctionAroundWrapper(registry,
functionObservationConvention.getIfAvailable(() -> null));
}
}

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

@@ -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();
}
}