Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make the MeterRegistry used by Micrometer configurable

Browse Source 
Resolves #4224
This commit is contained in:
Mahmoud Ben Hassine
2022-11-07 11:16:24 +01:00
parent d889f84735
commit d444a8abaf
9 changed files with 122 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 + "]";

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

@@ -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<B extends JobBuilderHelper<B>> {
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<B extends JobBuilderHelper<B>> {
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<B extends JobBuilderHelper<B>> {
private ObservationRegistry observationRegistry;
private MeterRegistry meterRegistry;
private JobParametersIncrementer jobParametersIncrementer;
private JobParametersValidator jobParametersValidator;
@@ -225,6 +244,7 @@ public abstract class JobBuilderHelper<B extends JobBuilderHelper<B>> {
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<B extends JobBuilderHelper<B>> {
this.observationRegistry = observationRegistry;
}
public MeterRegistry getMeterRegistry() {
return meterRegistry;
}
public void setMeterRegistry(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
}
public String getName() {
return name;
}

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

@@ -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:
*
* <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>
@@ -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);
}
/**

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

@@ -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;
}
}

8
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java
View File

@@ -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<I, O> extends AbstractTaskletStepBuilder<SimpleSt
private boolean readerTransactionalQueue = false;
private MeterRegistry meterRegistry = Metrics.globalRegistry;
/**
* Create a new builder initialized with any properties in the parent. The parent is
* copied, so it can be re-used.
@@ -109,6 +114,7 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
this.processor = parent.processor;
this.itemListeners = parent.itemListeners;
this.readerTransactionalQueue = parent.readerTransactionalQueue;
this.meterRegistry = parent.meterRegistry;
this.transactionManager(parent.getTransactionManager());
}
@@ -159,7 +165,9 @@ public class SimpleStepBuilder<I, O> extends AbstractTaskletStepBuilder<SimpleSt
SimpleChunkProvider<I> chunkProvider = new SimpleChunkProvider<>(getReader(), repeatOperations);
SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<>(getProcessor(), getWriter());
chunkProvider.setListeners(new ArrayList<>(itemListeners));
chunkProvider.setMeterRegistry(this.meterRegistry);
chunkProcessor.setListeners(new ArrayList<>(itemListeners));
chunkProcessor.setMeterRegistry(this.meterRegistry);
ChunkOrientedTasklet<I> tasklet = new ChunkOrientedTasklet<>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!readerTransactionalQueue);
return tasklet;

23
spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java
View File

@@ -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<B extends StepBuilderHelper<B>> {
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<B extends StepBuilderHelper<B>> {
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<B extends StepBuilderHelper<B>> {
private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
private MeterRegistry meterRegistry = Metrics.globalRegistry;
public CommonStepProperties() {
}
@@ -170,6 +184,7 @@ public abstract class StepBuilderHelper<B extends StepBuilderHelper<B>> {
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<B extends StepBuilderHelper<B>> {
this.observationRegistry = observationRegistry;
}
public MeterRegistry getMeterRegistry() {
return meterRegistry;
}
public void setMeterRegistry(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
}
public String getName() {
return name;
}

5
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
View File

@@ -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<I, O> extends SimpleChunkProcessor<I, O
@Override
public O doWithRetry(RetryContext context) throws Exception {
Timer.Sample sample = BatchMetrics.createTimerSample();
Timer.Sample sample = BatchMetrics.createTimerSample(meterRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
O output = null;
try {
@@ -336,7 +335,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
if (!data.scanning()) {
chunkMonitor.setChunkSize(inputs.size());
Timer.Sample sample = BatchMetrics.createTimerSample();
Timer.Sample sample = BatchMetrics.createTimerSample(meterRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
try {
doWrite(outputs);

19
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java
View File

@@ -18,6 +18,8 @@ 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;
@@ -47,6 +49,8 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
private final MulticasterBatchListener<I, O> listener = new MulticasterBatchListener<>();
protected MeterRegistry meterRegistry = Metrics.globalRegistry;
/**
* Default constructor for ease of configuration.
*/
@@ -79,6 +83,15 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, 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<I, O> implements ChunkProcessor<I>, Initializi
* @throws Exception if there is a problem
*/
protected void write(StepContribution contribution, Chunk<I> inputs, Chunk<O> 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<I, O> implements ChunkProcessor<I>, Initializi
for (Chunk<I>.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<I, O> implements ChunkProcessor<I>, 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()),

14
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java
View File

@@ -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<I> implements ChunkProvider<I> {
private final RepeatOperations repeatOperations;
private MeterRegistry meterRegistry = Metrics.globalRegistry;
public SimpleChunkProvider(ItemReader<? extends I> itemReader, RepeatOperations repeatOperations) {
this.itemReader = itemReader;
this.repeatOperations = repeatOperations;
@@ -71,6 +74,15 @@ public class SimpleChunkProvider<I> implements ChunkProvider<I> {
}
}
/**
* 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<I> implements ChunkProvider<I> {
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()),