diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java index 406afba6c..2d69bc9c6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java @@ -22,6 +22,8 @@ import java.util.List; 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; @@ -90,6 +92,8 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; + private MeterRegistry meterRegistry = Metrics.globalRegistry; + private BatchJobObservationConvention observationConvention = new DefaultBatchJobObservationConvention(); /** @@ -288,8 +292,9 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In 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), @@ -447,6 +452,10 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In this.observationRegistry = observationRegistry; } + public void setMeterRegistry(MeterRegistry meterRegistry) { + this.meterRegistry = meterRegistry; + } + @Override public String toString() { return ClassUtils.getShortName(getClass()) + ": [name=" + name + "]"; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java index 4c190e719..9f200da10 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java @@ -22,6 +22,7 @@ import java.util.LinkedHashSet; 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; @@ -113,6 +114,18 @@ public abstract class JobBuilderHelper> { 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 @@ -188,6 +201,10 @@ public abstract class JobBuilderHelper> { if (observationRegistry != null) { job.setObservationRegistry(observationRegistry); } + MeterRegistry meterRegistry = properties.getMeterRegistry(); + if (meterRegistry != null) { + job.setMeterRegistry(meterRegistry); + } Boolean restartable = properties.getRestartable(); if (restartable != null) { @@ -213,6 +230,8 @@ public abstract class JobBuilderHelper> { private ObservationRegistry observationRegistry; + private MeterRegistry meterRegistry; + private JobParametersIncrementer jobParametersIncrementer; private JobParametersValidator jobParametersValidator; @@ -225,6 +244,7 @@ public abstract class JobBuilderHelper> { 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; @@ -262,6 +282,14 @@ public abstract class JobBuilderHelper> { this.observationRegistry = observationRegistry; } + public MeterRegistry getMeterRegistry() { + return meterRegistry; + } + + public void setMeterRegistry(MeterRegistry meterRegistry) { + this.meterRegistry = meterRegistry; + } + public String getName() { return name; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/observability/BatchMetrics.java b/spring-batch-core/src/main/java/org/springframework/batch/core/observability/BatchMetrics.java index 7ae6845df..f0a22a739 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/observability/BatchMetrics.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/observability/BatchMetrics.java @@ -21,7 +21,7 @@ import java.util.Arrays; 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 @@ import org.springframework.lang.Nullable; * Central class for batch metrics. It provides: * *
    - *
  • the main entry point to interact with Micrometer's {@link Metrics#globalRegistry} - * with common metrics such as {@link Timer} and {@link LongTaskTimer}.
    • + *
  • the main entry point to interact with Micrometer's API to create common metrics + * such as {@link Timer} and {@link LongTaskTimer}.
    • *
  • Some utility methods like calculating durations and formatting them in a human * readable format.
    • *
@@ -59,15 +59,16 @@ public final class 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); } /** @@ -106,23 +107,26 @@ public final class BatchMetrics { /** * 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); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java index 695ce9814..162c1978b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java @@ -20,6 +20,8 @@ import java.time.LocalDateTime; 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; @@ -78,6 +80,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; + private MeterRegistry meterRegistry = Metrics.globalRegistry; + private BatchStepObservationConvention observationConvention = new DefaultBatchStepObservationConvention(); /** @@ -431,4 +435,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw this.observationRegistry = observationRegistry; } + public void setMeterRegistry(MeterRegistry meterRegistry) { + this.meterRegistry = meterRegistry; + } + } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java index ab86a7c15..01b71c934 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java @@ -21,6 +21,9 @@ import java.util.HashSet; import java.util.LinkedHashSet; import java.util.Set; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Metrics; + import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.ItemProcessListener; import org.springframework.batch.core.ItemReadListener; @@ -85,6 +88,8 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder extends AbstractTaskletStepBuilder extends AbstractTaskletStepBuilder chunkProvider = new SimpleChunkProvider<>(getReader(), repeatOperations); SimpleChunkProcessor chunkProcessor = new SimpleChunkProcessor<>(getProcessor(), getWriter()); chunkProvider.setListeners(new ArrayList<>(itemListeners)); + chunkProvider.setMeterRegistry(this.meterRegistry); chunkProcessor.setListeners(new ArrayList<>(itemListeners)); + chunkProcessor.setMeterRegistry(this.meterRegistry); ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet<>(chunkProvider, chunkProcessor); tasklet.setBuffering(!readerTransactionalQueue); return tasklet; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java index a97a9e488..3b0a3bba7 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java @@ -21,6 +21,8 @@ import java.util.HashSet; import java.util.List; import java.util.Set; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Metrics; import io.micrometer.observation.ObservationRegistry; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -73,6 +75,11 @@ public abstract class StepBuilderHelper> { return self(); } + public B meterRegistry(MeterRegistry meterRegistry) { + properties.meterRegistry = meterRegistry; + return self(); + } + public B startLimit(int startLimit) { properties.startLimit = startLimit; return self(); @@ -133,6 +140,11 @@ public abstract class StepBuilderHelper> { step.setObservationRegistry(observationRegistry); } + MeterRegistry meterRegistry = properties.getMeterRegistry(); + if (meterRegistry != null) { + step.setMeterRegistry(meterRegistry); + } + Boolean allowStartIfComplete = properties.allowStartIfComplete; if (allowStartIfComplete != null) { step.setAllowStartIfComplete(allowStartIfComplete); @@ -161,6 +173,8 @@ public abstract class StepBuilderHelper> { private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; + private MeterRegistry meterRegistry = Metrics.globalRegistry; + public CommonStepProperties() { } @@ -170,6 +184,7 @@ public abstract class StepBuilderHelper> { this.allowStartIfComplete = properties.allowStartIfComplete; this.jobRepository = properties.jobRepository; this.observationRegistry = properties.observationRegistry; + this.meterRegistry = properties.meterRegistry; this.stepExecutionListeners = new ArrayList<>(properties.stepExecutionListeners); } @@ -189,6 +204,14 @@ public abstract class StepBuilderHelper> { this.observationRegistry = observationRegistry; } + public MeterRegistry getMeterRegistry() { + return meterRegistry; + } + + public void setMeterRegistry(MeterRegistry meterRegistry) { + this.meterRegistry = meterRegistry; + } + public String getName() { return name; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java index 821400209..0a7727119 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java @@ -17,7 +17,6 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; -import java.util.Collections; import java.util.Iterator; import java.util.List; import java.util.concurrent.atomic.AtomicReference; @@ -220,7 +219,7 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor implements ChunkProcessor, Initializi private final MulticasterBatchListener listener = new MulticasterBatchListener<>(); + protected MeterRegistry meterRegistry = Metrics.globalRegistry; + /** * Default constructor for ease of configuration. */ @@ -79,6 +83,15 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi this.itemWriter = itemWriter; } + /** + * Set the meter registry to use for metrics. + * @param meterRegistry the meter registry + * @since 5.0 + */ + public void setMeterRegistry(MeterRegistry meterRegistry) { + this.meterRegistry = meterRegistry; + } + /** * Check mandatory properties. * @@ -278,7 +291,7 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi * @throws Exception if there is a problem */ protected void write(StepContribution contribution, Chunk inputs, Chunk outputs) throws Exception { - Timer.Sample sample = BatchMetrics.createTimerSample(); + Timer.Sample sample = BatchMetrics.createTimerSample(this.meterRegistry); String status = BatchMetrics.STATUS_SUCCESS; try { doWrite(outputs); @@ -303,7 +316,7 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi for (Chunk.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) { final I item = iterator.next(); O output; - Timer.Sample sample = BatchMetrics.createTimerSample(); + Timer.Sample sample = BatchMetrics.createTimerSample(this.meterRegistry); String status = BatchMetrics.STATUS_SUCCESS; try { output = doProcess(item); @@ -333,7 +346,7 @@ public class SimpleChunkProcessor implements ChunkProcessor, Initializi protected void stopTimer(Timer.Sample sample, StepExecution stepExecution, String metricName, String status, String description) { String fullyQualifiedMetricName = BatchMetrics.METRICS_PREFIX + metricName; - sample.stop(BatchMetrics.createTimer(metricName, description + " duration", + sample.stop(BatchMetrics.createTimer(this.meterRegistry, metricName, description + " duration", Tag.of(fullyQualifiedMetricName + ".job.name", stepExecution.getJobExecution().getJobInstance().getJobName()), Tag.of(fullyQualifiedMetricName + ".step.name", stepExecution.getStepName()), diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java index 8cb63e60c..c1f1343b1 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java @@ -18,6 +18,7 @@ package org.springframework.batch.core.step.item; import java.util.List; +import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Metrics; import io.micrometer.core.instrument.Tag; import io.micrometer.core.instrument.Timer; @@ -55,6 +56,8 @@ public class SimpleChunkProvider implements ChunkProvider { private final RepeatOperations repeatOperations; + private MeterRegistry meterRegistry = Metrics.globalRegistry; + public SimpleChunkProvider(ItemReader itemReader, RepeatOperations repeatOperations) { this.itemReader = itemReader; this.repeatOperations = repeatOperations; @@ -71,6 +74,15 @@ public class SimpleChunkProvider implements ChunkProvider { } } + /** + * Set the meter registry to use for metrics. + * @param meterRegistry the meter registry + * @since 5.0 + */ + public void setMeterRegistry(MeterRegistry meterRegistry) { + this.meterRegistry = meterRegistry; + } + /** * Register a listener for callbacks at the appropriate stages in a process. * @param listener a {@link StepListener} @@ -150,7 +162,7 @@ public class SimpleChunkProvider implements ChunkProvider { private void stopTimer(Timer.Sample sample, StepExecution stepExecution, String status) { String fullyQualifiedMetricName = BatchMetrics.METRICS_PREFIX + "item.read"; - sample.stop(BatchMetrics.createTimer("item.read", "Item reading duration", + sample.stop(BatchMetrics.createTimer(this.meterRegistry, "item.read", "Item reading duration", Tag.of(fullyQualifiedMetricName + ".job.name", stepExecution.getJobExecution().getJobInstance().getJobName()), Tag.of(fullyQualifiedMetricName + ".step.name", stepExecution.getStepName()),