From 6cca32fde0542f7e16c68a051604e63c8bb4cda5 Mon Sep 17 00:00:00 2001 From: Mahmoud Ben Hassine Date: Mon, 17 Feb 2020 14:10:02 +0100 Subject: [PATCH] Fix metrics collection in FaultTolerantChunkProcessor Before this commit, metrics were not collected in a fault-tolerant step. This commit updates the FaultTolerantChunkProcessor to collect metrics. For the record, chunk scanning is not covered for two reasons: 1. When scanning a chunk, there is a single item in each write operation, so it would be incorrect to report a metric called "chunk.write" for a single item. We could argue that it is a singleton chunk, but still.. If we want to time scanned (aka individual) items, we need a more fine grained timer called "scanned.item.write" for example. 2. The end result can be confusing and might distort the overall metrics view in case of errors (because of the noisy metrics of additional transactions for individual items). As a reminder, the goal of the "chunk.write" metric is to give an overview of the write operation time of the whole chunk and not to time each item individually (this could be done using an `ItemWriteListener` if needed). Resolves #3664 --- .../item/FaultTolerantChunkProcessor.java | 18 +++- .../core/step/item/SimpleChunkProcessor.java | 2 +- .../FaultTolerantChunkProcessorTests.java | 4 +- .../sample/metrics/BatchMetricsTests.java | 94 ++++++++++++++----- 4 files changed, 94 insertions(+), 24 deletions(-) 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 b61b1174c..5d2ad0426 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 @@ -1,5 +1,5 @@ /* - * Copyright 2006-2019 the original author or authors. + * Copyright 2006-2020 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. @@ -23,11 +23,15 @@ import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +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.listener.StepListenerFailedException; +import org.springframework.batch.core.metrics.BatchMetrics; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NonSkippableProcessException; import org.springframework.batch.core.step.skip.SkipLimitExceededException; @@ -222,6 +226,8 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor implements ChunkProcessor, Initializi return outputs; } - private void stopTimer(Timer.Sample sample, StepExecution stepExecution, String metricName, String status, String description) { + protected 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()), diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java index 7f5502513..ec854c422 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java @@ -27,6 +27,8 @@ import java.util.List; import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.listener.ItemListenerSupport; @@ -55,7 +57,7 @@ public class FaultTolerantChunkProcessorTests { private FaultTolerantChunkProcessor processor; private StepContribution contribution = new StepExecution("foo", - new JobExecution(0L)).createStepContribution(); + new JobExecution(new JobInstance(0L, "job"), new JobParameters())).createStepContribution(); @Before public void setUp() { diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java index e9fea17f6..efbdbcf50 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java +++ b/spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-2020 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. @@ -37,8 +37,6 @@ 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.core.metrics.BatchMetrics; -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; @@ -51,9 +49,12 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +/** + * @author Mahmoud Ben Hassine + */ public class BatchMetricsTests { - private static final int EXPECTED_SPRING_BATCH_METRICS = 6; + private static final int EXPECTED_SPRING_BATCH_METRICS = 10; @Test public void testCalculateDuration() { @@ -147,6 +148,8 @@ public class BatchMetricsTests { List meters = Metrics.globalRegistry.getMeters(); assertTrue(meters.size() >= EXPECTED_SPRING_BATCH_METRICS); + // Job metrics + try { Metrics.globalRegistry.get("spring.batch.job") .tag("name", "job") @@ -164,6 +167,8 @@ public class BatchMetricsTests { fail("There should be a meter of type LONG_TASK_TIMER named spring.batch.job.active" + " registered in the global registry: " + e.getMessage()); } + + // Step 1 (tasklet) metrics try { Metrics.globalRegistry.get("spring.batch.step") @@ -175,6 +180,8 @@ public class BatchMetricsTests { fail("There should be a meter of type TIMER named spring.batch.step" + " registered in the global registry: " + e.getMessage()); } + + // Step 2 (simple chunk-oriented) metrics try { Metrics.globalRegistry.get("spring.batch.step") @@ -219,6 +226,52 @@ public class BatchMetricsTests { fail("There should be a meter of type TIMER named spring.batch.chunk.write" + " registered in the global registry: " + e.getMessage()); } + + // Step 3 (fault-tolerant chunk-oriented) metrics + + try { + Metrics.globalRegistry.get("spring.batch.step") + .tag("name", "step3") + .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", "step3") + .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", "step3") + .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", "step3") + .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 @@ -240,26 +293,24 @@ public class BatchMetricsTests { .build(); } - @Bean - public ItemReader itemReader() { - return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)); - } - - @Bean - public ItemWriter itemWriter() { - return items -> { - for (Integer item : items) { - System.out.println("item = " + item); - } - }; - } - @Bean public Step step2() { return stepBuilderFactory.get("step2") - .chunk(5) - .reader(itemReader()) - .writer(itemWriter()) + .chunk(2) + .reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5))) + .writer(items -> items.forEach(System.out::println)) + .build(); + } + + @Bean + public Step step3() { + return stepBuilderFactory.get("step3") + .chunk(2) + .reader(new ListItemReader<>(Arrays.asList(6, 7, 8, 9, 10))) + .writer(items -> items.forEach(System.out::println)) + .faultTolerant() + .skip(Exception.class) + .skipLimit(3) .build(); } @@ -268,6 +319,7 @@ public class BatchMetricsTests { return jobBuilderFactory.get("job") .start(step1()) .next(step2()) + .next(step3()) .build(); } }