spring-projectsGH-2198: Spring Observability Initial Commit

Resolves spring-projects#2198

Author: garyrussell

build.gradle

@@ -335,12 +335,14 @@ project ('spring-kafka') {
 		optionalApi 'io.projectreactor:reactor-core'
 		optionalApi 'io.projectreactor.kafka:reactor-kafka'
 		optionalApi 'io.micrometer:micrometer-core'
+		api 'io.micrometer:micrometer-observation'
 		optionalApi 'io.micrometer:micrometer-tracing'
 
 		testImplementation project (':spring-kafka-test')
 		testImplementation 'io.projectreactor:reactor-test'
 		testImplementation "org.mockito:mockito-junit-jupiter:$mockitoVersion"
 		testImplementation "org.hibernate.validator:hibernate-validator:$hibernateValidationVersion"
+		testImplementation 'io.micrometer:micrometer-observation-test'
 		testImplementation 'io.micrometer:micrometer-tracing-bridge-brave'
 		testImplementation 'io.micrometer:micrometer-tracing-test'
 		testImplementation 'io.micrometer:micrometer-tracing-integration-test'

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -48,6 +48,8 @@
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationListener;
@@ -62,13 +64,20 @@
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
+import org.springframework.kafka.support.micrometer.DefaultKafkaTemplateObservationConvention;
+import org.springframework.kafka.support.micrometer.KafkaRecordSenderContext;
+import org.springframework.kafka.support.micrometer.KafkaTemplateObservation;
+import org.springframework.kafka.support.micrometer.KafkaTemplateObservationConvention;
 import org.springframework.kafka.support.micrometer.MicrometerHolder;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
+
 /**
  * A template for executing high-level operations. When used with a
  * {@link DefaultKafkaProducerFactory}, the template is thread-safe. The producer factory
@@ -90,7 +99,7 @@
  */
 @SuppressWarnings("deprecation")
 public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationContextAware, BeanNameAware,
-		ApplicationListener<ContextStoppedEvent>, DisposableBean {
+		ApplicationListener<ContextStoppedEvent>, DisposableBean, SmartInitializingSingleton {
 
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); //NOSONAR
 
@@ -126,11 +135,17 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private ConsumerFactory<K, V> consumerFactory;
 
-	private volatile boolean micrometerEnabled = true;
+	private ProducerInterceptor<K, V> producerInterceptor;
+
+	private boolean micrometerEnabled = true;
 
-	private volatile MicrometerHolder micrometerHolder;
+	private MicrometerHolder micrometerHolder;
 
-	private ProducerInterceptor<K, V> producerInterceptor;
+	private boolean observationEnabled;
+
+	private KafkaTemplateObservationConvention observationConvention;
+
+	private ObservationRegistry observationRegistry;
 
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
@@ -382,6 +397,37 @@ public void setProducerInterceptor(ProducerInterceptor<K, V> producerInterceptor
 		this.producerInterceptor = producerInterceptor;
 	}
 
+	/**
+	 * Set to true to enable observation via Micrometer.
+	 * @param observationEnabled true to enable.
+	 * @since 3.0
+	 * @see #setMicrometerEnabled(boolean)
+	 */
+	public void setObservationEnabled(boolean observationEnabled) {
+		this.observationEnabled = observationEnabled;
+	}
+
+	/**
+	 * Set a custom {@link KafkaTemplateObservationConvention}.
+	 * @param observationConvention the convention.
+	 * @since 3.0
+	 */
+	public void setObservationConvention(KafkaTemplateObservationConvention observationConvention) {
+		this.observationConvention = observationConvention;
+	}
+
+	@Override
+	public void afterSingletonsInstantiated() {
+		if (this.observationEnabled && this.observationRegistry == null && this.applicationContext != null) {
+			ObjectProvider<ObservationRegistry> registry =
+					this.applicationContext.getBeanProvider(ObservationRegistry.class);
+			this.observationRegistry = registry.getIfUnique();
+		}
+		else if (this.micrometerEnabled) {
+			this.micrometerHolder = obtainMicrometerHolder();
+		}
+	}
+
 	@Override
 	public void onApplicationEvent(ContextStoppedEvent event) {
 		if (this.customProducerFactory) {
@@ -412,33 +458,33 @@ public CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long t
 	@Override
 	public CompletableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
-		return doSend(producerRecord);
+		return observeSend(producerRecord);
 	}
 
 	@Override
 	public CompletableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, data);
-		return doSend(producerRecord);
+		return observeSend(producerRecord);
 	}
 
 	@Override
 	public CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, @Nullable V data) {
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
-		return doSend(producerRecord);
+		return observeSend(producerRecord);
 	}
 
 	@Override
 	public CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key,
 			@Nullable V data) {
 
 		ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, timestamp, key, data);
-		return doSend(producerRecord);
+		return observeSend(producerRecord);
 	}
 
 	@Override
 	public CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record) {
 		Assert.notNull(record, "'record' cannot be null");
-		return doSend(record);
+		return observeSend(record);
 	}
 
 	@SuppressWarnings("unchecked")
@@ -451,7 +497,7 @@ public CompletableFuture<SendResult<K, V>> send(Message<?> message) {
 				producerRecord.headers().add(KafkaHeaders.CORRELATION_ID, correlationId);
 			}
 		}
-		return doSend((ProducerRecord<K, V>) producerRecord);
+		return observeSend((ProducerRecord<K, V>) producerRecord);
 	}
 
 
@@ -621,6 +667,18 @@ protected void closeProducer(Producer<K, V> producer, boolean inTx) {
 		}
 	}
 
+	private CompletableFuture<SendResult<K, V>> observeSend(final ProducerRecord<K, V> producerRecord) {
+		Observation observation;
+		if (!this.observationEnabled || this.observationRegistry == null) {
+			observation = Observation.NOOP;
+		}
+		else {
+			observation = KafkaTemplateObservation.TEMPLATE_OBSERVATION.observation(
+					this.observationConvention, DefaultKafkaTemplateObservationConvention.INSTANCE,
+					new KafkaRecordSenderContext(producerRecord, this.beanName), this.observationRegistry);
+		}
+		return observation.observe(() -> doSend(producerRecord));
+	}
 	/**
 	 * Send the producer record.
 	 * @param producerRecord the producer record.
@@ -632,9 +690,6 @@ protected CompletableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V>
 		this.logger.trace(() -> "Sending: " + KafkaUtils.format(producerRecord));
 		final CompletableFuture<SendResult<K, V>> future = new CompletableFuture<>();
 		Object sample = null;
-		if (this.micrometerEnabled && this.micrometerHolder == null) {
-			this.micrometerHolder = obtainMicrometerHolder();
-		}
 		if (this.micrometerHolder != null) {
 			sample = this.micrometerHolder.start();
 		}

spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java

@@ -32,6 +32,7 @@
 import org.springframework.aop.support.AopUtils;
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.kafka.support.TopicPartitionOffset;
+import org.springframework.kafka.support.micrometer.KafkaListenerObservationConvention;
 import org.springframework.lang.Nullable;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.transaction.PlatformTransactionManager;
@@ -262,6 +263,8 @@ public enum EOSMode {
 
 	private boolean micrometerEnabled = true;
 
+	private boolean observationEnabled;
+
 	private Duration consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT;
 
 	private Boolean subBatchPerPartition;
@@ -282,6 +285,8 @@ public enum EOSMode {
 
 	private boolean pauseImmediate;
 
+	private KafkaListenerObservationConvention observationConvention;
+
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -635,13 +640,28 @@ public boolean isMicrometerEnabled() {
 
 	/**
 	 * Set to false to disable the Micrometer listener timers. Default true.
+	 * Disabled when {@link #setObservationEnabled(boolean)} is true.
 	 * @param micrometerEnabled false to disable.
 	 * @since 2.3
 	 */
 	public void setMicrometerEnabled(boolean micrometerEnabled) {
 		this.micrometerEnabled = micrometerEnabled;
 	}
 
+	public boolean isObservationEnabled() {
+		return this.observationEnabled;
+	}
+
+	/**
+	 * Set to true to enable observation via Micrometer.
+	 * @param observationEnabled true to enable.
+	 * @since 3.0
+	 * @see #setMicrometerEnabled(boolean)
+	 */
+	public void setObservationEnabled(boolean observationEnabled) {
+		this.observationEnabled = observationEnabled;
+	}
+
 	/**
 	 * Set additional tags for the Micrometer listener timers.
 	 * @param tags the tags.
@@ -912,6 +932,19 @@ private void adviseListenerIfNeeded() {
 		}
 	}
 
+	public KafkaListenerObservationConvention getObservationConvention() {
+		return this.observationConvention;
+	}
+
+	/**
+	 * Set a custom {@link KafkaListenerObservationConvention}.
+	 * @param observationConvention the convention.
+	 * @since 3.0
+	 */
+	public void setObservationConvention(KafkaListenerObservationConvention observationConvention) {
+		this.observationConvention = observationConvention;
+	}
+
 	@Override
 	public String toString() {
 		return "ContainerProperties ["
@@ -942,7 +975,12 @@ public String toString() {
 				+ "\n stopContainerWhenFenced=" + this.stopContainerWhenFenced
 				+ "\n stopImmediate=" + this.stopImmediate
 				+ "\n asyncAcks=" + this.asyncAcks
-				+ "\n idleBeforeDataMultiplier" + this.idleBeforeDataMultiplier
+				+ "\n idleBeforeDataMultiplier=" + this.idleBeforeDataMultiplier
+				+ "\n micrometerEnabled=" + this.micrometerEnabled
+				+ "\n observationEnabled=" + this.observationEnabled
+				+ (this.observationConvention != null
+						? "\n observationConvention=" + this.observationConvention
+						: "")
 				+ "\n]";
 	}