diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java new file mode 100644 index 0000000000..b614dfb0dd --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java @@ -0,0 +1,200 @@ +/* + * 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 DirectChannelMetrics implements MethodInterceptor, MessageChannelMetrics { + + 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 DirectChannelMetrics(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()); + } +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java new file mode 100644 index 0000000000..5ae504fb5b --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java @@ -0,0 +1,99 @@ +/* + * 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 MessageHandlerMetrics} 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 LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Lifecycle { + + private final Lifecycle lifecycle; + + private final MessageHandlerMetrics delegate; + + public LifecycleMessageHandlerMetrics(Lifecycle lifecycle, MessageHandlerMetrics 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(); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java new file mode 100644 index 0000000000..0e7f2a967a --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java @@ -0,0 +1,75 @@ +/* + * 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 MessageSourceMetrics} 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 LifecycleMessageSourceMetrics implements MessageSourceMetrics, Lifecycle { + + private final Lifecycle lifecycle; + + private final MessageSourceMetrics delegate; + + public LifecycleMessageSourceMetrics(Lifecycle lifecycle, MessageSourceMetrics 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.MessageSourceMetrics#getMessageCount() + */ + public int getMessageCount() { + return delegate.getMessageCount(); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java new file mode 100644 index 0000000000..5c294d58d2 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java @@ -0,0 +1,107 @@ +/* + * 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 MessageChannelMetrics { + + /** + * @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(); + +} \ No newline at end of file diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java new file mode 100644 index 0000000000..c9bb3f8845 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java @@ -0,0 +1,73 @@ +/* + * 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 MessageHandlerMetrics { + + /** + * @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(); + +} \ No newline at end of file diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java new file mode 100644 index 0000000000..931c7c3909 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java @@ -0,0 +1,36 @@ +/* + * 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 MessageSourceMetrics { + + /** + * @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(); + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java new file mode 100644 index 0000000000..81e48abd63 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java @@ -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.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 PollableChannelMetrics extends DirectChannelMetrics { + + private final AtomicInteger receiveCount = new AtomicInteger(); + + private final AtomicInteger receiveErrorCount = new AtomicInteger(); + + /** + * @param name + */ + public PollableChannelMetrics(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()); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java new file mode 100644 index 0000000000..d3764b7f9d --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java @@ -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 QueueChannelMetrics extends PollableChannelMetrics { + + private final QueueChannel channel; + + /** + * @param name + */ + public QueueChannelMetrics(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(); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java new file mode 100644 index 0000000000..2971034754 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java @@ -0,0 +1,175 @@ +/* + * 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 SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHandlerMetrics { + + private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMetrics.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 SimpleMessageHandlerMetrics(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); + } + +} \ No newline at end of file diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageProducingHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageProducingHandlerMetrics.java new file mode 100644 index 0000000000..aaa0bd632e --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageProducingHandlerMetrics.java @@ -0,0 +1,38 @@ +/* + * 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 SimpleMessageProducingHandlerMetrics extends SimpleMessageHandlerMetrics implements MessageProducer { + + public SimpleMessageProducingHandlerMetrics(MessageHandler handler) { + super(handler); + } + + public void setOutputChannel(MessageChannel outputChannel) { + ((MessageProducer)this.getMessageHandler()).setOutputChannel(outputChannel); + } +} \ No newline at end of file diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java new file mode 100644 index 0000000000..312fbb3c8f --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java @@ -0,0 +1,80 @@ +/* + * 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 SimpleMessageSourceMetrics implements MethodInterceptor, MessageSourceMetrics { + + private final AtomicInteger messageCount = new AtomicInteger(); + + private final MessageSource messageSource; + + private String source; + + private String name; + + public SimpleMessageSourceMetrics(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()); + } + +}