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

Initial support for batch metrics with Micrometer

Browse Source 
Implements BATCH-2774
This commit is contained in:
Mahmoud Ben Hassine
2019-02-07 16:14:30 +01:00
parent 145bacc34a
commit 57c8d70ff3
11 changed files with 462 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -98,6 +98,7 @@ allprojects {
beanshellVersion = '2.0b5'
jaxbApiVersion = '2.3.1'
jaxbImplVersion = '2.3.0.1'
micrometerVersion = '1.1.4'
docResourcesVersion = '0.1.1.RELEASE'
}
@@ -274,6 +275,7 @@ project('spring-batch-core') {
compile "org.springframework:spring-core:$springVersion"
compile "org.springframework:spring-tx:$springVersion"
compile "javax.batch:javax.batch-api:$javaxBatchApiVersion"
compile "io.micrometer:micrometer-core:$micrometerVersion"
testCompile "org.springframework:spring-test:$springVersion"
testCompile "org.mockito:mockito-core:$mockitoVersion"

14
spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2013 the original author or authors.
* Copyright 2006-2019 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,6 +22,7 @@ import java.io.Serializable;
* they can be applied at a chunk boundary.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
*/
@SuppressWarnings("serial")
@@ -43,11 +44,14 @@ public class StepContribution implements Serializable {
private ExitStatus exitStatus = ExitStatus.EXECUTING;
private volatile StepExecution stepExecution;
/**
* @param execution {@link StepExecution} the stepExecution used to initialize
* {@code skipCount}.
*/
public StepContribution(StepExecution execution) {
this.stepExecution = execution;
this.parentSkipCount = execution.getSkipCount();
}
@@ -191,6 +195,14 @@ public class StepContribution implements Serializable {
return processSkipCount;
}
/**
* Public getter for the parent step execution of this contribution.
* @return parent step execution of this contribution
*/
public StepExecution getStepExecution() {
return stepExecution;
}
/*
* (non-Javadoc)
*

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

@@ -19,6 +19,9 @@ package org.springframework.batch.core.job;
import java.util.Collection;
import java.util.Date;
import io.micrometer.core.instrument.LongTaskTimer;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Timer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
@@ -36,6 +39,7 @@ import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.launch.support.ExitCodeMapper;
import org.springframework.batch.core.listener.CompositeJobExecutionListener;
import org.springframework.batch.core.metrics.BatchMetrics;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.scope.context.JobSynchronizationManager;
@@ -297,7 +301,9 @@ InitializingBean {
}
JobSynchronizationManager.register(execution);
LongTaskTimer longTaskTimer = BatchMetrics.createLongTaskTimer("job.active", "Active jobs");
LongTaskTimer.Sample longTaskTimerSample = longTaskTimer.start();
Timer.Sample timerSample = BatchMetrics.createTimerSample();
try {
jobParametersValidator.validate(execution.getJobParameters());
@@ -353,6 +359,11 @@ InitializingBean {
execution.setExitStatus(exitStatus.and(newExitStatus));
}
timerSample.stop(BatchMetrics.createTimer("job", "Job duration",
Tag.of("name", execution.getJobInstance().getJobName()),
Tag.of("status", execution.getExitStatus().getExitCode())
));
longTaskTimerSample.stop();
execution.setEndTime(new Date());
try {

117
spring-batch-core/src/main/java/org/springframework/batch/core/metrics/BatchMetrics.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* Copyright 2019 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.metrics;
import java.util.Arrays;
import java.util.function.Supplier;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.LongTaskTimer;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Timer;
/**
* Main entry point to interact with Micrometer's {@link Metrics#globalRegistry}.
* Provides common metrics such as {@link Timer}, {@link Counter} and {@link Gauge}.
*
* Only intended for internal use.
*
* @author Mahmoud Ben Hassine
* @since 4.2
*/
public final class BatchMetrics {
private static final String METRICS_PREFIX = "spring.batch.";
public static final String STATUS_SUCCESS = "SUCCESS";
public static final String STATUS_FAILURE = "FAILURE";
private BatchMetrics() {}
/**
* Create a {@link Timer}.
*
* @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) {
return Timer.builder(METRICS_PREFIX + name)
.description(description)
.tags(Arrays.asList(tags))
.register(Metrics.globalRegistry);
}
/**
* Create a new {@link Timer.Sample}.
*
* @return a new timer sample instance
*/
public static Timer.Sample createTimerSample() {
return Timer.start(Metrics.globalRegistry);
}
/**
* Create a new {@link LongTaskTimer}.
*
* @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) {
return LongTaskTimer.builder(METRICS_PREFIX + name)
.description(description)
.tags(Arrays.asList(tags))
.register(Metrics.globalRegistry);
}
/**
* Create a new {@link Counter}.
*
* @param name of the counter. Will be prefixed with {@link BatchMetrics#METRICS_PREFIX}.
* @param description of the counter
* @param tags of the counter
* @return a new counter instance
*/
public static Counter createCounter(String name, String description, Tag... tags) {
return Counter.builder(METRICS_PREFIX + name)
.description(description)
.tags(Arrays.asList(tags))
.register(Metrics.globalRegistry);
}
/**
* Create a new {@link Gauge}.
*
* @param name of the gauge. Will be prefixed with {@link BatchMetrics#METRICS_PREFIX}.
* @param description of the gauge
* @param supplier A supplier that yields a value for the gauge.
* @param tags of the gauge
* @return a new gauge instance
*/
public static Gauge createGauge(String name, String description, Supplier<Number> supplier, Tag... tags) {
return Gauge.builder(METRICS_PREFIX + name, supplier)
.description(description)
.tags(Arrays.asList(tags))
.register(Metrics.globalRegistry);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2018 the original author or authors.
* Copyright 2006-2019 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.
@@ -17,6 +17,8 @@ package org.springframework.batch.core.step;
import java.util.Date;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Timer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
@@ -30,6 +32,7 @@ import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.launch.support.ExitCodeMapper;
import org.springframework.batch.core.listener.CompositeStepExecutionListener;
import org.springframework.batch.core.metrics.BatchMetrics;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.item.ExecutionContext;
@@ -188,6 +191,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
}
stepExecution.setStartTime(new Date());
stepExecution.setStatus(BatchStatus.STARTED);
Timer.Sample sample = BatchMetrics.createTimerSample();
getJobRepository().update(stepExecution);
// Start with a default value that will be trumped by anything
@@ -256,6 +260,11 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
+ "This job is now in an unknown state and should not be restarted.", name, stepExecution.getJobExecution().getJobInstance().getJobName()), e);
}
sample.stop(BatchMetrics.createTimer("step", "Step duration",
Tag.of("job.name", stepExecution.getJobExecution().getJobInstance().getJobName()),
Tag.of("name", stepExecution.getStepName()),
Tag.of("status", stepExecution.getExitStatus().getExitCode())
));
stepExecution.setEndTime(new Date());
stepExecution.setExitStatus(exitStatus);

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

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2018 the original author or authors.
* Copyright 2006-2019 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.
@@ -18,9 +18,14 @@ package org.springframework.batch.core.step.item;
import java.util.List;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Timer;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.MulticasterBatchListener;
import org.springframework.batch.core.metrics.BatchMetrics;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.InitializingBean;
@@ -283,6 +288,8 @@ 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();
String status = BatchMetrics.STATUS_SUCCESS;
try {
doWrite(outputs.getItems());
}
@@ -292,8 +299,12 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
* here, so prevent any more processing of these inputs.
*/
inputs.clear();
status = BatchMetrics.STATUS_FAILURE;
throw e;
}
finally {
stopTimer(sample, contribution.getStepExecution(), "chunk.write", status, "Chunk writing");
}
contribution.incrementWriteCount(outputs.size());
}
@@ -302,6 +313,8 @@ 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();
String status = BatchMetrics.STATUS_SUCCESS;
try {
output = doProcess(item);
}
@@ -311,8 +324,12 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
* here, so prevent any more processing of these inputs.
*/
inputs.clear();
status = BatchMetrics.STATUS_FAILURE;
throw e;
}
finally {
stopTimer(sample, contribution.getStepExecution(), "item.process", status, "Item processing");
}
if (output != null) {
outputs.add(output);
}
@@ -323,4 +340,12 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
return outputs;
}
private void stopTimer(Timer.Sample sample, StepExecution stepExecution, String metricName, String status, String description) {
sample.stop(BatchMetrics.createTimer(metricName, description + " duration",
Tag.of("job.name", stepExecution.getJobExecution().getJobInstance().getJobName()),
Tag.of("step.name", stepExecution.getStepName()),
Tag.of("status", status)
));
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2018 the original author or authors.
* Copyright 2006-2019 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.
@@ -18,11 +18,16 @@ package org.springframework.batch.core.step.item;
import java.util.List;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Timer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.listener.MulticasterBatchListener;
import org.springframework.batch.core.metrics.BatchMetrics;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
@@ -115,14 +120,20 @@ public class SimpleChunkProvider<I> implements ChunkProvider<I> {
@Override
public RepeatStatus doInIteration(final RepeatContext context) throws Exception {
I item = null;
Timer.Sample sample = Timer.start(Metrics.globalRegistry);
String status = BatchMetrics.STATUS_SUCCESS;
try {
item = read(contribution, inputs);
}
catch (SkipOverflowException e) {
// read() tells us about an excess of skips by throwing an
// exception
status = BatchMetrics.STATUS_FAILURE;
return RepeatStatus.FINISHED;
}
finally {
stopTimer(sample, contribution.getStepExecution(), status);
}
if (item == null) {
inputs.setEnd();
return RepeatStatus.FINISHED;
@@ -138,6 +149,14 @@ public class SimpleChunkProvider<I> implements ChunkProvider<I> {
}
private void stopTimer(Timer.Sample sample, StepExecution stepExecution, String status) {
sample.stop(BatchMetrics.createTimer("item.read", "Item reading duration",
Tag.of("job.name", stepExecution.getJobExecution().getJobInstance().getJobName()),
Tag.of("step.name", stepExecution.getStepName()),
Tag.of("status", status)
));
}
@Override
public void postProcess(StepContribution contribution, Chunk<I> chunk) {
// do nothing

2
spring-batch-docs/asciidoc/index-single.adoc
View File

@@ -34,6 +34,8 @@ include::jsr-352.adoc[]
include::spring-batch-integration.adoc[]
include::monitoring-and-metrics.adoc[]
include::appendix.adoc[]
include::schema-appendix.adoc[]

2
spring-batch-docs/asciidoc/index.adoc
View File

@@ -30,6 +30,8 @@ and guidelines.
with Spring Batch.
<<spring-batch-integration.adoc#springBatchIntegration,Spring Batch Integration>> :: Integration
between Spring Batch and Spring Integration projects.
<<monitoring-and-metrics.adoc#monitoring-and-metrics,Monitoring and metrics>> :: Batch jobs
monitoring and metrics
The following appendices are available:

69
spring-batch-docs/asciidoc/monitoring-and-metrics.adoc Normal file
View File

@@ -0,0 +1,69 @@
:batch-asciidoc: ./
:toc: left
:toclevels: 4
[[monitoring-and-metrics]]
== Monitoring and metrics
Since version 4.2, Spring Batch provides support for batch monitoring and metrics
based on link:$$https://micrometer.io/$$[Micrometer]. This section describes
which metrics are provided out-of-the-box and how to contribute custom metrics.
[[built-in-metrics]]
=== Built-in metrics
Metrics collection does not require any specific configuration. All metrics provided
by the framework are registered in
link:$$https://micrometer.io/docs/concepts#_global_registry$$[Micrometer's global registry]
under the `spring.batch.` prefix. The following table explains all metrics in details:
|===============
|__Metric Name__|__Type__|__Description__
|`spring.batch.job`|`TIMER`|Duration of job execution
|`spring.batch.job.active`|`LONG_TASK_TIMER`|Currently active jobs
|`spring.batch.step`|`TIMER`|Duration of step execution
|`spring.batch.item.read`|`TIMER`|Duration of item reading
|`spring.batch.item.process`|`TIMER`|Duration of item processing
|`spring.batch.chunk.write`|`TIMER`|Duration of chunk writing
|===============
[[custom-metrics]]
=== Custom metrics
If you want to use your own metrics in your custom components, we recommend using
Micrometer APIs directly. The following is an example of how to time a `Tasklet`:
[source, java]
----
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Timer;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
public class MyTimedTasklet implements Tasklet {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
Timer.Sample sample = Timer.start(Metrics.globalRegistry);
String status = "success";
try {
// do some work
} catch (Exception e) {
// handle exception
status = "failure";
} finally {
sample.stop(Timer.builder("my.tasklet.timer")
.description("Duration of MyTimedTasklet")
.tag("status", status)
.register(Metrics.globalRegistry));
}
return RepeatStatus.FINISHED;
}
}
----

189
spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java Normal file
View File

@@ -0,0 +1,189 @@
/*
* Copyright 2019 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
*
* http://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.sample.metrics;
import java.util.Arrays;
import java.util.List;
import io.micrometer.core.instrument.Meter;
import io.micrometer.core.instrument.Metrics;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
public class BatchMetricsTests {
@Test
public void testBatchMetrics() throws Exception {
// given
ApplicationContext context = new AnnotationConfigApplicationContext(MyJobConfiguration.class);
JobLauncher jobLauncher = context.getBean(JobLauncher.class);
Job job = context.getBean(Job.class);
// when
JobExecution jobExecution = jobLauncher.run(job, new JobParameters());
// then
assertEquals(ExitStatus.COMPLETED, jobExecution.getExitStatus());
List<Meter> meters = Metrics.globalRegistry.getMeters();
assertThat(meters, Matchers.hasSize(7));
try {
Metrics.globalRegistry.get("spring.batch.job")
.tag("name", "job")
.tag("status", "COMPLETED")
.timer();
} catch (Exception e) {
fail("There should be a meter of type TIMER named spring.batch.job " +
"registered in the global registry: " + e.getMessage());
}
try {
Metrics.globalRegistry.get("spring.batch.job.active")
.longTaskTimer();
} catch (Exception e) {
fail("There should be a meter of type LONG_TASK_TIMER named spring.batch.job.active" +
" registered in the global registry: " + e.getMessage());
}
try {
Metrics.globalRegistry.get("spring.batch.step")
.tag("name", "step1")
.tag("job.name", "job")
.tag("status", "COMPLETED")
.timer();
} catch (Exception e) {
fail("There should be a meter of type TIMER named spring.batch.step" +
" registered in the global registry: " + e.getMessage());
}
try {
Metrics.globalRegistry.get("spring.batch.step")
.tag("name", "step2")
.tag("job.name", "job")
.tag("status", "COMPLETED")
.timer();
} catch (Exception e) {
fail("There should be a meter of type TIMER named spring.batch.step" +
" registered in the global registry: " + e.getMessage());
}
try {
Metrics.globalRegistry.get("spring.batch.item.read")
.tag("job.name", "job")
.tag("step.name", "step2")
.tag("status", "SUCCESS")
.timer();
} catch (Exception e) {
fail("There should be a meter of type TIMER named spring.batch.item.read" +
" registered in the global registry: " + e.getMessage());
}
try {
Metrics.globalRegistry.get("spring.batch.item.process")
.tag("job.name", "job")
.tag("step.name", "step2")
.tag("status", "SUCCESS")
.timer();
} catch (Exception e) {
fail("There should be a meter of type TIMER named spring.batch.item.process" +
" registered in the global registry: " + e.getMessage());
}
try {
Metrics.globalRegistry.get("spring.batch.chunk.write")
.tag("job.name", "job")
.tag("step.name", "step2")
.tag("status", "SUCCESS")
.timer();
} catch (Exception e) {
fail("There should be a meter of type TIMER named spring.batch.chunk.write" +
" registered in the global registry: " + e.getMessage());
}
}
@Configuration
@EnableBatchProcessing
static class MyJobConfiguration {
private JobBuilderFactory jobBuilderFactory;
private StepBuilderFactory stepBuilderFactory;
public MyJobConfiguration(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
this.jobBuilderFactory = jobBuilderFactory;
this.stepBuilderFactory = stepBuilderFactory;
}
@Bean
public Step step1() {
return stepBuilderFactory.get("step1")
.tasklet((contribution, chunkContext) -> RepeatStatus.FINISHED)
.build();
}
@Bean
public ItemReader<Integer> itemReader() {
return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
}
@Bean
public ItemWriter<Integer> itemWriter() {
return items -> {
for (Integer item : items) {
System.out.println("item = " + item);
}
};
}
@Bean
public Step step2() {
return stepBuilderFactory.get("step2")
.<Integer, Integer>chunk(5)
.reader(itemReader())
.writer(itemWriter())
.build();
}
@Bean
public Job job() {
return jobBuilderFactory.get("job")
.start(step1())
.next(step2())
.build();
}
}
}