From a4803542105d5a185a1fed89db6fec3691c90edb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 26 Jun 2015 15:13:34 -0400 Subject: [PATCH] INT-3752: Add AggregatingMetricsFactory JIRA: https://jira.spring.io/browse/INT-3752 JIRA: https://jira.spring.io/browse/INT-3754 The statistics are based on this count so, for example the `meanSendDuration` reflects the mean of total duration from the first send to the last in each batch. Also change the linked lists to `Deque`s. Also remove `isTraceEnabled()` calls - too expensive in high volume environments. Polishing Assign the count to a local variable so the mod (%) operation is short circuited rather than short circuiting the call to isFullStatsEnabled(). INT-3752: Fix Aggregating Metrics Previously, the duration was for the first message in each sample. We need to capture the total elapsed time for the sample and use it for the duration calculation. Add 'newCount' to the context so the after... method can calculate the duration at the end of the sample. The start field does not need to be volatile because its updates are synchronized. INT-3752: Fix Javadocs --- .../AggregatingMessageChannelMetrics.java | 111 ++++++++++++++++++ .../DefaultMessageChannelMetrics.java | 34 ++---- .../handler/AbstractMessageHandler.java | 2 +- .../AbstractMessageHandlerMetrics.java | 6 +- .../AggregatingMessageHandlerMetrics.java | 99 ++++++++++++++++ .../DefaultMessageHandlerMetrics.java | 20 ++-- .../management/ExponentialMovingAverage.java | 7 +- .../ExponentialMovingAverageRate.java | 13 +- .../ExponentialMovingAverageRatio.java | 17 ++- .../ExponentialMovingAverageTests.java | 9 ++ .../monitor/AggregatingMetricsFactory.java | 53 +++++++++ .../AggregatingMetricsTests-context.xml | 32 +++++ .../monitor/AggregatingMetricsTests.java | 107 +++++++++++++++++ src/reference/asciidoc/jmx.adoc | 23 ++++ 14 files changed, 472 insertions(+), 61 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java create mode 100644 spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java create mode 100644 spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml create mode 100644 spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java new file mode 100644 index 0000000000..ad9cac5a42 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AggregatingMessageChannelMetrics.java @@ -0,0 +1,111 @@ +/* + * Copyright 2009-2015 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.integration.channel.management; + +import org.springframework.integration.support.management.ExponentialMovingAverage; +import org.springframework.integration.support.management.ExponentialMovingAverageRate; +import org.springframework.integration.support.management.ExponentialMovingAverageRatio; +import org.springframework.integration.support.management.MetricsContext; + +/** + * An implementation of {@link MessageChannelMetrics} that aggregates the total response + * time over a sample, to avoid fetching the system time twice for every message. + * + * @author Gary Russell + * @since 4.2 + */ +public class AggregatingMessageChannelMetrics extends DefaultMessageChannelMetrics { + + private static final int DEFAULT_SAMPLE_SIZE = 1000; + + private final int sampleSize; + + private long start; + + public AggregatingMessageChannelMetrics() { + this(null, DEFAULT_SAMPLE_SIZE); + } + + /** + * Construct an instance with default metrics with {@code window=10, period=1 second, + * lapsePeriod=1 minute}. + * @param name the name. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageChannelMetrics(String name, int sampleSize) { + super(name); + this.sampleSize = sampleSize; + } + + /** + * Construct an instance with the supplied metrics. For proper representation of metrics, the + * supplied sendDuration must have a {@code factor=1000000.} and the the other arguments + * must be created with the {@code millis} constructor argument set to true. + * @param name the name. + * @param sendDuration an {@link ExponentialMovingAverage} for calculating the send duration. + * @param sendErrorRate an {@link ExponentialMovingAverageRate} for calculating the send error rate. + * @param sendSuccessRatio an {@link ExponentialMovingAverageRatio} for calculating the success ratio. + * @param sendRate an {@link ExponentialMovingAverageRate} for calculating the send rate. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageChannelMetrics(String name, ExponentialMovingAverage sendDuration, + ExponentialMovingAverageRate sendErrorRate, ExponentialMovingAverageRatio sendSuccessRatio, + ExponentialMovingAverageRate sendRate, int sampleSize) { + super(name, sendDuration, sendErrorRate, sendSuccessRatio, sendRate); + this.sampleSize = sampleSize; + } + + @Override + public synchronized MetricsContext beforeSend() { + long count = this.sendCount.getAndIncrement(); + if (isFullStatsEnabled() && count % this.sampleSize == 0) { + this.start = System.nanoTime(); + this.sendRate.increment(start); + } + return new AggregatingChannelMetricsContext(this.start, count + 1); + } + + @Override + public void afterSend(MetricsContext context, boolean result) { + AggregatingChannelMetricsContext aggregatingContext = (AggregatingChannelMetricsContext) context; + long newCount = aggregatingContext.newCount; + if (result) { + if (isFullStatsEnabled() && newCount % this.sampleSize == 0) { + long now = System.nanoTime(); + this.sendSuccessRatio.success(now); + this.sendDuration.append(now - aggregatingContext.start); + } + } + else { + if (isFullStatsEnabled() && newCount % this.sampleSize == 0) { + long now = System.nanoTime(); + this.sendSuccessRatio.failure(now); + this.sendErrorRate.increment(now); + } + this.sendErrorCount.incrementAndGet(); + } + } + + protected static class AggregatingChannelMetricsContext extends DefaultChannelMetricsContext { + + protected long newCount; + + public AggregatingChannelMetricsContext(long start, long newCount) { + super(start); + this.newCount = newCount; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java index 1124dd367c..4d20542080 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java @@ -15,9 +15,6 @@ package org.springframework.integration.channel.management; import java.util.concurrent.atomic.AtomicLong; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.integration.support.management.ExponentialMovingAverage; import org.springframework.integration.support.management.ExponentialMovingAverageRate; import org.springframework.integration.support.management.ExponentialMovingAverageRatio; @@ -34,25 +31,23 @@ import org.springframework.integration.support.management.Statistics; */ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics { - protected final Log logger = LogFactory.getLog(getClass()); - public static final long ONE_SECOND_SECONDS = 1; public static final long ONE_MINUTE_SECONDS = 60; public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; - private final ExponentialMovingAverage sendDuration; + protected final ExponentialMovingAverage sendDuration; - private final ExponentialMovingAverageRate sendErrorRate; + protected final ExponentialMovingAverageRate sendErrorRate; - private final ExponentialMovingAverageRatio sendSuccessRatio; + protected final ExponentialMovingAverageRatio sendSuccessRatio; - private final ExponentialMovingAverageRate sendRate; + protected final ExponentialMovingAverageRate sendRate; - private final AtomicLong sendCount = new AtomicLong(); + protected final AtomicLong sendCount = new AtomicLong(); - private final AtomicLong sendErrorCount = new AtomicLong(); + protected final AtomicLong sendErrorCount = new AtomicLong(); protected final AtomicLong receiveCount = new AtomicLong(); @@ -106,10 +101,6 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics @Override public MetricsContext beforeSend() { - if (logger.isTraceEnabled()) { - logger.trace("Recording send on channel(" + this.name + ")"); - } - long start = 0; if (isFullStatsEnabled()) { start = System.nanoTime(); @@ -134,10 +125,6 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics } this.sendErrorCount.incrementAndGet(); } - if (logger.isTraceEnabled()) { - logger.trace("Elapsed: " + this.name - + ": " + ((System.nanoTime() / 1000000. - ((DefaultChannelMetricsContext) context).start)) + "ms"); - } } @Override @@ -229,9 +216,6 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics @Override public void afterReceive() { - if (logger.isTraceEnabled()) { - logger.trace("Recording receive on channel(" + this.name + ") "); - } this.receiveCount.incrementAndGet(); } @@ -267,11 +251,11 @@ public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics + "]", name, sendCount.get()); } - private static class DefaultChannelMetricsContext implements MetricsContext { + protected static class DefaultChannelMetricsContext implements MetricsContext { - private final long start; + protected final long start; - public DefaultChannelMetricsContext(long start) { + protected DefaultChannelMetricsContext(long start) { this.start = start; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index ef2efffabd..13c1c30592 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -110,7 +110,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); } if (countsEnabled) { - start = handlerMetrics.beforeHandle(message); + start = handlerMetrics.beforeHandle(); } this.handleMessageInternal(message); if (countsEnabled) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java index ef92188342..9e287efcbe 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java @@ -21,7 +21,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.support.management.ConfigurableMetrics; import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.support.management.Statistics; -import org.springframework.messaging.Message; /** * Abstract base class for handler metrics implementations. @@ -57,14 +56,13 @@ public abstract class AbstractMessageHandlerMetrics implements ConfigurableMetri /** * Begin a handle event. - * @param message the message to handle. * @return the context to be used in the {@link #afterHandle(MetricsContext, boolean)}. */ - public abstract MetricsContext beforeHandle(Message message); + public abstract MetricsContext beforeHandle(); /** * End a handle event - * @param context the context from the previous {@link #beforeHandle(Message)}. + * @param context the context from the previous {@link #beforeHandle()}. * @param success true for success, false otherwise. */ public abstract void afterHandle(MetricsContext context, boolean success); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java new file mode 100644 index 0000000000..c6cb917383 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AggregatingMessageHandlerMetrics.java @@ -0,0 +1,99 @@ +/* + * Copyright 2002-2015 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.integration.handler.management; + +import org.springframework.integration.support.management.ExponentialMovingAverage; +import org.springframework.integration.support.management.MetricsContext; + +/** + * An implementation of {@link MessageHandlerMetrics} that aggregates the total response + * time over a sample, to avoid fetching the system time twice for every message. + * + * @author Gary Russell + * @since 2.0 + */ +public class AggregatingMessageHandlerMetrics extends DefaultMessageHandlerMetrics { + + private static final int DEFAULT_SAMPLE_SIZE = 1000; + + private final int sampleSize; + + private long start; + + public AggregatingMessageHandlerMetrics() { + this(null, DEFAULT_SAMPLE_SIZE); + } + + /** + * Construct an instance with the default moving average window (10). + * @param name the name. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageHandlerMetrics(String name, int sampleSize) { + super(name); + this.sampleSize = sampleSize; + } + + /** + * Construct an instance with the supplied {@link ExponentialMovingAverage} calculating + * the duration of processing by the message handler (and any downstream synchronous + * endpoints). + * @param name the name. + * @param duration an {@link ExponentialMovingAverage} for calculating the duration. + * @param sampleSize the sample size over which to aggregate the duration. + */ + public AggregatingMessageHandlerMetrics(String name, ExponentialMovingAverage duration, int sampleSize) { + super(name, duration); + this.sampleSize = sampleSize; + } + + @Override + public synchronized MetricsContext beforeHandle() { + long count = this.handleCount.getAndIncrement(); + if (isFullStatsEnabled() && count % this.sampleSize == 0) { + this.start = System.nanoTime(); + } + this.activeCount.incrementAndGet(); + return new AggregatingHandlerMetricsContext(this.start, count + 1); + } + + @Override + public void afterHandle(MetricsContext context, boolean success) { + this.activeCount.decrementAndGet(); + AggregatingHandlerMetricsContext aggregatingContext = (AggregatingHandlerMetricsContext) context; + if (success) { + if (isFullStatsEnabled() && aggregatingContext.newCount % this.sampleSize == 0) { + this.duration.append(System.nanoTime() - aggregatingContext.start); + } + } + else { + this.errorCount.incrementAndGet(); + } + } + + protected static class AggregatingHandlerMetricsContext extends DefaultHandlerMetricsContext { + + protected long newCount; + + public AggregatingHandlerMetricsContext(long start, long newCount) { + super(start); + this.newCount = newCount; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java index a574d490c9..6e0e0cb1d9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java @@ -21,7 +21,6 @@ import java.util.concurrent.atomic.AtomicLong; import org.springframework.integration.support.management.ExponentialMovingAverage; import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.support.management.Statistics; -import org.springframework.messaging.Message; /** * Default implementation; use the full constructor to customize the moving averages. @@ -35,13 +34,13 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; - private final AtomicLong activeCount = new AtomicLong(); + protected final AtomicLong activeCount = new AtomicLong(); - private final AtomicLong handleCount = new AtomicLong(); + protected final AtomicLong handleCount = new AtomicLong(); - private final AtomicLong errorCount = new AtomicLong(); + protected final AtomicLong errorCount = new AtomicLong(); - private final ExponentialMovingAverage duration; + protected final ExponentialMovingAverage duration; public DefaultMessageHandlerMetrics() { this(null); @@ -69,10 +68,7 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics } @Override - public MetricsContext beforeHandle(Message message) { - if (logger.isTraceEnabled()) { - logger.trace("messageHandler(" + this.name + ") message(" + message + ") :"); - } + public MetricsContext beforeHandle() { long start = 0; if (isFullStatsEnabled()) { start = System.nanoTime(); @@ -158,11 +154,11 @@ public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics return this.duration.getStatistics(); } - private static class DefaultHandlerMetricsContext implements MetricsContext { + protected static class DefaultHandlerMetricsContext implements MetricsContext { - private final long start; + protected final long start; - public DefaultHandlerMetricsContext(long start) { + protected DefaultHandlerMetricsContext(long start) { this.start = start; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java index da0d4879de..87b0bda91c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java @@ -13,8 +13,9 @@ package org.springframework.integration.support.management; +import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.LinkedList; +import java.util.Deque; import java.util.List; @@ -41,7 +42,7 @@ public class ExponentialMovingAverage { private volatile double max; - private final List samples = new LinkedList(); + private final Deque samples = new ArrayDeque(); private final int retention; @@ -86,7 +87,7 @@ public class ExponentialMovingAverage { */ public synchronized void append(double value) { if (this.samples.size() == this.retention) { - samples.remove(0); + this.samples.poll(); } samples.add(value); count++;//NOSONAR - false positive, we're synchronized diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java index 5a2f250d00..cdedb4517c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java @@ -13,8 +13,9 @@ package org.springframework.integration.support.management; +import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.LinkedList; +import java.util.Deque; import java.util.List; @@ -52,7 +53,7 @@ public class ExponentialMovingAverageRate { private final double period; - private final List times = new LinkedList(); + private final Deque times = new ArrayDeque(); private final int retention; @@ -109,7 +110,7 @@ public class ExponentialMovingAverageRate { */ public synchronized void increment(long t) { if (this.times.size() == this.retention) { - this.times.remove(0); + this.times.poll(); } this.times.add(t); this.count++;//NOSONAR - false positive, we're synchronized @@ -202,11 +203,9 @@ public class ExponentialMovingAverageRate { return count / (count / calc().getMean() + value); } - private double lastTime() { + private synchronized double lastTime() { if (this.times.size() > 0) { - synchronized (this) { - return this.times.get(this.times.size() - 1) / this.factor; - } + return this.times.peekFirst() / this.factor; } else { return this.t0; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java index a4a2fcfe6a..d82e759eeb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java @@ -13,9 +13,10 @@ package org.springframework.integration.support.management; +import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Deque; import java.util.Iterator; -import java.util.LinkedList; import java.util.List; @@ -50,9 +51,9 @@ public class ExponentialMovingAverageRatio { private final double lapse; - private final List times = new LinkedList(); + private final Deque times = new ArrayDeque(); - private final List values = new LinkedList(); + private final Deque values = new ArrayDeque(); private final int retention; @@ -124,8 +125,8 @@ public class ExponentialMovingAverageRatio { private synchronized void append(int value, long t) { if (this.times.size() == this.retention) { - this.times.remove(0); - this.values.remove(0); + this.times.poll(); + this.values.poll(); } this.times.add(t); this.values.add(value); @@ -221,11 +222,9 @@ public class ExponentialMovingAverageRatio { return alpha * mean + 1 - alpha; } - private double lastTime() { + private synchronized double lastTime() { if (this.times.size() > 0) { - synchronized (this) { - return this.times.get(this.times.size() - 1); - } + return this.times.peekFirst(); } else { return this.t0 * this.factor; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java index 02a648da5a..db6da494b9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java @@ -25,12 +25,21 @@ import org.junit.Test; /** * @author Dave Syer * @author Artem Bilan + * @author Gary Russell * */ public class ExponentialMovingAverageTests { private final ExponentialMovingAverage history = new ExponentialMovingAverage(10); + + @Test @Ignore // used to compare LinkedList to ArrayDeque which was 35% faster + public void perf() { + for (int i = 0; i < 100000000; i++) { + history.append(0.0); + } + } + @Test public void testGetCount() { assertEquals(0, history.getCount()); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java new file mode 100644 index 0000000000..f30a571535 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/AggregatingMetricsFactory.java @@ -0,0 +1,53 @@ +/* + * Copyright 2015 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.integration.monitor; + +import org.springframework.integration.channel.management.AbstractMessageChannelMetrics; +import org.springframework.integration.channel.management.AggregatingMessageChannelMetrics; +import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics; +import org.springframework.integration.handler.management.AggregatingMessageHandlerMetrics; + + + +/** + * Implementation that returns aggregating metrics. + * + * @author Gary Russell + * @since 4.2 + * + */ +public class AggregatingMetricsFactory implements MetricsFactory { + + private final int sampleSize; + + /** + * @param sampleSize the number of messages over which to aggregate the elapsed time. + */ + public AggregatingMetricsFactory(int sampleSize) { + this.sampleSize = sampleSize; + } + + @Override + public AbstractMessageChannelMetrics createChannelMetrics(String name) { + return new AggregatingMessageChannelMetrics(name, this.sampleSize); + } + + @Override + public AbstractMessageHandlerMetrics createHandlerMetrics(String name) { + return new AggregatingMessageHandlerMetrics(name, this.sampleSize); + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml new file mode 100644 index 0000000000..d513159e94 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests-context.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java new file mode 100644 index 0000000000..4fac604c46 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/AggregatingMetricsTests.java @@ -0,0 +1,107 @@ +/* + * Copyright 2015 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.integration.monitor; + +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; + +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.management.AggregatingMessageChannelMetrics; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.handler.ServiceActivatingHandler; +import org.springframework.integration.handler.management.AggregatingMessageHandlerMetrics; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 4.2 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class AggregatingMetricsTests { + + @Autowired + private MessageChannel input; + + @Autowired + private AbstractMessageChannel delay; + + @Autowired + private QueueChannel output; + + @Autowired + private BridgeHandler handler; + + @Autowired + private ServiceActivatingHandler delayer; + + @Test + public void testCounts() { + Message message = new GenericMessage("foo"); + int count = 2000; + for (int i = 0; i < count; i++) { + input.send(message); + } + assertEquals(count, this.output.getQueueSize()); + assertEquals(count, this.output.getSendCount()); + assertEquals(Long.valueOf(count / 1000).longValue(), this.output.getSendDuration().getCountLong()); + assertEquals(count, this.handler.getHandleCount()); + assertEquals(Long.valueOf(count / 1000).longValue(), this.handler.getDuration().getCountLong()); + } + + @Test + public void testElapsed() { + int sampleSize = 2; + this.delay.configureMetrics(new AggregatingMessageChannelMetrics("foo", sampleSize)); + this.delay.enableStats(true); + this.delayer.configureMetrics(new AggregatingMessageHandlerMetrics("bar", sampleSize)); + this.delayer.enableStats(true); + GenericMessage message = new GenericMessage("foo"); + int count = 4; + for (int i = 0; i < count; i++) { + this.delay.send(message); + } + assertEquals(count, this.delay.getSendCount()); + assertEquals(count / sampleSize, this.delay.getSendDuration().getCount()); + assertThat((int) this.delay.getMeanSendDuration() / sampleSize, greaterThanOrEqualTo(50)); + assertEquals(count, this.delayer.getHandleCount()); + assertEquals(count / sampleSize, this.delayer.getDuration().getCount()); + assertThat((int) this.delayer.getMeanDuration() / sampleSize, greaterThanOrEqualTo(50)); + } + + @Test @Ignore + public void perf() { + AggregatingMessageHandlerMetrics metrics = new AggregatingMessageHandlerMetrics(); + for (int i = 0; i < 100000000; i++) { + metrics.afterHandle(metrics.beforeHandle(), true); + } + } + +} diff --git a/src/reference/asciidoc/jmx.adoc b/src/reference/asciidoc/jmx.adoc index 4fd4c7e3e4..c08e73f215 100644 --- a/src/reference/asciidoc/jmx.adoc +++ b/src/reference/asciidoc/jmx.adoc @@ -509,6 +509,29 @@ By default, a `DefaultMetricsFactory` provides default implementation of `Messag To override the default `MetricsFactory` use the MBean exporter's `metrics-factory` attribute to provide a reference to your `MetricsFactory` bean instance. You can either customize the default implementations as described in the next bullet, or provide completely different implementations by overriding `AbstractMessageChannelMetrics` and/or `AbstractMessageHandlerMetrics`. +In addition to the default metrics factory described above, the framework provides the `AggregatingMetricsFactory`. +This factory creates `AggregatingMessageChannelMetrics` and `AggregatingMessageHandlerMetrics`. +In very high volume scenarios, the cost of capturing statistics can be prohibitive (2 calls to the system time and +storing the data). +The aggregating metrics aggregate the response time over a sample of messages. +This can save significant CPU time. + +CAUTION: The statistics will be skewed if messages arrive in bursts. +These metrics are intended for use with high, constant-volume, message rates. + +[source, xml] +---- + + + + + + + +---- + +The above configuration aggregates the response time over 1000 messages. +Counts (send, error) are maintained per-message but the statistics are per 1000 messages. * *Customizing the Default Channel/Handler Statistics*