Make the MeterRegistry used by Micrometer configurable

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchObservabilityBeanPostProcessor.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 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.batch.core.configuration.annotation;
+
+import io.micrometer.observation.ObservationRegistry;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.batch.core.job.AbstractJob;
+import org.springframework.batch.core.step.AbstractStep;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+
+/**
+ * Bean post processor that configures observable batch artifacts (jobs and steps) with
+ * Micrometer's observation registry.
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 5.0
+ */
+public class BatchObservabilityBeanPostProcessor implements BeanFactoryPostProcessor, BeanPostProcessor {
+
+	private static final Log LOGGER = LogFactory.getLog(BatchObservabilityBeanPostProcessor.class);
+
+	private ConfigurableListableBeanFactory beanFactory;
+
+	@Override
+	public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+		try {
+			ObservationRegistry observationRegistry = this.beanFactory.getBean(ObservationRegistry.class);
+			if (bean instanceof AbstractJob) {
+				((AbstractJob) bean).setObservationRegistry(observationRegistry);
+			}
+			if (bean instanceof AbstractStep) {
+				((AbstractStep) bean).setObservationRegistry(observationRegistry);
+			}
+		}
+		catch (NoSuchBeanDefinitionException e) {
+			LOGGER.info("No Micrometer observation registry found, defaulting to ObservationRegistry.NOOP");
+		}
+		return bean;
+	}
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/EnableBatchProcessing.java

@@ -151,7 +151,8 @@
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
-@Import({ BatchRegistrar.class, ScopeConfiguration.class, AutomaticJobRegistrarBeanPostProcessor.class })
+@Import({ BatchRegistrar.class, ScopeConfiguration.class, AutomaticJobRegistrarBeanPostProcessor.class,
+		BatchObservabilityBeanPostProcessor.class })
 public @interface EnableBatchProcessing {
 
 	/**

spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java

@@ -22,8 +22,11 @@
 import java.util.stream.Collectors;
 
 import io.micrometer.core.instrument.LongTaskTimer;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -87,6 +90,10 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In
 
 	private StepHandler stepHandler;
 
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
+
+	private MeterRegistry meterRegistry = Metrics.globalRegistry;
+
 	private BatchJobObservationConvention observationConvention = new DefaultBatchJobObservationConvention();
 
 	/**
@@ -285,11 +292,13 @@ public final void execute(JobExecution execution) {
 
 		JobSynchronizationManager.register(execution);
 		String activeJobMeterName = "job.active";
-		LongTaskTimer longTaskTimer = BatchMetrics.createLongTaskTimer(activeJobMeterName, "Active jobs", Tag.of(
-				BatchMetrics.METRICS_PREFIX + activeJobMeterName + ".name", execution.getJobInstance().getJobName()));
+		LongTaskTimer longTaskTimer = BatchMetrics.createLongTaskTimer(this.meterRegistry, activeJobMeterName,
+				"Active jobs", Tag.of(BatchMetrics.METRICS_PREFIX + activeJobMeterName + ".name",
+						execution.getJobInstance().getJobName()));
 		LongTaskTimer.Sample longTaskTimerSample = longTaskTimer.start();
 		Observation observation = BatchMetrics
-				.createObservation(BatchJobObservation.BATCH_JOB_OBSERVATION.getName(), new BatchJobContext(execution))
+				.createObservation(BatchJobObservation.BATCH_JOB_OBSERVATION.getName(), new BatchJobContext(execution),
+						this.observationRegistry)
 				.contextualName(execution.getJobInstance().getJobName())
 				.observationConvention(this.observationConvention).start();
 		try (Observation.Scope scope = observation.openScope()) {
@@ -439,6 +448,14 @@ public void setObservationConvention(BatchJobObservationConvention observationCo
 		this.observationConvention = observationConvention;
 	}
 
+	public void setObservationRegistry(ObservationRegistry observationRegistry) {
+		this.observationRegistry = observationRegistry;
+	}
+
+	public void setMeterRegistry(MeterRegistry meterRegistry) {
+		this.meterRegistry = meterRegistry;
+	}
+
 	@Override
 	public String toString() {
 		return ClassUtils.getShortName(getClass()) + ": [name=" + name + "]";

spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2020 the original author or authors.
+ * Copyright 2006-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.
@@ -22,8 +22,11 @@
 import java.util.List;
 import java.util.Set;
 
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.observation.ObservationRegistry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecutionListener;
 import org.springframework.batch.core.JobParametersIncrementer;
@@ -99,6 +102,30 @@ public B repository(JobRepository jobRepository) {
 		return result;
 	}
 
+	/**
+	 * Sets the observation registry for the job.
+	 * @param observationRegistry the observation registry (optional)
+	 * @return this to enable fluent chaining
+	 */
+	public B observationRegistry(ObservationRegistry observationRegistry) {
+		properties.observationRegistry = observationRegistry;
+		@SuppressWarnings("unchecked")
+		B result = (B) this;
+		return result;
+	}
+
+	/**
+	 * Sets the meter registry for the job.
+	 * @param meterRegistry the meter registry (optional)
+	 * @return this to enable fluent chaining
+	 */
+	public B meterRegistry(MeterRegistry meterRegistry) {
+		properties.meterRegistry = meterRegistry;
+		@SuppressWarnings("unchecked")
+		B result = (B) this;
+		return result;
+	}
+
 	/**
 	 * Registers objects using the annotation based listener configuration.
 	 * @param listener the object that has a method configured with listener annotation
@@ -170,6 +197,14 @@ protected void enhance(Job target) {
 			if (jobParametersValidator != null) {
 				job.setJobParametersValidator(jobParametersValidator);
 			}
+			ObservationRegistry observationRegistry = properties.getObservationRegistry();
+			if (observationRegistry != null) {
+				job.setObservationRegistry(observationRegistry);
+			}
+			MeterRegistry meterRegistry = properties.getMeterRegistry();
+			if (meterRegistry != null) {
+				job.setMeterRegistry(meterRegistry);
+			}
 
 			Boolean restartable = properties.getRestartable();
 			if (restartable != null) {
@@ -193,6 +228,10 @@ public static class CommonJobProperties {
 
 		private JobRepository jobRepository;
 
+		private ObservationRegistry observationRegistry;
+
+		private MeterRegistry meterRegistry;
+
 		private JobParametersIncrementer jobParametersIncrementer;
 
 		private JobParametersValidator jobParametersValidator;
@@ -204,6 +243,8 @@ public CommonJobProperties(CommonJobProperties properties) {
 			this.name = properties.name;
 			this.restartable = properties.restartable;
 			this.jobRepository = properties.jobRepository;
+			this.observationRegistry = properties.observationRegistry;
+			this.meterRegistry = properties.meterRegistry;
 			this.jobExecutionListeners = new LinkedHashSet<>(properties.jobExecutionListeners);
 			this.jobParametersIncrementer = properties.jobParametersIncrementer;
 			this.jobParametersValidator = properties.jobParametersValidator;
@@ -233,6 +274,22 @@ public void setJobRepository(JobRepository jobRepository) {
 			this.jobRepository = jobRepository;
 		}
 
+		public ObservationRegistry getObservationRegistry() {
+			return observationRegistry;
+		}
+
+		public void setObservationRegistry(ObservationRegistry observationRegistry) {
+			this.observationRegistry = observationRegistry;
+		}
+
+		public MeterRegistry getMeterRegistry() {
+			return meterRegistry;
+		}
+
+		public void setMeterRegistry(MeterRegistry meterRegistry) {
+			this.meterRegistry = meterRegistry;
+		}
+
 		public String getName() {
 			return name;
 		}

spring-batch-core/src/main/java/org/springframework/batch/core/observability/BatchMetrics.java

@@ -21,7 +21,7 @@
 import java.util.concurrent.TimeUnit;
 
 import io.micrometer.core.instrument.LongTaskTimer;
-import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Timer;
 import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler;
@@ -34,8 +34,8 @@
  * Central class for batch metrics. It provides:
  *
  * <ul>
- * <li>the main entry point to interact with Micrometer's {@link Metrics#globalRegistry}
- * with common metrics such as {@link Timer} and {@link LongTaskTimer}.</li>
+ * <li>the main entry point to interact with Micrometer's API to create common metrics
+ * such as {@link Timer} and {@link LongTaskTimer}.</li>
  * <li>Some utility methods like calculating durations and formatting them in a human
  * readable format.</li>
  * </ul>
@@ -54,32 +54,21 @@ public final class BatchMetrics {
 
 	public static final String STATUS_FAILURE = "FAILURE";
 
-	/**
-	 * Global {@link ObservationRegistry}. A {@link DefaultMeterObservationHandler} is
-	 * attached to create a {@link Timer} for every finished {@link Observation}.
-	 */
-	public static ObservationRegistry observationRegistry;
-
-	static {
-		observationRegistry = ObservationRegistry.create();
-		observationRegistry.observationConfig()
-				.observationHandler(new DefaultMeterObservationHandler(Metrics.globalRegistry));
-	}
-
 	private BatchMetrics() {
 	}
 
 	/**
 	 * Create a {@link Timer}.
+	 * @param meterRegistry the meter registry to use
 	 * @param name of the timer. Will be prefixed with
 	 * {@link BatchMetrics#METRICS_PREFIX}.
 	 * @param description of the timer
 	 * @param tags of the timer
 	 * @return a new timer instance
 	 */
-	public static Timer createTimer(String name, String description, Tag... tags) {
+	public static Timer createTimer(MeterRegistry meterRegistry, String name, String description, Tag... tags) {
 		return Timer.builder(METRICS_PREFIX + name).description(description).tags(Arrays.asList(tags))
-				.register(Metrics.globalRegistry);
+				.register(meterRegistry);
 	}
 
 	/**
@@ -94,7 +83,8 @@ public static Timer createTimer(String name, String description, Tag... tags) {
 	 * @return a new observation instance
 	 * @since 5.0
 	 */
-	public static Observation createObservation(String name, BatchJobContext context) {
+	public static Observation createObservation(String name, BatchJobContext context,
+			ObservationRegistry observationRegistry) {
 		return Observation.createNotStarted(name, context, observationRegistry);
 	}
 
@@ -110,29 +100,33 @@ public static Observation createObservation(String name, BatchJobContext context
 	 * @return a new observation instance
 	 * @since 5.0
 	 */
-	public static Observation createObservation(String name, BatchStepContext context) {
+	public static Observation createObservation(String name, BatchStepContext context,
+			ObservationRegistry observationRegistry) {
 		return Observation.createNotStarted(name, context, observationRegistry);
 	}
 
 	/**
 	 * Create a new {@link Timer.Sample}.
+	 * @param meterRegistry the meter registry to use
 	 * @return a new timer sample instance
 	 */
-	public static Timer.Sample createTimerSample() {
-		return Timer.start(Metrics.globalRegistry);
+	public static Timer.Sample createTimerSample(MeterRegistry meterRegistry) {
+		return Timer.start(meterRegistry);
 	}
 
 	/**
 	 * Create a new {@link LongTaskTimer}.
+	 * @param meterRegistry the meter registry to use
 	 * @param name of the long task timer. Will be prefixed with
 	 * {@link BatchMetrics#METRICS_PREFIX}.
 	 * @param description of the long task timer.
 	 * @param tags of the timer
 	 * @return a new long task timer instance
 	 */
-	public static LongTaskTimer createLongTaskTimer(String name, String description, Tag... tags) {
+	public static LongTaskTimer createLongTaskTimer(MeterRegistry meterRegistry, String name, String description,
+			Tag... tags) {
 		return LongTaskTimer.builder(METRICS_PREFIX + name).description(description).tags(Arrays.asList(tags))
-				.register(Metrics.globalRegistry);
+				.register(meterRegistry);
 	}
 
 	/**

spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java

@@ -20,7 +20,10 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
 import io.micrometer.observation.Observation;
+import io.micrometer.observation.ObservationRegistry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -75,6 +78,10 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 
 	private JobRepository jobRepository;
 
+	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
+
+	private MeterRegistry meterRegistry = Metrics.globalRegistry;
+
 	private BatchStepObservationConvention observationConvention = new DefaultBatchStepObservationConvention();
 
 	/**
@@ -203,7 +210,7 @@ public final void execute(StepExecution stepExecution)
 		stepExecution.setStatus(BatchStatus.STARTED);
 		Observation observation = BatchMetrics
 				.createObservation(BatchStepObservation.BATCH_STEP_OBSERVATION.getName(),
-						new BatchStepContext(stepExecution))
+						new BatchStepContext(stepExecution), this.observationRegistry)
 				.contextualName(stepExecution.getStepName()).observationConvention(this.observationConvention).start();
 		getJobRepository().update(stepExecution);
 
@@ -424,4 +431,12 @@ public void setObservationConvention(BatchStepObservationConvention observationC
 		this.observationConvention = observationConvention;
 	}
 
+	public void setObservationRegistry(ObservationRegistry observationRegistry) {
+		this.observationRegistry = observationRegistry;
+	}
+
+	public void setMeterRegistry(MeterRegistry meterRegistry) {
+		this.meterRegistry = meterRegistry;
+	}
+
 }