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

INT-1265, INT-1266, INT-1267: create new mbean exporter and move monitoring to there

Browse Source
This commit is contained in:
David Syer
2010-09-03 12:02:54 +00:00
parent 69b296443a
commit 53f975311a
36 changed files with 2501 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
spring-integration-jmx/src/main/java/org/springframework/integration/control/ControlBus.java
View File

@@ -16,43 +16,22 @@
package org.springframework.integration.control;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.jmx.JmxHeaders;
import org.springframework.integration.jmx.OperationInvokingMessageHandler;
import org.springframework.integration.monitor.ObjectNameLocator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jmx.export.MBeanExporter;
import org.springframework.jmx.export.assembler.AbstractConfigurableMBeanInfoAssembler;
import org.springframework.jmx.export.assembler.MBeanInfoAssembler;
import org.springframework.jmx.support.ObjectNameManager;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ObjectUtils;
/**
* JMX-based Control Bus implementation. Exports all channel and endpoint beans from a given BeanFactory as MBeans.
@@ -62,57 +41,23 @@ import org.springframework.util.ObjectUtils;
*/
public class ControlBus implements BeanFactoryAware, InitializingBean {
public static final String DEFAULT_DOMAIN = "org.springframework.integration";
public static final String TARGET_BEAN_NAME = JmxHeaders.PREFIX + "_controlBus_targetBeanName";
private volatile SubscribableChannel operationChannel;
private final MBeanExporter exporter;
private final String domain;
private final Map<String, ObjectName> exportedBeanObjectNameMap = new HashMap<String, ObjectName>();
private final Map<String, String> objectNameStaticProperties = new HashMap<String, String>();
private volatile ListableBeanFactory beanFactory;
private final Set<Class<?>> managedTypes = new HashSet<Class<?>>(Arrays.asList(new Class<?>[] {
MessageChannel.class, AbstractEndpoint.class }));
private final ObjectNameLocator exporter;
private final MBeanServer server;
/**
* Static properties that will be added to all object names.
*
* @param objectNameStaticProperties the objectNameStaticProperties to set
* Create a {@link ControlBus}.
*/
public void setObjectNameStaticProperties(Map<String, String> objectNameStaticProperties) {
this.objectNameStaticProperties.putAll(objectNameStaticProperties);
}
/**
* Create a {@link ControlBus} that will register channels and endpoints as MBeans with the given MBeanServer using
* the default domain name.
* @see #DEFAULT_DOMAIN
*/
public ControlBus(MBeanServer server) {
this(server, null);
}
/**
* Create a {@link ControlBus} that will register channels and endpoints as MBeans with the given MBeanServer using
* the specified domain name.
*/
public ControlBus(MBeanServer server, String domain) {
Assert.notNull(server, "MBeanServer must not be null.");
this.domain = (domain != null) ? domain : DEFAULT_DOMAIN;
Assert.isTrue(!ObjectUtils.containsElement(server.getDomains(), this.domain), "Domain [" + this.domain
+ "] is already in use within this MBeanServer.");
MBeanExporter exporter = new MBeanExporter();
exporter.setServer(server);
exporter.setAutodetect(false);
exporter.setAssembler(new ControlBusMBeanInfoAssembler());
this.exporter = exporter;
public ControlBus(ObjectNameLocator locator, MBeanServer server) {
this.exporter = locator;
this.server = server;
}
public void setOperationChannel(SubscribableChannel operationChannel) {
@@ -133,28 +78,12 @@ public class ControlBus implements BeanFactoryAware, InitializingBean {
Assert.isTrue(beanFactory instanceof ListableBeanFactory, "A ListableBeanFactory is required.");
this.beanFactory = (ListableBeanFactory) beanFactory;
}
public void afterPropertiesSet() throws Exception {
this.exporter.afterPropertiesSet();
for (Class<?> type : this.managedTypes) {
Map<String, ?> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(this.beanFactory, type);
for (Map.Entry<String, ?> entry : beans.entrySet()) {
Object bean = entry.getValue();
String beanName = entry.getKey();
Class<?> beanType = bean.getClass();
try {
ObjectName objectName = this.generateObjectName(beanName, beanType);
this.exporter.registerManagedResource(bean, objectName);
this.exportedBeanObjectNameMap.put(beanName, objectName);
}
catch (MalformedObjectNameException e) {
throw new BeanInitializationException("Failed to generate JMX ObjectName.", e);
}
}
}
OperationInvokingMessageHandler handler = new ControlBusOperationInvokingMessageHandler();
handler.setBeanFactory(this.beanFactory);
handler.setServer(this.exporter.getServer());
handler.setServer(this.server);
handler.afterPropertiesSet();
if (this.operationChannel == null) {
this.operationChannel = new DirectChannel();
@@ -162,89 +91,13 @@ public class ControlBus implements BeanFactoryAware, InitializingBean {
this.operationChannel.subscribe(handler);
}
protected ObjectName generateObjectName(String beanName, Class<?> beanType) throws MalformedObjectNameException {
String name = beanName.startsWith("org.springframework.integration") ? "anonymous,generated="+beanName : beanName;
StringBuilder sb = new StringBuilder(this.domain + ":name=" + name + ",");
if (MessageChannel.class.isAssignableFrom(beanType)) {
sb.append("type=channel");
}
else if (AbstractEndpoint.class.isAssignableFrom(beanType)) {
sb.append("type=endpoint");
}
else {
sb.append("type=" + ClassUtils.getShortNameAsProperty(beanType));
}
for (String key : objectNameStaticProperties.keySet()) {
sb.append("," + key + "=" + objectNameStaticProperties.get(key));
}
return ObjectNameManager.getInstance(sb.toString());
}
/**
* An {@link MBeanInfoAssembler} implementation for channels and endpoints.
*/
private static class ControlBusMBeanInfoAssembler extends AbstractConfigurableMBeanInfoAssembler {
@Override
protected boolean includeOperation(Method method, String beanKey) {
Class<?> declaringClass = method.getDeclaringClass();
return this.shouldInclude(method, declaringClass);
}
@Override
protected boolean includeReadAttribute(Method method, String beanKey) {
Class<?> declaringClass = method.getDeclaringClass();
return this.shouldInclude(method, declaringClass);
}
@Override
protected boolean includeWriteAttribute(Method method, String beanKey) {
Class<?> declaringClass = method.getDeclaringClass();
return this.shouldInclude(method, declaringClass);
}
private boolean shouldInclude(Method method, Class<?> declaringClass) {
if (MessageChannel.class.isAssignableFrom(declaringClass)
|| AbstractEndpoint.class.isAssignableFrom(declaringClass)) {
Class<?> managementInterface = this.getManagementInterface(declaringClass);
if (managementInterface != null) {
for (Method interfaceMethod : managementInterface.getMethods()) {
if (interfaceMethod.getName().equals(method.getName())
&& Arrays.equals(interfaceMethod.getParameterTypes(), method.getParameterTypes())) {
return true;
}
}
}
}
return false;
}
private Class<?> getManagementInterface(Class<?> type) {
if (AbstractEndpoint.class.isAssignableFrom(type)) {
return Lifecycle.class;
}
if (QueueChannel.class.isAssignableFrom(type)) {
return QueueChannelInfo.class;
}
if (PollableChannel.class.isAssignableFrom(type)) {
return PollableChannelInfo.class;
}
if (MessageChannel.class.isAssignableFrom(type)) {
return MessageChannelInfo.class;
}
return null;
}
}
private class ControlBusOperationInvokingMessageHandler extends OperationInvokingMessageHandler {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
String beanName = requestMessage.getHeaders().get(TARGET_BEAN_NAME, String.class);
Assert.notNull(beanName, "The ControlBus.TARGET_BEAN_NAME is required.");
ObjectName objectName = exportedBeanObjectNameMap.get(beanName);
String objectName = exporter.getObjectName(beanName);
Assert.notNull(objectName, "ControlBus has not exported an MBean for '" + beanName + "'");
requestMessage = MessageBuilder.fromMessage(requestMessage).setHeader(JmxHeaders.OBJECT_NAME, objectName)
.setHeader(TARGET_BEAN_NAME, null).build();

41
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/ControlBusParser.java
View File

@@ -18,13 +18,14 @@ package org.springframework.integration.jmx.config;
import javax.management.MBeanServerFactory;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Mark Fisher
@@ -39,20 +40,30 @@ public class ControlBusParser extends AbstractSimpleBeanDefinitionParser {
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
Object source = parserContext.extractSource(element);
builder.getRawBeanDefinition().setSource(source);
String mbeanServer = element.getAttribute("mbean-server");
if (StringUtils.hasText(mbeanServer)) {
builder.addConstructorArgReference(mbeanServer);
}
else {
builder.addConstructorArgValue(MBeanServerFactory.createMBeanServer());
}
String domain = element.getAttribute("domain");
if (StringUtils.hasText(domain)) {
builder.addConstructorArgValue(domain);
}
builder.getRawBeanDefinition().setSource(parserContext.extractSource(element));
Object mbeanServer = getMBeanServer(element, parserContext);
builder.addConstructorArgValue(getMBeanExporter(element, parserContext, mbeanServer));
builder.addConstructorArgValue(mbeanServer);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "operation-channel");
}
private Object getMBeanServer(Element element, ParserContext parserContext) {
String mbeanServer = element.getAttribute("mbean-server");
if (StringUtils.hasText(mbeanServer)) {
return new RuntimeBeanReference(mbeanServer);
}
else {
return MBeanServerFactory.createMBeanServer();
}
}
private BeanMetadataElement getMBeanExporter(Element element, ParserContext parserContext, Object mbeanServer) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition("org.springframework.integration.monitor.IntegrationMBeanExporter");
builder.getRawBeanDefinition().setSource(parserContext.extractSource(element));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "domain");
builder.addPropertyValue("server", mbeanServer);
return builder.getBeanDefinition();
}
}

85
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageCumulativeHistory.java Normal file
View File

@@ -0,0 +1,85 @@
/*
* 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;
/**
* Cumulative statistics for a series of real numbers with higher weight given to recent data but without storing any
* history. Older values are given exponentially smaller weight, with a decay factor determined by a "window" size
* chosen by the client.
*
* @author Dave Syer
*
*/
public class ExponentialMovingAverageCumulativeHistory {
private int count;
private double weight;
private double sum;
private double sumSquares;
private double min;
private double max;
private final double decay;
/**
* @param window the exponential lapse window (number of measurements)
*/
public ExponentialMovingAverageCumulativeHistory(int window) {
this.decay = 1 - 1. / window;
}
public void append(double value) {
if (value > max || count == 0)
max = value;
if (value < min || count == 0)
min = value;
sum = decay * sum + value;
sumSquares = decay * sumSquares + value * value;
weight = decay * weight + 1;
count++;
}
public int getCount() {
return count;
}
public double getMean() {
return weight > 0 ? sum / weight : 0.;
}
public double getStandardDeviation() {
double mean = getMean();
double var = weight > 0 ? sumSquares / weight - mean * mean : 0.;
return var > 0 ? Math.sqrt(var) : 0;
}
public double getMax() {
return max;
}
public double getMin() {
return min;
}
@Override
public String toString() {
return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", count, min, max, getMean(),
getStandardDeviation());
}
}

108
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRateCumulativeHistory.java Normal file
View File

@@ -0,0 +1,108 @@
/*
* 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;
/**
* Cumulative statistics for rate with higher weight given to recent data but without storing any history. Older values
* are given exponentially smaller weight, with a decay factor determined by a duration chosen by the client.
*
* @author Dave Syer
*
*/
public class ExponentialMovingAverageRateCumulativeHistory {
private final ExponentialMovingAverageCumulativeHistory rates;
private double weight;
private double sum;
private double min;
private double max;
private volatile long t0 = System.currentTimeMillis();
private final double lapse;
private final double period;
/**
* @param period the period to base the rate measurement (in seconds)
* @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
* @param window the exponential lapse window (number of measurements)
*/
public ExponentialMovingAverageRateCumulativeHistory(double period, double lapsePeriod, int window) {
rates = new ExponentialMovingAverageCumulativeHistory(10);
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
this.period = period * 1000; // convert to millisecs
}
public void increment() {
long t = System.currentTimeMillis();
double value = t > t0 ? (t - t0) / period : 0;
if (value > max || getCount() == 0) {
max = value;
}
if (value < min || getCount() == 0) {
min = value;
}
double alpha = Math.exp((t0 - t) * lapse);
t0 = t;
sum = alpha * sum + value;
weight = alpha * weight + 1;
rates.append(sum > 0 ? weight / sum : 0);
}
public int getCount() {
return rates.getCount();
}
/**
* @return the time in seconds since the last measurement
*/
public double getTimeSinceLastMeasurement() {
return (System.currentTimeMillis() - t0) / 1000.;
}
public double getMean() {
int count = rates.getCount();
if (count==0) {
return 0;
}
long t = System.currentTimeMillis();
double value = t > t0 ? (t - t0) / period : 0;
return count / (count / rates.getMean() + value);
}
public double getStandardDeviation() {
return rates.getStandardDeviation();
}
public double getMax() {
return min > 0 ? 1 / min : 0;
}
public double getMin() {
return max > 0 ? 1 / max : 0;
}
@Override
public String toString() {
return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f, timeSinceLast=%f]", getCount(), getMin(),
getMax(), getMean(), getStandardDeviation(), getTimeSinceLastMeasurement());
}
}

102
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioCumulativeHistory.java Normal file
View File

@@ -0,0 +1,102 @@
/*
* 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;
/**
* Cumulative statistics for success rate (ratio) with higher weight given to recent data but without storing any
* history. Older values are given exponentially smaller weight, with a decay factor determined by a duration chosen by
* the client.
*
* @author Dave Syer
*
*/
public class ExponentialMovingAverageRatioCumulativeHistory {
private double weight;
private double sum;
private volatile long t0 = System.currentTimeMillis();
private final double lapse;
private final ExponentialMovingAverageCumulativeHistory cumulative;
/**
* @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
* @param window the exponential lapse window (number of measurements)
*/
public ExponentialMovingAverageRatioCumulativeHistory(double lapsePeriod, int window) {
this.cumulative = new ExponentialMovingAverageCumulativeHistory(window);
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
}
public void success() {
append(1);
}
public void failure() {
append(0);
}
private void append(int value) {
long t = System.currentTimeMillis();
double alpha = Math.exp((t0 - t) * lapse);
t0 = t;
sum = alpha * sum + value;
weight = alpha * weight + 1;
cumulative.append(sum / weight);
}
public int getCount() {
return cumulative.getCount();
}
/**
* @return the time in seconds since the last measurement
*/
public double getTimeSinceLastMeasurement() {
return (System.currentTimeMillis() - t0) / 1000.;
}
public double getMean() {
int count = cumulative.getCount();
if (count == 0) {
return 0;
}
long t = System.currentTimeMillis();
double alpha = Math.exp((t0 - t) * lapse);
return alpha * cumulative.getMean() + 1 - alpha;
}
public double getStandardDeviation() {
return cumulative.getStandardDeviation();
}
public double getMax() {
return cumulative.getMax();
}
public double getMin() {
return cumulative.getMin();
}
@Override
public String toString() {
return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f, timeSinceLast=%f]", getCount(), getMin(),
getMax(), getMean(), getStandardDeviation(), getTimeSinceLastMeasurement());
}
}

544
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java Normal file
View File

@@ -0,0 +1,544 @@
/*
* 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.lang.reflect.Field;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.PointcutAdvisor;
import org.springframework.aop.TargetSource;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.jmx.export.MBeanExporter;
import org.springframework.jmx.export.annotation.AnnotationJmxAttributeSource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.jmx.export.assembler.MetadataMBeanInfoAssembler;
import org.springframework.jmx.export.naming.MetadataNamingStrategy;
import org.springframework.jmx.support.MetricType;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* MBean exporter for Spring Integration components in an existing application.
*
* @author Dave Syer
* @author Helena Edelson
*/
@ManagedResource
public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware,
BeanClassLoaderAware, SmartLifecycle, ObjectNameLocator {
private static final Log logger = LogFactory.getLog(IntegrationMBeanExporter.class);
public static final String DEFAULT_DOMAIN = "spring.application";
private Set<String> channelKeys = new HashSet<String>();
private Set<String> handlerKeys = new HashSet<String>();
private final AnnotationJmxAttributeSource attributeSource = new AnnotationJmxAttributeSource();
private ListableBeanFactory beanFactory;
private Map<Object, AtomicLong> anonymousCounters = new HashMap<Object, AtomicLong>();
private Set<SimpleMessageHandlerMonitor> handlers = new HashSet<SimpleMessageHandlerMonitor>();
private Set<SimpleMessageChannelMonitor> channels = new HashSet<SimpleMessageChannelMonitor>();
private Map<String, SimpleMessageChannelMonitor> channelsByName = new HashMap<String, SimpleMessageChannelMonitor>();
private Map<String, MessageHandlerMonitor> handlersByName = new HashMap<String, MessageHandlerMonitor>();
private Map<String, String> objectNamesByName = new HashMap<String, String>();
private ClassLoader beanClassLoader;
private volatile boolean autoStartup = true;
private volatile int phase = 0;
private volatile boolean running;
private final ReentrantLock lifecycleLock = new ReentrantLock();
private String domain = DEFAULT_DOMAIN;
private final Map<String, String> objectNameStaticProperties = new HashMap<String, String>();
public IntegrationMBeanExporter() {
super();
// Shouldn't be necessary, but to be on the safe side...
setAutodetect(false);
setNamingStrategy(new MetadataNamingStrategy(attributeSource));
setAssembler(new MetadataMBeanInfoAssembler(attributeSource));
}
@Override
public void setBeanClassLoader(ClassLoader classLoader) {
this.beanClassLoader = classLoader;
super.setBeanClassLoader(classLoader);
}
/**
* Static properties that will be added to all object names.
*
* @param objectNameStaticProperties the objectNameStaticProperties to set
*/
public void setObjectNameStaticProperties(Map<String, String> objectNameStaticProperties) {
this.objectNameStaticProperties.putAll(objectNameStaticProperties);
}
/**
* The JMX domain to use for MBeans registered. Defaults to <code>spring.application</code> (which is useful in
* SpringSource HQ).
*
* @param domain the domain name to set
*/
public void setDomain(String domain) {
this.domain = domain;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
super.setBeanFactory(beanFactory);
Assert.isTrue(beanFactory instanceof ListableBeanFactory, "A ListableBeanFactory is required.");
this.beanFactory = (ListableBeanFactory) beanFactory;
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof MessageHandler) {
SimpleMessageHandlerMonitor monitor = new SimpleMessageHandlerMonitor((MessageHandler) bean);
handlers.add(monitor);
return monitor;
}
if (bean instanceof MessageChannel) {
SimpleMessageChannelMonitor monitor;
if (bean instanceof PollableChannel) {
Object target = extractTarget(bean);
if (target instanceof QueueChannel) {
monitor = new QueueChannelMonitor((QueueChannel) target, beanName);
}
else {
monitor = new PollableChannelMonitor(beanName);
}
}
else {
monitor = new SimpleMessageChannelMonitor(beanName);
}
Object advised = applyChannelInterceptor(bean, monitor, beanClassLoader);
channels.add(monitor);
return advised;
}
return bean;
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@Override
protected void registerBeans() {
// Completely disable sup class registration to avoid duplicates
}
public final boolean isAutoStartup() {
return this.autoStartup;
}
public final int getPhase() {
return this.phase;
}
public final boolean isRunning() {
this.lifecycleLock.lock();
try {
return this.running;
}
finally {
this.lifecycleLock.unlock();
}
}
public final void start() {
this.lifecycleLock.lock();
try {
if (!this.running) {
this.doStart();
this.running = true;
if (logger.isInfoEnabled()) {
logger.info("started " + this);
}
}
}
finally {
this.lifecycleLock.unlock();
}
}
public final void stop() {
this.lifecycleLock.lock();
try {
if (this.running) {
this.doStop();
this.running = false;
if (logger.isInfoEnabled()) {
logger.info("stopped " + this);
}
}
}
finally {
this.lifecycleLock.unlock();
}
}
public final void stop(Runnable callback) {
this.lifecycleLock.lock();
try {
this.stop();
callback.run();
}
finally {
this.lifecycleLock.unlock();
}
}
protected void doStop() {
}
protected void doStart() {
registerChannels();
registerHandlers();
logger.info("Summary on start: " + objectNamesByName);
}
@Override
public void destroy() {
super.destroy();
for (MessageChannelMonitor monitor : channels) {
logger.info("Summary on shutdown: " + monitor);
}
for (MessageHandlerMonitor monitor : handlers) {
logger.info("Summary on shutdown: " + monitor);
}
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Channel Count")
public double getChannelCount() {
return channelKeys.size();
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageHandler Handler Count")
public double getHandlerCount() {
return handlerKeys.size();
}
@ManagedAttribute
public Collection<String> getHandlerNames() {
return handlersByName.keySet();
}
@ManagedAttribute
public Collection<String> getChannelNames() {
return channelsByName.keySet();
}
@ManagedOperation(description = "Get the JMX object name (as a String) for the specified Spring bean name")
public String getObjectName(String beanName) {
return objectNamesByName.get(beanName);
}
@ManagedAttribute(description = "Get a map from Spring bean names to JMX object names registered by this exporter")
public Map<String, String> getObjectNames() {
return Collections.unmodifiableMap(objectNamesByName);
}
public double getHandlerMeanDuration(String name) {
if (handlersByName.containsKey(name)) {
return handlersByName.get(name).getMeanDuration();
}
logger.debug("No handler found for (" + name + ")");
return -1;
}
public long getChannelSendCount(String name) {
if (channelsByName.containsKey(name)) {
return channelsByName.get(name).getSendCount();
}
logger.debug("No channel found for (" + name + ")");
return -1;
}
public long getChannelSendErrorCount(String name) {
if (channelsByName.containsKey(name)) {
return channelsByName.get(name).getSendErrorCount();
}
logger.debug("No channel found for (" + name + ")");
return -1;
}
public long getChannelReceiveCount(String name) {
if (channelsByName.containsKey(name)) {
if (channelsByName.get(name) instanceof PollableChannelMonitor) {
return ((PollableChannelMonitor) channelsByName.get(name)).getReceiveCount();
}
}
logger.debug("No channel found for (" + name + ")");
return -1;
}
public double getChannelSendRate(String name) {
if (channelsByName.containsKey(name)) {
return channelsByName.get(name).getSendRate();
}
logger.debug("No channel found for (" + name + ")");
return -1;
}
public double getChannelErrorRate(String name) {
if (channelsByName.containsKey(name)) {
return channelsByName.get(name).getErrorRate();
}
logger.debug("No channel found for (" + name + ")");
return -1;
}
public double getChannelMeanSendDuration(String name) {
if (channelsByName.containsKey(name)) {
return channelsByName.get(name).getMeanSendDuration();
}
logger.debug("No channel found for (" + name + ")");
return -1;
}
private void registerChannels() {
for (SimpleMessageChannelMonitor monitor : channels) {
String name = monitor.getName();
// Only register once...
if (!channelsByName.containsKey(name)) {
String beanKey = getChannelBeanKey(name);
logger.info("Registering MessageChannel " + name);
if (name != null) {
channelsByName.put(name, monitor);
objectNamesByName.put(name, beanKey);
}
registerBeanNameOrInstance(monitor, beanKey);
}
}
}
private void registerHandlers() {
for (SimpleMessageHandlerMonitor source : handlers) {
MessageHandlerMonitor monitor = enhanceMonitor(source);
String name = monitor.getName();
// Only register once...
if (!handlersByName.containsKey(name)) {
String beanKey = getHandlerBeanKey(monitor);
if (name != null) {
handlersByName.put(name, monitor);
objectNamesByName.put(name, beanKey);
}
registerBeanNameOrInstance(monitor, beanKey);
}
}
}
private Object applyChannelInterceptor(Object bean, SimpleMessageChannelMonitor interceptor,
ClassLoader beanClassLoader) {
NameMatchMethodPointcutAdvisor channelsAdvice = new NameMatchMethodPointcutAdvisor(interceptor);
channelsAdvice.addMethodName("send");
channelsAdvice.addMethodName("receive");
return applyAdvice(bean, channelsAdvice, beanClassLoader);
}
private Object extractTarget(Object bean) {
if (!(bean instanceof Advised)) {
return bean;
}
Advised advised = (Advised) bean;
if (advised.getTargetSource() == null) {
return null;
}
try {
return extractTarget(advised.getTargetSource().getTarget());
}
catch (Exception e) {
logger.error("Could not extract target", e);
return null;
}
}
private Object applyAdvice(Object bean, PointcutAdvisor advisor, ClassLoader beanClassLoader) {
Class<?> targetClass = AopUtils.getTargetClass(bean);
if (AopUtils.canApply(advisor.getPointcut(), targetClass)) {
if (bean instanceof Advised) {
((Advised) bean).addAdvisor(advisor);
return bean;
}
else {
ProxyFactory proxyFactory = new ProxyFactory(bean);
proxyFactory.addAdvisor(advisor);
return proxyFactory.getProxy(beanClassLoader);
}
}
return bean;
}
private String getChannelBeanKey(String channel) {
String name = "" + channel;
if (name.startsWith("org.springframework.integration")) {
name = name + ",source=anonymous";
}
return String.format(domain + ":type=MessageChannel,name=%s" + getStaticNames(), name);
}
private String getHandlerBeanKey(MessageHandlerMonitor 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 getStaticNames() {
if (objectNameStaticProperties.isEmpty()) {
return "";
}
StringBuilder builder = new StringBuilder();
for (String key : objectNameStaticProperties.keySet()) {
builder.append("," + key + "=" + objectNameStaticProperties.get(key));
}
return builder.toString();
}
private MessageHandlerMonitor enhanceMonitor(SimpleMessageHandlerMonitor monitor) {
MessageHandlerMonitor result = monitor;
if (monitor.getName() != null && monitor.getSource() != null) {
return monitor;
}
// Assignment algorithm and bean id, with bean id pulled reflectively out of enclosing endpoint if possible
String[] names = beanFactory.getBeanNamesForType(AbstractEndpoint.class);
String name = null;
String source = "endpoint";
Object endpoint = null;
for (String beanName : names) {
endpoint = beanFactory.getBean(beanName);
Object field = null;
try {
field = getField(endpoint, "handler");
}
catch (Exception e) {
logger.debug("Could not get handler from bean = " + beanName);
}
if (field == monitor) {
name = beanName;
break;
}
}
if (name != null && endpoint != null && name.startsWith("_org.springframework.integration")) {
name = name.substring("_org.springframework.integration".length() + 1);
source = "internal";
}
if (name != null && endpoint != null && name.startsWith("org.springframework.integration")) {
Object target = endpoint;
if (endpoint instanceof Advised) {
TargetSource targetSource = ((Advised) endpoint).getTargetSource();
if (targetSource != null) {
try {
target = targetSource.getTarget();
}
catch (Exception e) {
logger.debug("Could not get handler from bean = " + name);
}
}
}
Object field = getField(target, "inputChannel");
if (field != null) {
if (!anonymousCounters.containsKey(field)) {
anonymousCounters.put(field, new AtomicLong());
}
AtomicLong count = anonymousCounters.get(field);
long total = count.incrementAndGet();
String suffix = "";
/*
* Short hack to makes sure object names are unique if more than one endpoint has the same input channel
*/
if (total > 1) {
suffix = "#" + total;
}
name = field + suffix;
source = "anonymous";
}
}
if (endpoint instanceof Lifecycle) {
// Wrap the monitor in a lifecycle so it exposes the start/stop operations
result = new LifecycleMessageHandlerMonitor((Lifecycle) endpoint, monitor);
}
if (name == null) {
name = monitor.getMessageHandler().toString();
source = "handler";
}
monitor.setSource(source);
monitor.setName(name);
return result;
}
private static Object getField(Object target, String name) {
Assert.notNull(target, "Target object must not be null");
Field field = ReflectionUtils.findField(target.getClass(), name);
if (field == null) {
throw new IllegalArgumentException("Could not find field [" + name + "] on target [" + target + "]");
}
if (logger.isDebugEnabled()) {
logger.debug("Getting field [" + name + "] from target [" + target + "]");
}
ReflectionUtils.makeAccessible(field);
return ReflectionUtils.getField(field, target);
}
}

87
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMonitor.java Normal file
View File

@@ -0,0 +1,87 @@
/*
* 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;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
@ManagedResource
public class LifecycleMessageHandlerMonitor implements MessageHandlerMonitor {
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 String getName() {
return delegate.getName();
}
public String getSource() {
return delegate.getSource();
}
}

57
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMonitor.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* 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 dsyer
*
*/
public interface MessageChannelMonitor {
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Sends")
int getSendCount();
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Errors")
int getSendErrorCount();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Time Since Last Send in Seconds")
double getTimeSinceLastSend();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second")
double getSendRate();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
double getErrorRate();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
double getErrorRatio();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Mean Duration")
double getMeanSendDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Min Duration")
double getMinSendDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Max Duration")
double getMaxSendDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration")
double getStandardDeviationSendDuration();
}

50
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMonitor.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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 {
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Execution Count", description = "rate=1h")
int getHandleCount();
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Error Count", description = "rate=1h")
int getErrorCount();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration")
double getMeanDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Min Duration")
double getMinDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration")
double getMaxDuration();
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration")
double getStandardDeviationDuration();
String getName();
String getSource();
}

32
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ObjectNameLocator.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* 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.Map;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public interface ObjectNameLocator {
String getObjectName(String beanName);
Map<String, String> getObjectNames();
}

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

@@ -0,0 +1,86 @@
/*
* 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.AtomicLong;
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 SimpleMessageChannelMonitor {
private final AtomicLong receiveCount = new AtomicLong();
private final AtomicLong receiveErrorCount = new AtomicLong();
/**
* @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 long getReceiveCount() {
return receiveCount.get();
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Errors")
public long 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 Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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();
}
}

182
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageChannelMonitor.java Normal file
View File

@@ -0,0 +1,182 @@
/*
* 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 SimpleMessageChannelMonitor 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 ExponentialMovingAverageCumulativeHistory sendDuration = new ExponentialMovingAverageCumulativeHistory(
DEFAULT_MOVING_AVERAGE_WINDOW);
private final ExponentialMovingAverageRateCumulativeHistory sendErrorRate = new ExponentialMovingAverageRateCumulativeHistory(
ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
private final ExponentialMovingAverageRatioCumulativeHistory sendSuccessRatio = new ExponentialMovingAverageRatioCumulativeHistory(
ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
private final ExponentialMovingAverageRateCumulativeHistory sendRate = new ExponentialMovingAverageRateCumulativeHistory(
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 SimpleMessageChannelMonitor(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();
sendSuccessRatio.success();
sendDuration.append(timer.getTotalTimeSeconds());
return result;
}
catch (Throwable e) {
sendErrorCount.incrementAndGet();
sendSuccessRatio.failure();
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 getSendRate() {
return sendRate.getMean();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
public double getErrorRate() {
return sendErrorRate.getMean();
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
public double getErrorRatio() {
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();
}
@Override
public String toString() {
return String.format("MessageChannelMonitor: [name=%s, sends=%d]", name, sendCount.get());
}
}

137
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMonitor.java Normal file
View File

@@ -0,0 +1,137 @@
/**
*
*/
package org.springframework.integration.monitor;
import java.util.concurrent.atomic.AtomicInteger;
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 MessageHandler, 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 handleCount = new AtomicInteger();
private final AtomicInteger errorCount = new AtomicInteger();
private final ExponentialMovingAverageCumulativeHistory duration = new ExponentialMovingAverageCumulativeHistory(
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 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();
handler.handleMessage(message);
timer.stop();
duration.append(timer.getTotalTimeSeconds());
} catch (RuntimeException e) {
errorCount.incrementAndGet();
throw e;
} catch (Error e) {
errorCount.incrementAndGet();
throw e;
}
}
@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();
}
@Override
public String toString() {
return String.format("MessageHandlerMonitor: [name=%s, source=%s, duration=%s]", name, source, duration);
}
}