diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml index 326abe5f9..7e1cff0d3 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/pom.xml @@ -84,6 +84,39 @@ awaitility test + + io.micrometer + context-propagation + optional + + + + io.micrometer + micrometer-tracing-integration-test + test + + + io.opentelemetry + * + + + com.wavefront + * + + + io.zipkin.reporter2 + * + + + io.micrometer + micrometer-tracing-bridge-otel + + + io.micrometer + micrometer-tracing-reporter-wavefront + + + diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index ab08c90b8..d8dca0a04 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -25,6 +25,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Pattern; +import io.micrometer.observation.ObservationRegistry; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -84,6 +85,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Byungjun You * @author Omer Celik + * @author Soby Chacko * @since 4.0 * */ @@ -111,11 +113,14 @@ public class ReactorKafkaBinder private final Map messageProducers = new ConcurrentHashMap<>(); + private final ObservationRegistry observationRegistry; + public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, - KafkaTopicProvisioner provisioner) { + KafkaTopicProvisioner provisioner, @Nullable ObservationRegistry observationRegistry) { super(new String[0], provisioner, null, null); this.configurationProperties = configurationProperties; + this.observationRegistry = observationRegistry; } public void setConsumerConfigCustomizer(ConsumerConfigCustomizer consumerConfigCustomizer) { @@ -194,6 +199,9 @@ public class ReactorKafkaBinder SenderOptions opts = this.senderOptionsCustomizer.apply(producerProperties.getBindingName(), SenderOptions.create(configs)); + if (this.configurationProperties.isEnableObservation() && this.observationRegistry != null) { + opts = opts.withObservation(this.observationRegistry); + } // TODO bean for converter; MCB doesn't use one on the producer side. RecordMessageConverter converter = new MessagingMessageConverter(); AbstractApplicationContext applicationContext = getApplicationContext(); @@ -405,7 +413,7 @@ public class ReactorKafkaBinder @SuppressWarnings("unchecked") SenderRecord sr = SenderRecord.create( (ProducerRecord) converter.fromMessage(message, topic), correlation); - Flux> result = sender.send(Flux.just(sr)); + Flux> result = sender.send(Flux.just(sr)).contextCapture(); result.subscribe(res -> { if (this.results != null) { this.results.send(MessageBuilder.withPayload(res) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java index 3f40f3091..e507b6a34 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binder.reactorkafka; +import io.micrometer.observation.ObservationRegistry; + import org.springframework.beans.factory.ObjectProvider; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; @@ -37,6 +39,7 @@ import org.springframework.context.annotation.Import; * * @author Gary Russell * @author Chris Bono + * @author Soby Chacko */ @Configuration(proxyBeanMethods = false) @ConditionalOnMissingBean(Binder.class) @@ -73,13 +76,15 @@ public class ReactorKafkaBinderConfiguration { @Bean ReactorKafkaBinder reactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, - KafkaTopicProvisioner provisioningProvider, - KafkaExtendedBindingProperties extendedBindingProperties, - ObjectProvider consumerConfigCustomizer, - ObjectProvider producerConfigCustomizer, - ObjectProvider receiverOptionsCustomizers, - ObjectProvider senderOptionsptionsCustomizers) { - ReactorKafkaBinder reactorKafkaBinder = new ReactorKafkaBinder(configurationProperties, provisioningProvider); + KafkaTopicProvisioner provisioningProvider, + KafkaExtendedBindingProperties extendedBindingProperties, + ObjectProvider consumerConfigCustomizer, + ObjectProvider producerConfigCustomizer, + ObjectProvider receiverOptionsCustomizers, + ObjectProvider senderOptionsptionsCustomizers, + ObjectProvider observationRegistryObjectProvider) { + ReactorKafkaBinder reactorKafkaBinder = new ReactorKafkaBinder(configurationProperties, provisioningProvider, + observationRegistryObjectProvider.getIfUnique()); reactorKafkaBinder.setExtendedBindingProperties(extendedBindingProperties); reactorKafkaBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique()); reactorKafkaBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique()); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java new file mode 100644 index 000000000..474325af3 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderObservationTests.java @@ -0,0 +1,160 @@ +/* + * Copyright 2022-2024 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.stream.binder.reactorkafka; + +import java.lang.reflect.Type; +import java.time.Duration; +import java.util.function.Function; +import java.util.stream.Collectors; + +import brave.handler.SpanHandler; +import brave.test.TestSpanHandler; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.observation.contextpropagation.ObservationThreadLocalAccessor; +import io.micrometer.tracing.brave.bridge.BraveFinishedSpan; +import io.micrometer.tracing.test.simple.SpansAssert; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.kafka.receiver.ReceiverRecord; +import reactor.kafka.receiver.observation.KafkaReceiverObservation; +import reactor.kafka.receiver.observation.KafkaRecordReceiverContext; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.autoconfigure.actuate.observability.AutoConfigureObservability; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.function.StreamBridge; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.converter.MessagingMessageConverter; +import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.messaging.Message; +import org.springframework.test.annotation.DirtiesContext; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * @author Artem Bilan + * @author Soby Chacko + * @since 4.2.0 + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { + "spring.kafka.consumer.metadata.max.age.ms=1000", + "spring.cloud.function.definition=receive", + "spring.cloud.stream.function.reactive.uppercase=true", + "spring.cloud.stream.bindings.receive-in-0.group=rkbot-in-group", + "spring.cloud.stream.bindings.receive-in-0.destination=rkbot-in-topic", + "spring.cloud.stream.bindings.receive-out-0.destination=rkbot-out-topic", + "spring.cloud.stream.kafka.binder.enable-observation=true", + "spring.cloud.stream.kafka.binder.brokers=${spring.kafka.bootstrap-servers}", + "management.tracing.sampling.probability=1", + "spring.cloud.stream.kafka.bindings.receive-in-0.consumer.converterBeanName=fullRR" + }) +@DirtiesContext +@AutoConfigureObservability +@EmbeddedKafka(topics = { "rkbot-out-topic" }) +public class ReactorKafkaBinderObservationTests { + + private static final TestSpanHandler SPANS = new TestSpanHandler(); + + @Autowired + StreamBridge streamBridge; + + @Autowired + ObservationRegistry observationRegistry; + + @Autowired + TestConfiguration testConfiguration; + + @Autowired + private EmbeddedKafkaBroker embeddedKafka; + + @Test + void endToEndReactorKafkaBinder1() { + + streamBridge.send("rkbot-in-topic", MessageBuilder.withPayload("data") + .build()); + + await().timeout(Duration.ofSeconds(10)).untilAsserted(() -> assertThat(SPANS.spans()).hasSize(3)); + SpansAssert.assertThat(SPANS.spans().stream().map(BraveFinishedSpan::fromBrave).collect(Collectors.toList())) + .haveSameTraceId(); + } + + @SpringBootConfiguration + @EnableAutoConfiguration(exclude = org.springframework.cloud.function.observability.ObservationAutoConfiguration.class) + public static class TestConfiguration { + + @Bean + SpanHandler testSpanHandler() { + return SPANS; + } + + @Bean + RecordMessageConverter fullRR() { + return new RecordMessageConverter() { + + private final RecordMessageConverter converter = new MessagingMessageConverter(); + + @Override + public Message toMessage(ConsumerRecord record, Acknowledgment acknowledgment, + org.apache.kafka.clients.consumer.Consumer consumer, Type payloadType) { + + return MessageBuilder.withPayload(record).build(); + } + + @Override + public ProducerRecord fromMessage(Message message, String defaultTopic) { + return this.converter.fromMessage(message, defaultTopic); + } + + }; + } + + @Bean + Function>, Flux>> receive(ObservationRegistry observationRegistry) { + return s -> s + .flatMap(record -> { + Observation receiverObservation = + KafkaReceiverObservation.RECEIVER_OBSERVATION.start(null, + KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention.INSTANCE, + () -> + new KafkaRecordReceiverContext( + record, "user.receiver", "localhost:9092"), + observationRegistry); + + return Mono.deferContextual(contextView -> Mono.just(record) + .map(rec -> new String(rec.value()).toLowerCase()) + .map(rec -> MessageBuilder.withPayload(rec).setHeader(IntegrationMessageHeaderAccessor.REACTOR_CONTEXT, contextView).build())) + .doOnTerminate(receiverObservation::stop) + .doOnError(receiverObservation::error) + .contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, receiverObservation)); + }); + } + } + +} + diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java index f42017e6c..e34c7d704 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java @@ -83,7 +83,7 @@ class ReactorKafkaBinderTests { KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> { }); provisioner.setMetadataRetryOperations(new RetryTemplate()); - ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner); + ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null); binder.setApplicationContext(mock(GenericApplicationContext.class)); CountDownLatch latch = new CountDownLatch(2); @@ -148,7 +148,7 @@ class ReactorKafkaBinderTests { KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> { }); provisioner.setMetadataRetryOperations(new RetryTemplate()); - ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner); + ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null); binder.setApplicationContext(mock(GenericApplicationContext.class)); CountDownLatch subscriptionLatch = new CountDownLatch(1); @@ -229,7 +229,7 @@ class ReactorKafkaBinderTests { KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> { }); provisioner.setMetadataRetryOperations(new RetryTemplate()); - ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner); + ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null); binder.setApplicationContext(mock(GenericApplicationContext.class)); CountDownLatch subscriptionLatch = new CountDownLatch(1); @@ -298,7 +298,7 @@ class ReactorKafkaBinderTests { KafkaTopicProvisioner provisioner = new KafkaTopicProvisioner(binderProps, kafkaProperties, prop -> { }); provisioner.setMetadataRetryOperations(new RetryTemplate()); - ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner); + ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner, null); CountDownLatch latch = new CountDownLatch(1); GenericApplicationContext context = new GenericApplicationContext(); context.registerBean("sendResults", FluxMessageChannel.class); diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index d591fddc2..2fa9be83f 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -35,11 +35,13 @@ import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.StreamSupport; +import io.micrometer.context.ContextSnapshotFactory; 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 reactor.util.function.Tuples; import org.springframework.beans.BeansException; @@ -90,6 +92,7 @@ import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.Environment; import org.springframework.core.type.MethodMetadata; import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.AbstractSubscribableChannel; import org.springframework.integration.channel.FluxMessageChannel; @@ -139,6 +142,10 @@ import org.springframework.util.StringUtils; @ConditionalOnBean(FunctionRegistry.class) public class FunctionConfiguration { + private static final boolean isContextPropagationPresent = ClassUtils.isPresent( + "io.micrometer.context.ContextSnapshot", FunctionConfiguration.class.getClassLoader()); + + @SuppressWarnings("rawtypes") @Bean public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, @@ -573,7 +580,19 @@ public class FunctionConfiguration { if (!(message instanceof Message)) { message = MessageBuilder.withPayload(message).build(); } - outputChannel.send((Message) message); + if (isContextPropagationPresent && outputChannel instanceof FluxMessageChannel) { + ContextView reactorContext = StaticMessageHeaderAccessor.getReactorContext((Message) message); + try (AutoCloseable autoCloseable = ContextSnapshotHelper.setContext(reactorContext)) { + outputChannel.send((Message) message); + } + catch (Exception e) { + + } + } + else { + outputChannel.send((Message) message); + } + } }) .doOnError(e -> { @@ -1027,4 +1046,15 @@ public class FunctionConfiguration { } } + + private static final class ContextSnapshotHelper { + + private static final ContextSnapshotFactory CONTEXT_SNAPSHOT_FACTORY = ContextSnapshotFactory.builder().build(); + + static AutoCloseable setContext(ContextView context) { + return CONTEXT_SNAPSHOT_FACTORY.setThreadLocalsFrom(context); + } + + } + } diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index 11ddd7872..45549646b 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/docs/modules/ROOT/nav.adoc @@ -74,6 +74,7 @@ **** xref:kafka/kafka-reactive-binder/pattern.adoc[] **** xref:kafka/kafka-reactive-binder/sender_result.adoc[] **** xref:kafka/kafka-reactive-binder/health_indicator.adoc[] +**** xref:kafka/kafka-reactive-binder/reactive_observability.adoc[] *** Kafka Stream Binder **** xref:kafka/kafka-streams-binder/usage.adoc[] **** xref:kafka/kafka-streams-binder/overview.adoc[] diff --git a/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/reactive_observability.adoc b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/reactive_observability.adoc new file mode 100644 index 000000000..50ec81c26 --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-reactive-binder/reactive_observability.adoc @@ -0,0 +1,81 @@ +[[reactive-kafka-binder-observability]] += Observability in Reactive Kafka Binder + +This section describes how Micrometer-based observability is enabled in the reactive Kafka binder. + +== Producer Binding + +There is built-in support for observability in producer binding. +To enable it, set the following property: + +``` +spring.cloud.stream.kafka.binder.enable-observation +``` + +When this property is set to `true`, you can observe the publishing of records. +Both publishing records using `StreamBridge` and regular `Supplier` beans can be observed. + +== Consumer Binding + +Enabling observability on the consumer side is more complex than on the producer side. +There are two starting points for consumer binding: + +1. A topic where data is published via a producer binding +2. A topic where data is produced outside of Spring Cloud Stream + +In the first case, the application ideally wants to carry the observability headers down to the consumer inbound. +In the second case, if there was no upstream observation started, it will start a new observation. + +=== Example: Function with Observability + +``` +@Bean +Function>, Flux>> receive(ObservationRegistry observationRegistry) { + + return s -> s.flatMap(record -> { + Observation receiverObservation = KafkaReceiverObservation.RECEIVER_OBSERVATION.start( + null, + KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention.INSTANCE, + () -> new KafkaRecordReceiverContext(record, "user.receiver", "localhost:9092"), + observationRegistry + ); + + return Mono.deferContextual(contextView -> Mono.just(record) + .map(rec -> new String(rec.value()).toLowerCase()) + .map(rec -> MessageBuilder.withPayload(rec) + .setHeader(IntegrationMessageHeaderAccessor.REACTOR_CONTEXT, contextView) + .build())) + .doOnTerminate(receiverObservation::stop) + .doOnError(receiverObservation::error) + .contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, receiverObservation)); + }); +} +``` + +In this example: + +1. When a record is received, an observation is created. +2. If there's an upstream observation, it will be part of the `KafkaRecordReceiverContext`. +3. A `Mono` is created with context deferred. +4. When the `map` operation is invoked, the context has access to the correct observation. +5. The result of the `flatMap` operation is sent back to the binding as `Flux>`. +6. The outbound record will have the same observability headers from the input binding. + +=== Example: Consumer with Observability + +``` +@Bean +Consumer>> receive(ObservationRegistry observationRegistry, @Value("${spring.kafka.bootstrap-servers}") String bootstrap) { + return f -> f.doOnNext(record -> KafkaReceiverObservation.RECEIVER_OBSERVATION.observation( + null, + KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention.INSTANCE, + () -> new KafkaRecordReceiverContext(record, "user.receiver", bootstrap), + observationRegistry).observe(() -> System.out.println(record))) + .subscribe(); +} +``` + +In this case: + +1. Since there's no output binding, `doOnNext` is used on the `Flux` instead of `flatMap`. +2. The direct call to `observe` starts the observation and properly shuts it down when finished.