Add Observations for send/receive

* Observes sends on PulsarTemplate
* Observes receives on PulsarListener
* Adds auto-generated adocs

Closes #29

spring-pulsar-dependencies/build.gradle

@@ -17,6 +17,8 @@ ext {
 	jaywayJsonPathVersion = '2.6.0'
 	junitJupiterVersion = '5.9.0'
 	log4jVersion = '2.18.0'
+	micrometerVersion = '1.10.0-SNAPSHOT'
+	micrometerTracingVersion = '1.0.0-SNAPSHOT'
 	mockitoVersion = '4.6.1'
 	protobufJavaVersion = '3.21.5'
 	pulsarTestcontainersVersion = '1.17.3'
@@ -35,6 +37,8 @@ dependencies {
 	api platform("org.junit:junit-bom:$junitJupiterVersion")
 	api platform("org.mockito:mockito-bom:$mockitoVersion")
 	api platform("org.springframework:spring-framework-bom:$springVersion")
+	api platform("io.micrometer:micrometer-bom:$micrometerVersion")
+	api platform("io.micrometer:micrometer-tracing-bom:$micrometerTracingVersion")
 
 	constraints {
 		api "com.github.ben-manes.caffeine:caffeine:$caffeineVersion"

spring-pulsar-docs/build.gradle

@@ -5,14 +5,21 @@ plugins {
 
 description = 'Spring Pulsar Docs'
 
+ext {
+	micrometerDocsVersion="1.0.0-SNAPSHOT"
+}
+
 configurations {
 	configurationProperties
+	observationDocs
 }
 
 dependencies {
 	api project (':spring-pulsar')
 	api 'org.springframework.boot:spring-boot-starter'
 	configurationProperties(project(path: ":spring-pulsar-spring-boot-autoconfigure", configuration: "configurationPropertiesMetadata"))
+	observationDocs "io.micrometer:micrometer-docs-generator-spans:$micrometerDocsVersion"
+	observationDocs "io.micrometer:micrometer-docs-generator-metrics:$micrometerDocsVersion"
 }
 
 task aggregatedJavadoc(type: Javadoc) {
@@ -53,6 +60,29 @@ task documentConfigurationProperties(type: org.springframework.pulsar.gradle.doc
 	outputDir = file("${buildDir}/docs/generated/")
 }
 
+def observationsInputDir = file("${rootDir}/spring-pulsar/src/main/java/org/springframework/pulsar/observation").absolutePath
+def observationsOutputDir = file("${buildDir}/docs/generated/observation/").absolutePath
+
+task generateObservabilityMetricsDocs(type: JavaExec) {
+	mainClass = 'io.micrometer.docs.metrics.DocsFromSources'
+	inputs.dir(observationsInputDir)
+	outputs.dir(observationsOutputDir)
+	classpath configurations.observationDocs
+	args observationsInputDir, '.*', observationsOutputDir
+}
+
+task generateObservabilitySpansDocs(type: JavaExec) {
+	mainClass = 'io.micrometer.docs.spans.DocsFromSources'
+	inputs.dir(observationsInputDir)
+	outputs.dir(observationsOutputDir)
+	classpath configurations.observationDocs
+	args observationsInputDir, '.*', observationsOutputDir
+}
+
+task generateObservabilityDocs {
+	dependsOn generateObservabilityMetricsDocs, generateObservabilitySpansDocs
+}
+
 tasks.withType(org.asciidoctor.gradle.jvm.AbstractAsciidoctorTask) {
 	asciidoctorj {
 		fatalWarnings = ['^((?!successfully validated).)*$']
@@ -81,7 +111,7 @@ task asciidoctorMultipage(type: org.asciidoctor.gradle.jvm.AsciidoctorTask) {
 }
 
 syncDocumentationSourceForAsciidoctor {
-	dependsOn documentConfigurationProperties
+	dependsOn documentConfigurationProperties, generateObservabilityDocs
 	from("${buildDir}/docs/generated") {
 		into "asciidoc"
 	}
@@ -91,7 +121,7 @@ syncDocumentationSourceForAsciidoctor {
 }
 
 syncDocumentationSourceForAsciidoctorMultipage {
-	dependsOn documentConfigurationProperties
+	dependsOn documentConfigurationProperties, generateObservabilityDocs
 	from("${buildDir}/docs/generated") {
 		into "asciidoc"
 	}
@@ -101,7 +131,7 @@ syncDocumentationSourceForAsciidoctorMultipage {
 }
 
 syncDocumentationSourceForAsciidoctorPdf {
-	dependsOn documentConfigurationProperties
+	dependsOn documentConfigurationProperties, generateObservabilityDocs
 	from("${buildDir}/docs/generated") {
 		into "asciidoc"
 	}

spring-pulsar-docs/src/main/asciidoc/pulsar.adoc

@@ -1222,6 +1222,27 @@ PulsarTopic partitionedTopic {
 ----
 ====
 
+[[micrometer]]
+=== Observability
+
+[[observation]]
+==== Micrometer Observation
+The `PulsarTemplate` and `PulsarListener` are instrumented with the Micrometer observations API.
+When enabled, send and receive operations are traced and timed.
+
+To enable, set `observationEnabled` on each component.
+
+===== Custom tags
+The default implementation adds the `bean.name` tag for template observations and `listener.id` tag for listener observations.
+To add other tags to timers/traces, configure a custom `PulsarTemplateObservationConvention` or `PulsarListenerObservationConvention` to the template or listener container, respectively.
+
+TIP: You can either subclass `DefaultPulsarTemplateObservationConvention` or `DefaultPulsarListenerObservationConvention` or provide completely new implementations.
+
+include::observation/_metrics.adoc[leveloffset=+2]
+
+include::observation/_spans.adoc[leveloffset=+2]
+
+Refer to https://micrometer.io/docs/tracing[Micrometer Tracing] for more information.
 
 ==== Appendix
 The reference documentation has the following appendices:

spring-pulsar/build.gradle

@@ -7,6 +7,7 @@ description = 'Spring Pulsar Support'
 dependencies {
 	api 'com.github.ben-manes.caffeine:caffeine'
 	api 'com.google.protobuf:protobuf-java'
+	api 'io.micrometer:micrometer-observation'
 	api 'org.apache.pulsar:pulsar-client-all'
 	api 'org.springframework:spring-context'
 	api 'org.springframework:spring-messaging'
@@ -26,6 +27,10 @@ dependencies {
 	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 	testRuntimeOnly 'org.apache.logging.log4j:log4j-core'
 	testRuntimeOnly 'org.apache.logging.log4j:log4j-jcl'
+	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'
 	testImplementation 'org.assertj:assertj-core'
 	testImplementation 'org.awaitility:awaitility'
 	testImplementation 'org.hamcrest:hamcrest'

spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java

@@ -118,7 +118,7 @@ public void afterPropertiesSet() {
 	@Override
 	public C createListenerContainer(PulsarListenerEndpoint endpoint) {
 		C instance = createContainerInstance(endpoint);
-		JavaUtils.INSTANCE.acceptIfNotNull(endpoint.getSubscriptionName(), instance::setBeanName);
+		JavaUtils.INSTANCE.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
 		if (endpoint instanceof AbstractPulsarListenerEndpoint) {
 			configureEndpoint((AbstractPulsarListenerEndpoint<C>) endpoint);
 		}
@@ -171,6 +171,8 @@ else if (this.autoStartup != null) {
 		instanceProperties.setMaxNumMessages(this.containerProperties.getMaxNumMessages());
 		instanceProperties.setMaxNumBytes(this.containerProperties.getMaxNumBytes());
 		instanceProperties.setBatchTimeoutMillis(this.containerProperties.getBatchTimeoutMillis());
+		instanceProperties.setObservationEnabled(this.containerProperties.isObservationEnabled());
+		instanceProperties.setObservationConvention(this.containerProperties.getObservationConvention());
 
 		JavaUtils.INSTANCE.acceptIfNotNull(this.phase, instance::setPhase)
 				.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)

spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java

@@ -73,7 +73,6 @@ protected ConcurrentPulsarMessageListenerContainer<T> createContainerInstance(Pu
 		}
 
 		properties.setSchemaType(endpoint.getSchemaType());
-
 		return new ConcurrentPulsarMessageListenerContainer<T>(getPulsarConsumerFactory(), properties);
 	}
 

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java

@@ -29,7 +29,19 @@
 import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
 
+import org.springframework.beans.factory.BeanNameAware;
+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.core.log.LogAccessor;
+import org.springframework.pulsar.observation.DefaultPulsarTemplateObservationConvention;
+import org.springframework.pulsar.observation.PulsarMessageSenderContext;
+import org.springframework.pulsar.observation.PulsarTemplateObservation;
+import org.springframework.pulsar.observation.PulsarTemplateObservationConvention;
+
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
 
 /**
  * A thread-safe template for executing high-level Pulsar operations.
@@ -39,16 +51,27 @@
  * @author Chris Bono
  * @author Alexander Preuß
  */
-public class PulsarTemplate<T> implements PulsarOperations<T> {
+public class PulsarTemplate<T>
+		implements PulsarOperations<T>, ApplicationContextAware, BeanNameAware, SmartInitializingSingleton {
 
 	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
 
 	private final PulsarProducerFactory<T> producerFactory;
 
 	private final List<ProducerInterceptor> interceptors;
 
+	private ApplicationContext applicationContext;
+
+	private String beanName;
+
 	private Schema<T> schema;
 
+	private boolean observationEnabled;
+
+	private PulsarTemplateObservationConvention observationConvention;
+
+	private ObservationRegistry observationRegistry;
+
 	/**
 	 * Construct a template instance.
 	 * @param producerFactory the factory used to create the backing Pulsar producers.
@@ -92,14 +115,52 @@ public SendMessageBuilder<T> newMessage(T message) {
 		return new SendMessageBuilderImpl<>(this, message);
 	}
 
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) {
+		this.applicationContext = applicationContext;
+	}
+
+	@Override
+	public void setBeanName(String beanName) {
+		this.beanName = beanName;
+	}
+
 	/**
-	 * Setter for schema.
+	 * Set the schema to use on this template.
 	 * @param schema provides the {@link Schema} used on this template
 	 */
 	public void setSchema(Schema<T> schema) {
 		this.schema = schema;
 	}
 
+	/**
+	 * Set to true to enable observation via Micrometer.
+	 * @param observationEnabled true to enable.
+	 */
+	public void setObservationEnabled(boolean observationEnabled) {
+		this.observationEnabled = observationEnabled;
+	}
+
+	/**
+	 * Set a custom observation convention.
+	 * @param observationConvention the convention.
+	 */
+	public void setObservationConvention(PulsarTemplateObservationConvention observationConvention) {
+		this.observationConvention = observationConvention;
+	}
+
+	@Override
+	public void afterSingletonsInstantiated() {
+		// TODO is this how we want to do this? What about SBAC?
+		// TODO when would AC be null? Should we assert or at least log the fact if it
+		// happens?
+		if (this.observationEnabled && this.observationRegistry == null && this.applicationContext != null) {
+			ObjectProvider<ObservationRegistry> registry = this.applicationContext
+					.getBeanProvider(ObservationRegistry.class);
+			this.observationRegistry = registry.getIfUnique();
+		}
+	}
+
 	private MessageId doSend(String topic, T message, TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
 			MessageRouter messageRouter, ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
 		try {
@@ -115,22 +176,49 @@ private CompletableFuture<MessageId> doSendAsync(String topic, T message,
 			ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
 		final String topicName = ProducerUtils.resolveTopicName(topic, this.producerFactory);
 		this.logger.trace(() -> String.format("Sending msg to '%s' topic", topicName));
-		final Producer<T> producer = prepareProducerForSend(topic, message, messageRouter, producerCustomizer);
-		TypedMessageBuilder<T> messageBuilder = producer.newMessage().value(message);
-		if (typedMessageBuilderCustomizer != null) {
-			typedMessageBuilderCustomizer.customize(messageBuilder);
-		}
-		return messageBuilder.sendAsync().whenComplete((msgId, ex) -> {
-			if (ex == null) {
-				this.logger.trace(() -> String.format("Sent msg to '%s' topic", topicName));
-				// TODO success metrics
-			}
-			else {
-				this.logger.error(ex, () -> String.format("Failed to send msg to '%s' topic", topicName));
-				// TODO fail metrics
+
+		PulsarMessageSenderContext senderContext = PulsarMessageSenderContext.newContext(topicName, this.beanName);
+		Observation observation = newObservation(senderContext);
+		try {
+			observation.start();
+			final Producer<T> producer = prepareProducerForSend(topic, message, messageRouter, producerCustomizer);
+			TypedMessageBuilder<T> messageBuilder = producer.newMessage().value(message);
+			if (typedMessageBuilderCustomizer != null) {
+				typedMessageBuilderCustomizer.customize(messageBuilder);
 			}
-			ProducerUtils.closeProducerAsync(producer, this.logger);
-		});
+			senderContext.properties().forEach(messageBuilder::property); // propagate
+																			// props to
+																			// message
+			return messageBuilder.sendAsync().whenComplete((msgId, ex) -> {
+				if (ex == null) {
+					this.logger.trace(() -> String.format("Sent msg to '%s' topic", topicName));
+					observation.stop();
+				}
+				else {
+					this.logger.error(ex, () -> String.format("Failed to send msg to '%s' topic", topicName));
+					observation.error(ex);
+					observation.stop();
+				}
+				ProducerUtils.closeProducerAsync(producer, this.logger);
+			});
+		}
+		catch (RuntimeException ex) {
+			observation.error(ex);
+			observation.stop();
+			throw ex;
+		}
+	}
+
+	private Observation newObservation(PulsarMessageSenderContext senderContext) {
+		Observation observation;
+		if (!this.observationEnabled || this.observationRegistry == null) {
+			observation = Observation.NOOP;
+		}
+		else {
+			observation = PulsarTemplateObservation.TEMPLATE_OBSERVATION.observation(this.observationConvention,
+					DefaultPulsarTemplateObservationConvention.INSTANCE, senderContext, this.observationRegistry);
+		}
+		return observation;
 	}
 
 	private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter,

spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java

@@ -111,7 +111,7 @@ public void setBeanName(String name) {
 	 */
 	@Nullable
 	public String getBeanName() {
-		return this.beanName;
+		return this.beanName; // the container factory sets this to the listener id
 	}
 
 	@Override