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

INT-1503: *Monitor->*Metrics

Browse Source
This commit is contained in:
Dave Syer
2010-10-08 06:28:12 -07:00
parent 48c1c6b98a
commit 517aad3d69
14 changed files with 57 additions and 1076 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMonitor.java
View File

@@ -1,200 +0,0 @@
/*
* Copyright 2009-2010 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 java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.jmx.support.MetricType;
import org.springframework.util.StopWatch;
/**
* Registers all message channels, and accumulates statistics about their performance. The statistics are then published
* locally for other components to consume and publish remotely.
*
* @author Dave Syer
* @author Helena Edelson
*/
@ManagedResource
public class DirectChannelMonitor implements MethodInterceptor, MessageChannelMonitor {
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 ExponentialMovingAverage sendDuration = new ExponentialMovingAverage(
DEFAULT_MOVING_AVERAGE_WINDOW);
private final ExponentialMovingAverageRate sendErrorRate = new ExponentialMovingAverageRate(
ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
private final ExponentialMovingAverageRatio sendSuccessRatio = new ExponentialMovingAverageRatio(
ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
private final ExponentialMovingAverageRate sendRate = new ExponentialMovingAverageRate(
ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
private final AtomicInteger sendCount = new AtomicInteger();
private final AtomicInteger sendErrorCount = new AtomicInteger();
private final String name;
public DirectChannelMonitor(String name) {
this.name = name;
}
public void destroy() {
if (logger.isDebugEnabled()) {
logger.debug(sendDuration);
}
}
public String getName() {
return name;
}
public Object invoke(MethodInvocation invocation) throws Throwable {
String method = invocation.getMethod().getName();
MessageChannel channel = (MessageChannel) invocation.getThis();
return doInvoke(invocation, method, channel);
}
protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable {
if ("send".equals(method)) {
Message<?> message = (Message<?>) invocation.getArguments()[0];
return monitorSend(invocation, channel, message);
}
return invocation.proceed();
}
private Object monitorSend(MethodInvocation invocation, MessageChannel channel, Message<?> message)
throws Throwable {
if (logger.isTraceEnabled()) {
logger.trace("Recording send on channel(" + channel + ") : message(" + message + ")");
}
final StopWatch timer = new StopWatch(channel + ".send:execution");
try {
timer.start();
sendCount.incrementAndGet();
sendRate.increment();
Object result = invocation.proceed();
timer.stop();
if ((Boolean)result) {
sendSuccessRatio.success();
sendDuration.append(timer.getTotalTimeSeconds());
} else {
sendSuccessRatio.failure();
sendErrorCount.incrementAndGet();
sendErrorRate.increment();
}
return result;
}
catch (Throwable e) {
sendSuccessRatio.failure();
sendErrorCount.incrementAndGet();
sendErrorRate.increment();
throw e;
}
finally {
if (logger.isTraceEnabled()) {
logger.trace(timer);
}
}
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Sends")
public int getSendCount() {
return sendCount.get();
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Errors")
public int getSendErrorCount() {
return sendErrorCount.get();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Time Since Last Send in Seconds")
public double getTimeSinceLastSend() {
return sendRate.getTimeSinceLastMeasurement();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second")
public double getMeanSendRate() {
return sendRate.getMean();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
public double getMeanErrorRate() {
return sendErrorRate.getMean();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
public double getMeanErrorRatio() {
return 1 - sendSuccessRatio.getMean();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Mean Duration")
public double getMeanSendDuration() {
return sendDuration.getMean();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Min Duration")
public double getMinSendDuration() {
return sendDuration.getMin();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Max Duration")
public double getMaxSendDuration() {
return sendDuration.getMax();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration")
public double getStandardDeviationSendDuration() {
return sendDuration.getStandardDeviation();
}
public Statistics getSendDuration() {
return sendDuration.getStatistics();
}
public Statistics getSendRate() {
return sendRate.getStatistics();
}
public Statistics getErrorRate() {
return sendErrorRate.getStatistics();
}
@Override
public String toString() {
return String.format("MessageChannelMonitor: [name=%s, sends=%d]", name, sendCount.get());
}
}

80
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java
View File

@@ -99,17 +99,17 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
private Map<Object, AtomicLong> anonymousSourceCounters = new HashMap<Object, AtomicLong>();
private Set<SimpleMessageHandlerMonitor> handlers = new HashSet<SimpleMessageHandlerMonitor>();
private Set<SimpleMessageHandlerMetrics> handlers = new HashSet<SimpleMessageHandlerMetrics>();
private Set<SimpleMessageSourceMonitor> sources = new HashSet<SimpleMessageSourceMonitor>();
private Set<SimpleMessageSourceMetrics> sources = new HashSet<SimpleMessageSourceMetrics>();
private Set<DirectChannelMonitor> channels = new HashSet<DirectChannelMonitor>();
private Set<DirectChannelMetrics> channels = new HashSet<DirectChannelMetrics>();
private Map<String, DirectChannelMonitor> channelsByName = new HashMap<String, DirectChannelMonitor>();
private Map<String, DirectChannelMetrics> channelsByName = new HashMap<String, DirectChannelMetrics>();
private Map<String, MessageHandlerMonitor> handlersByName = new HashMap<String, MessageHandlerMonitor>();
private Map<String, MessageHandlerMetrics> handlersByName = new HashMap<String, MessageHandlerMetrics>();
private Map<String, MessageSourceMonitor> sourcesByName = new HashMap<String, MessageSourceMonitor>();
private Map<String, MessageSourceMetrics> sourcesByName = new HashMap<String, MessageSourceMetrics>();
private Map<String, String> objectNamesByName = new HashMap<String, String>();
@@ -171,8 +171,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (bean instanceof Advised) {
for (Advisor advisor : ((Advised) bean).getAdvisors()) {
Advice advice = advisor.getAdvice();
if (advice instanceof MessageHandlerMonitor || advice instanceof MessageSourceMonitor
|| advice instanceof MessageChannelMonitor) {
if (advice instanceof MessageHandlerMetrics || advice instanceof MessageSourceMetrics
|| advice instanceof MessageChannelMetrics) {
// Already advised - so probably a factory bean product
return bean;
}
@@ -180,34 +180,34 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
if (bean instanceof MessageHandler) {
SimpleMessageHandlerMonitor monitor = null;
SimpleMessageHandlerMetrics monitor = null;
if (bean instanceof MessageProducer) {
// We need to maintain semantics of the handler also being a producer
monitor = new SimpleMessageProducingHandlerMonitor((MessageHandler) bean);
monitor = new SimpleMessageProducingHandlerMetrics((MessageHandler) bean);
} else {
monitor = new SimpleMessageHandlerMonitor((MessageHandler) bean);
monitor = new SimpleMessageHandlerMetrics((MessageHandler) bean);
}
Object advised = applyHandlerInterceptor(bean, monitor, beanClassLoader);
handlers.add(monitor);
return advised;
} else if (bean instanceof MessageSource<?>) {
SimpleMessageSourceMonitor monitor = new SimpleMessageSourceMonitor((MessageSource<?>) bean);
SimpleMessageSourceMetrics monitor = new SimpleMessageSourceMetrics((MessageSource<?>) bean);
Object advised = applySourceInterceptor(bean, monitor, beanClassLoader);
sources.add(monitor);
return advised;
}
if (bean instanceof MessageChannel) {
DirectChannelMonitor monitor;
DirectChannelMetrics monitor;
if (bean instanceof PollableChannel) {
Object target = extractTarget(bean);
if (target instanceof QueueChannel) {
monitor = new QueueChannelMonitor((QueueChannel) target, beanName);
monitor = new QueueChannelMetrics((QueueChannel) target, beanName);
} else {
monitor = new PollableChannelMonitor(beanName);
monitor = new PollableChannelMetrics(beanName);
}
} else {
monitor = new DirectChannelMonitor(beanName);
monitor = new DirectChannelMetrics(beanName);
}
Object advised = applyChannelInterceptor(bean, monitor, beanClassLoader);
channels.add(monitor);
@@ -301,10 +301,10 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
@Override
public void destroy() {
super.destroy();
for (MessageChannelMonitor monitor : channels) {
for (MessageChannelMetrics monitor : channels) {
logger.info("Summary on shutdown: " + monitor);
}
for (MessageHandlerMonitor monitor : handlers) {
for (MessageHandlerMetrics monitor : handlers) {
logger.info("Summary on shutdown: " + monitor);
}
}
@@ -327,7 +327,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Active Handler Count")
public int getActiveHandlerCount() {
int count = 0;
for (MessageHandlerMonitor monitor : handlers) {
for (MessageHandlerMetrics monitor : handlers) {
count += monitor.getActiveCount();
}
return count;
@@ -336,9 +336,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Queued Message Count")
public int getQueuedMessageCount() {
int count = 0;
for (MessageChannelMonitor monitor : channels) {
if (monitor instanceof QueueChannelMonitor) {
count += ((QueueChannelMonitor) monitor).getQueueSize();
for (MessageChannelMetrics monitor : channels) {
if (monitor instanceof QueueChannelMetrics) {
count += ((QueueChannelMetrics) monitor).getQueueSize();
}
}
return count;
@@ -369,8 +369,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
public int getChannelReceiveCount(String name) {
if (channelsByName.containsKey(name)) {
if (channelsByName.get(name) instanceof PollableChannelMonitor) {
return ((PollableChannelMonitor) channelsByName.get(name)).getReceiveCount();
if (channelsByName.get(name) instanceof PollableChannelMetrics) {
return ((PollableChannelMetrics) channelsByName.get(name)).getReceiveCount();
}
}
logger.debug("No channel found for (" + name + ")");
@@ -394,7 +394,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
private void registerChannels() {
for (DirectChannelMonitor monitor : channels) {
for (DirectChannelMetrics monitor : channels) {
String name = monitor.getName();
// Only register once...
if (!channelsByName.containsKey(name)) {
@@ -410,8 +410,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
private void registerHandlers() {
for (SimpleMessageHandlerMonitor source : handlers) {
MessageHandlerMonitor monitor = enhanceHandlerMonitor(source);
for (SimpleMessageHandlerMetrics source : handlers) {
MessageHandlerMetrics monitor = enhanceHandlerMonitor(source);
String name = monitor.getName();
// Only register once...
if (!handlersByName.containsKey(name)) {
@@ -426,8 +426,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
private void registerSources() {
for (SimpleMessageSourceMonitor source : sources) {
MessageSourceMonitor monitor = enhanceSourceMonitor(source);
for (SimpleMessageSourceMetrics source : sources) {
MessageSourceMetrics monitor = enhanceSourceMonitor(source);
String name = monitor.getName();
// Only register once...
if (!sourcesByName.containsKey(name)) {
@@ -441,21 +441,21 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
}
private Object applyChannelInterceptor(Object bean, DirectChannelMonitor interceptor, ClassLoader beanClassLoader) {
private Object applyChannelInterceptor(Object bean, DirectChannelMetrics interceptor, ClassLoader beanClassLoader) {
NameMatchMethodPointcutAdvisor channelsAdvice = new NameMatchMethodPointcutAdvisor(interceptor);
channelsAdvice.addMethodName("send");
channelsAdvice.addMethodName("receive");
return applyAdvice(bean, channelsAdvice, beanClassLoader);
}
private Object applyHandlerInterceptor(Object bean, SimpleMessageHandlerMonitor interceptor,
private Object applyHandlerInterceptor(Object bean, SimpleMessageHandlerMetrics interceptor,
ClassLoader beanClassLoader) {
NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(interceptor);
handlerAdvice.addMethodName("handleMessage");
return applyAdvice(bean, handlerAdvice, beanClassLoader);
}
private Object applySourceInterceptor(Object bean, SimpleMessageSourceMonitor interceptor,
private Object applySourceInterceptor(Object bean, SimpleMessageSourceMetrics interceptor,
ClassLoader beanClassLoader) {
NameMatchMethodPointcutAdvisor sourceAdvice = new NameMatchMethodPointcutAdvisor(interceptor);
sourceAdvice.addMethodName("receive");
@@ -501,13 +501,13 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
return String.format(domain + ":type=MessageChannel,name=%s" + getStaticNames(), name);
}
private String getHandlerBeanKey(MessageHandlerMonitor handler) {
private String getHandlerBeanKey(MessageHandlerMetrics handler) {
// This ordering of keys seems to work with default settings of JConsole
return String.format(domain + ":type=MessageHandler,name=%s,bean=%s" + getStaticNames(), handler.getName(),
handler.getSource());
}
private String getSourceBeanKey(MessageSourceMonitor handler) {
private String getSourceBeanKey(MessageSourceMetrics handler) {
// This ordering of keys seems to work with default settings of JConsole
return String.format(domain + ":type=MessageSource,name=%s,bean=%s" + getStaticNames(), handler.getName(),
handler.getSource());
@@ -524,9 +524,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
return builder.toString();
}
private MessageHandlerMonitor enhanceHandlerMonitor(SimpleMessageHandlerMonitor monitor) {
private MessageHandlerMetrics enhanceHandlerMonitor(SimpleMessageHandlerMetrics monitor) {
MessageHandlerMonitor result = monitor;
MessageHandlerMetrics result = monitor;
if (monitor.getName() != null && monitor.getSource() != null) {
return monitor;
@@ -589,7 +589,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (endpoint instanceof Lifecycle) {
// Wrap the monitor in a lifecycle so it exposes the start/stop operations
result = new LifecycleMessageHandlerMonitor((Lifecycle) endpoint, monitor);
result = new LifecycleMessageHandlerMetrics((Lifecycle) endpoint, monitor);
}
if (name == null) {
@@ -604,9 +604,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
private MessageSourceMonitor enhanceSourceMonitor(SimpleMessageSourceMonitor monitor) {
private MessageSourceMetrics enhanceSourceMonitor(SimpleMessageSourceMetrics monitor) {
MessageSourceMonitor result = monitor;
MessageSourceMetrics result = monitor;
if (monitor.getName() != null && monitor.getSource() != null) {
return monitor;
@@ -669,7 +669,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (endpoint instanceof Lifecycle) {
// Wrap the monitor in a lifecycle so it exposes the start/stop operations
result = new LifecycleMessageSourceMonitor((Lifecycle) endpoint, monitor);
result = new LifecycleMessageSourceMetrics((Lifecycle) endpoint, monitor);
}
if (name == null) {

99
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMonitor.java
View File

@@ -1,99 +0,0 @@
/*
* Copyright 2002-2010 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.context.Lifecycle;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
/**
* A {@link MessageHandlerMonitor} that exposes in addition the {@link Lifecycle} interface. The lifecycle methods can
* be used to stop and start polling endpoints, for instance, in a live system.
*
* @author Dave Syer
*
* @since 2.0
*
*/
@ManagedResource
public class LifecycleMessageHandlerMonitor implements MessageHandlerMonitor, Lifecycle {
private final Lifecycle lifecycle;
private final MessageHandlerMonitor delegate;
public LifecycleMessageHandlerMonitor(Lifecycle lifecycle, MessageHandlerMonitor delegate) {
this.lifecycle = lifecycle;
this.delegate = delegate;
}
@ManagedAttribute
public boolean isRunning() {
return lifecycle.isRunning();
}
@ManagedOperation
public void start() {
lifecycle.start();
}
@ManagedOperation
public void stop() {
lifecycle.stop();
}
public int getErrorCount() {
return delegate.getErrorCount();
}
public int getHandleCount() {
return delegate.getHandleCount();
}
public double getMaxDuration() {
return delegate.getMaxDuration();
}
public double getMeanDuration() {
return delegate.getMeanDuration();
}
public double getMinDuration() {
return delegate.getMinDuration();
}
public double getStandardDeviationDuration() {
return delegate.getStandardDeviationDuration();
}
public Statistics getDuration() {
return delegate.getDuration();
}
public String getName() {
return delegate.getName();
}
public String getSource() {
return delegate.getSource();
}
public int getActiveCount() {
return delegate.getActiveCount();
}
}

75
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMonitor.java
View File

@@ -1,75 +0,0 @@
/*
* Copyright 2002-2010 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.context.Lifecycle;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
/**
* A {@link MessageSourceMonitor} that exposes in addition the {@link Lifecycle} interface. The lifecycle methods can
* be used to stop and start polling endpoints, for instance, in a live system.
*
* @author Dave Syer
*
* @since 2.0
*
*/
@ManagedResource
public class LifecycleMessageSourceMonitor implements MessageSourceMonitor, Lifecycle {
private final Lifecycle lifecycle;
private final MessageSourceMonitor delegate;
public LifecycleMessageSourceMonitor(Lifecycle lifecycle, MessageSourceMonitor delegate) {
this.lifecycle = lifecycle;
this.delegate = delegate;
}
@ManagedAttribute
public boolean isRunning() {
return lifecycle.isRunning();
}
@ManagedOperation
public void start() {
lifecycle.start();
}
@ManagedOperation
public void stop() {
lifecycle.stop();
}
public String getName() {
return delegate.getName();
}
public String getSource() {
return delegate.getSource();
}
/**
* @return
* @see org.springframework.integration.monitor.MessageSourceMonitor#getMessageCount()
*/
public int getMessageCount() {
return delegate.getMessageCount();
}
}

107
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMonitor.java
View File

@@ -1,107 +0,0 @@
/*
* Copyright 2002-2010 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.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.support.MetricType;
/**
* Interface for all message channel monitors containing accessors for various useful metrics that are generic for all
* channel types.
*
* @author Dave Syer
*
* @since 2.0
*
*/
public interface MessageChannelMonitor {
/**
* @return the number of successful sends
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Sends")
int getSendCount();
/**
* @return the number of failed sends (either throwing an exception or rejected by the channel)
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Errors")
int getSendErrorCount();
/**
* @return the time in seconds since the last send
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Time Since Last Send in Seconds")
double getTimeSinceLastSend();
/**
* @return the mean send rate (per second)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second")
double getMeanSendRate();
/**
* @return the mean error rate (per second). Errors comprise all failed sends.
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
double getMeanErrorRate();
/**
* @return the mean ratio of failed to successful sends in approximately the last minute
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
double getMeanErrorRatio();
/**
* @return the mean send duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Mean Duration")
double getMeanSendDuration();
/**
* @return the minimum send duration (milliseconds) since startup
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Min Duration")
double getMinSendDuration();
/**
* @return the maximum send duration (milliseconds) since startup
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Max Duration")
double getMaxSendDuration();
/**
* @return the standard deviation send duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration")
double getStandardDeviationSendDuration();
/**
* @return summary statistics about the send duration (milliseconds)
*/
Statistics getSendDuration();
/**
* @return summary statistics about the send rates (per second)
*/
Statistics getSendRate();
/**
* @return summary statistics about the error rates (per second)
*/
Statistics getErrorRate();
}

73
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMonitor.java
View File

@@ -1,73 +0,0 @@
/*
* Copyright 2002-2010 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.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*/
public interface MessageHandlerMonitor {
/**
* @return the number of successful handler calls
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Execution Count", description = "rate=1h")
int getHandleCount();
/**
* @return the number of failed handler calls
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Error Count", description = "rate=1h")
int getErrorCount();
/**
* @return the maximum handler duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration")
double getMeanDuration();
/**
* @return the minimum handler duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Min Duration")
double getMinDuration();
/**
* @return the standard deviation handler duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration")
double getMaxDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration")
double getStandardDeviationDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Active Status")
int getActiveCount();
/**
* @return summary statistics about the handler duration (milliseconds)
*/
Statistics getDuration();
String getName();
String getSource();
}

36
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMonitor.java
View File

@@ -1,36 +0,0 @@
/*
* Copyright 2002-2010 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.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*/
public interface MessageSourceMonitor {
/**
* @return the number of successful handler calls
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Message Source Message Count", description = "rate=1h")
int getMessageCount();
String getName();
String getSource();
}

86
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMonitor.java
View File

@@ -1,86 +0,0 @@
/*
* Copyright 2002-2010 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 java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.integration.MessageChannel;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public class PollableChannelMonitor extends DirectChannelMonitor {
private final AtomicInteger receiveCount = new AtomicInteger();
private final AtomicInteger receiveErrorCount = new AtomicInteger();
/**
* @param name
*/
public PollableChannelMonitor(String name) {
super(name);
}
@Override
protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable {
if ("receive".equals(method)) {
return monitorReceive(invocation, channel);
}
return super.doInvoke(invocation, method, channel);
}
private Object monitorReceive(MethodInvocation invocation, MessageChannel channel) throws Throwable {
if (logger.isTraceEnabled()) {
logger.trace("Recording receive on channel(" + channel + ") ");
}
try {
Object object = invocation.proceed();
if (object!=null) {
receiveCount.incrementAndGet();
}
return object;
}
catch (Throwable e) {
receiveErrorCount.incrementAndGet();
throw e;
}
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receives")
public int getReceiveCount() {
return receiveCount.get();
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Errors")
public int getReceiveErrorCount() {
return receiveErrorCount.get();
}
@Override
public String toString() {
return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]", getName(), getSendCount(),
receiveCount.get());
}
}

50
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMonitor.java
View File

@@ -1,50 +0,0 @@
/*
* Copyright 2002-2010 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.QueueChannel;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public class QueueChannelMonitor extends PollableChannelMonitor {
private final QueueChannel channel;
/**
* @param name
*/
public QueueChannelMonitor(QueueChannel channel, String name) {
super(name);
this.channel = channel;
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Queue Size")
public int getQueueSize() {
return channel.getQueueSize();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Remaining Capacity")
public int getRemainingCapacity() {
return channel.getRemainingCapacity();
}
}

175
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMonitor.java
View File

@@ -1,175 +0,0 @@
/*
* Copyright 2002-2010 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 java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.jmx.support.MetricType;
import org.springframework.util.StopWatch;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
@ManagedResource
public class SimpleMessageHandlerMonitor implements MethodInterceptor, MessageHandlerMonitor {
private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMonitor.class);
private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
private final MessageHandler handler;
private final AtomicInteger activeCount = new AtomicInteger();
private final AtomicInteger handleCount = new AtomicInteger();
private final AtomicInteger errorCount = new AtomicInteger();
private final ExponentialMovingAverage duration = new ExponentialMovingAverage(
DEFAULT_MOVING_AVERAGE_WINDOW);
private String name;
private String source;
public SimpleMessageHandlerMonitor(MessageHandler handler) {
this.handler = handler;
}
public void setName(String name) {
this.name = name;
}
public String getName() {
return name;
}
public void setSource(String source) {
this.source = source;
}
public String getSource() {
return this.source;
}
public MessageHandler getMessageHandler() {
return handler;
}
public Object invoke(MethodInvocation invocation) throws Throwable {
String method = invocation.getMethod().getName();
if ("handleMessage".equals(method)) {
Message<?> message = (Message<?>) invocation.getArguments()[0];
handleMessage(message);
return null;
}
return invocation.proceed();
}
private void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
if (logger.isTraceEnabled()) {
logger.trace("messageHandler(" + handler + ") message(" + message + ") :");
}
String name = this.name;
if (name == null) {
name = handler.toString();
}
StopWatch timer = new StopWatch(name + ".handle:execution");
try {
timer.start();
handleCount.incrementAndGet();
activeCount.incrementAndGet();
handler.handleMessage(message);
timer.stop();
duration.append(timer.getTotalTimeSeconds());
} catch (RuntimeException e) {
errorCount.incrementAndGet();
throw e;
} catch (Error e) {
errorCount.incrementAndGet();
throw e;
} finally {
activeCount.decrementAndGet();
}
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Execution Count", description = "rate=1h")
public int getHandleCount() {
if (logger.isTraceEnabled()) {
logger.trace("Getting Handle Count:" + this);
}
return handleCount.get();
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Error Count", description = "rate=1h")
public int getErrorCount() {
return errorCount.get();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration")
public double getMeanDuration() {
return duration.getMean();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Min Duration")
public double getMinDuration() {
return duration.getMin();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration")
public double getMaxDuration() {
return duration.getMax();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration")
public double getStandardDeviationDuration() {
return duration.getStandardDeviation();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Active Count")
public int getActiveCount() {
return activeCount.get();
}
public Statistics getDuration() {
return duration.getStatistics();
}
@Override
public String toString() {
return String.format("MessageHandlerMonitor: [name=%s, source=%s, duration=%s]", name, source, duration);
}
}

38
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageProducingHandlerMonitor.java
View File

@@ -1,38 +0,0 @@
/*
* Copyright 2002-2010 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.MessageChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.MessageProducer;
import org.springframework.jmx.export.annotation.ManagedResource;
/**
* @author Oleg Zhurakousky
* @since 2.0
*
*/
@ManagedResource
public class SimpleMessageProducingHandlerMonitor extends SimpleMessageHandlerMonitor implements MessageProducer {
public SimpleMessageProducingHandlerMonitor(MessageHandler handler) {
super(handler);
}
public void setOutputChannel(MessageChannel outputChannel) {
((MessageProducer)this.getMessageHandler()).setOutputChannel(outputChannel);
}
}

80
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMonitor.java
View File

@@ -1,80 +0,0 @@
/*
* Copyright 2002-2010 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 java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.integration.core.MessageSource;
/**
* @author Dave Syer
*
* @since 2.0
*/
public class SimpleMessageSourceMonitor implements MethodInterceptor, MessageSourceMonitor {
private final AtomicInteger messageCount = new AtomicInteger();
private final MessageSource<?> messageSource;
private String source;
private String name;
public SimpleMessageSourceMonitor(MessageSource<?> messageSource) {
this.messageSource = messageSource;
}
public void setName(String name) {
this.name = name;
}
public String getName() {
return name;
}
public void setSource(String source) {
this.source = source;
}
public String getSource() {
return this.source;
}
public MessageSource<?> getMessageSource() {
return messageSource;
}
public int getMessageCount() {
return messageCount.get();
}
public Object invoke(MethodInvocation invocation) throws Throwable {
String method = invocation.getMethod().getName();
Object result = invocation.proceed();
if ("receive".equals(method) && result!=null) {
messageCount.incrementAndGet();
}
return result;
}
@Override
public String toString() {
return String.format("MessageSourceMonitor: [name=%s, source=%s, count=%d]", name, source, messageCount.get());
}
}

18
spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java
View File

@@ -43,9 +43,9 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.integration.monitor.LifecycleMessageHandlerMonitor;
import org.springframework.integration.monitor.QueueChannelMonitor;
import org.springframework.integration.monitor.DirectChannelMonitor;
import org.springframework.integration.monitor.LifecycleMessageHandlerMetrics;
import org.springframework.integration.monitor.QueueChannelMetrics;
import org.springframework.integration.monitor.DirectChannelMetrics;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.jmx.support.MBeanServerFactoryBean;
import org.springframework.jmx.support.ObjectNameManager;
@@ -88,7 +88,7 @@ public class ControlBusTests {
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test1:type=MessageChannel,name=directChannel"));
assertEquals(DirectChannelMonitor.class.getName(), instance.getClassName());
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
@@ -101,7 +101,7 @@ public class ControlBusTests {
ObjectInstance instance = mbeanServer
.getObjectInstance(ObjectNameManager
.getInstance("domain.test1b:type=MessageChannel,name=org.springframework.integration.generated#0,source=anonymous"));
assertEquals(DirectChannelMonitor.class.getName(), instance.getClassName());
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
@@ -113,7 +113,7 @@ public class ControlBusTests {
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test1a:type=MessageChannel,name=directChannel,foo=bar"));
assertEquals(DirectChannelMonitor.class.getName(), instance.getClassName());
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
@@ -124,7 +124,7 @@ public class ControlBusTests {
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test2:type=MessageChannel,name=queueChannel"));
assertEquals(QueueChannelMonitor.class.getName(), instance.getClassName());
assertEquals(QueueChannelMetrics.class.getName(), instance.getClassName());
}
@Test
@@ -139,7 +139,7 @@ public class ControlBusTests {
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test3:type=MessageHandler,name=eventDrivenConsumer,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName());
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
@Test
@@ -158,7 +158,7 @@ public class ControlBusTests {
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test4:type=MessageHandler,name=pollingConsumer,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName());
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
@Test

16
spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java
View File

@@ -26,9 +26,9 @@ import javax.management.ObjectInstance;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.monitor.LifecycleMessageHandlerMonitor;
import org.springframework.integration.monitor.QueueChannelMonitor;
import org.springframework.integration.monitor.DirectChannelMonitor;
import org.springframework.integration.monitor.LifecycleMessageHandlerMetrics;
import org.springframework.integration.monitor.QueueChannelMetrics;
import org.springframework.integration.monitor.DirectChannelMetrics;
import org.springframework.jmx.support.ObjectNameManager;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -51,21 +51,21 @@ public class ControlBusXmlTests {
public void directChannelRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageChannel,name=testDirectChannel"));
assertEquals(DirectChannelMonitor.class.getName(), instance.getClassName());
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void queueChannelRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageChannel,name=testQueueChannel"));
assertEquals(QueueChannelMonitor.class.getName(), instance.getClassName());
assertEquals(QueueChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void eventDrivenConsumerRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,name=testEventDrivenBridge,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName());
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
@Test
@@ -73,14 +73,14 @@ public class ControlBusXmlTests {
Set<ObjectInstance> instances = mbeanServer.queryMBeans(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,bean=anonymous,*"), null);
assertEquals(1, instances.size());
assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instances.iterator().next().getClassName());
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instances.iterator().next().getClassName());
}
@Test
public void pollingConsumerRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,name=testPollingBridge,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName());
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
}