You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gr...@apache.org on 2011/03/08 01:15:00 UTC
svn commit: r1079043 [1/2] - in /qpid/trunk/qpid/java:
broker/src/main/java/org/apache/qpid/server/
broker/src/main/java/org/apache/qpid/server/configuration/
broker/src/main/java/org/apache/qpid/server/information/management/
broker/src/main/java/org/...
Author: grkvlt
Date: Tue Mar 8 00:14:59 2011
New Revision: 1079043
URL: http://svn.apache.org/viewvc?rev=1079043&view=rev
Log:
QPID-2984: Add statistics generation for broker message delivery
Port of QPID-2932 changes from 0.5.x-dev branch to trunk.
Added:
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsCounter.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsGatherer.java
qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/stats/
qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/stats/StatisticsCounterTest.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsDeliveryTest.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsReportingTest.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTestCase.java
Modified:
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/information/management/ServerInformationMBean.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Broker_logmessages.properties
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/VirtualHost_logmessages.properties
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
qpid/trunk/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java
qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java
qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java
qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/JMXTestUtils.java
qpid/trunk/qpid/java/test-profiles/Java010Excludes
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQBrokerManagerMBean.java Tue Mar 8 00:14:59 2011
@@ -327,4 +327,74 @@ public class AMQBrokerManagerMBean exten
{
return getObjectNameForSingleInstanceMBean();
}
+
+ public void resetStatistics() throws Exception
+ {
+ getVirtualHost().resetStatistics();
+ }
+
+ public double getPeakMessageDeliveryRate()
+ {
+ return getVirtualHost().getMessageDeliveryStatistics().getPeak();
+ }
+
+ public double getPeakDataDeliveryRate()
+ {
+ return getVirtualHost().getDataDeliveryStatistics().getPeak();
+ }
+
+ public double getMessageDeliveryRate()
+ {
+ return getVirtualHost().getMessageDeliveryStatistics().getRate();
+ }
+
+ public double getDataDeliveryRate()
+ {
+ return getVirtualHost().getDataDeliveryStatistics().getRate();
+ }
+
+ public long getTotalMessagesDelivered()
+ {
+ return getVirtualHost().getMessageDeliveryStatistics().getTotal();
+ }
+
+ public long getTotalDataDelivered()
+ {
+ return getVirtualHost().getDataDeliveryStatistics().getTotal();
+ }
+
+ public double getPeakMessageReceiptRate()
+ {
+ return getVirtualHost().getMessageReceiptStatistics().getPeak();
+ }
+
+ public double getPeakDataReceiptRate()
+ {
+ return getVirtualHost().getDataReceiptStatistics().getPeak();
+ }
+
+ public double getMessageReceiptRate()
+ {
+ return getVirtualHost().getMessageReceiptStatistics().getRate();
+ }
+
+ public double getDataReceiptRate()
+ {
+ return getVirtualHost().getDataReceiptStatistics().getRate();
+ }
+
+ public long getTotalMessagesReceived()
+ {
+ return getVirtualHost().getMessageReceiptStatistics().getTotal();
+ }
+
+ public long getTotalDataReceived()
+ {
+ return getVirtualHost().getDataReceiptStatistics().getTotal();
+ }
+
+ public boolean isStatisticsEnabled()
+ {
+ return getVirtualHost().isStatisticsEnabled();
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Tue Mar 8 00:14:59 2011
@@ -345,6 +345,9 @@ public class AMQChannel implements Sessi
}
finally
{
+ long bodySize = _currentMessage.getSize();
+ long timestamp = ((BasicContentHeaderProperties) _currentMessage.getContentHeader().properties).getTimestamp();
+ _session.registerMessageReceived(bodySize, timestamp);
_currentMessage = null;
}
}
@@ -1037,6 +1040,7 @@ public class AMQChannel implements Sessi
{
getProtocolSession().getProtocolOutputConverter().writeDeliver(entry, getChannelId(),
deliveryTag, sub.getConsumerTag());
+ _session.registerMessageDelivered(entry.getMessage().getSize());
}
};
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/Main.java Tue Mar 8 00:14:59 2011
@@ -320,8 +320,7 @@ public class Main
ConfigurationManagementMBean configMBean = new ConfigurationManagementMBean();
configMBean.register();
- ServerInformationMBean sysInfoMBean =
- new ServerInformationMBean(QpidProperties.getBuildVersion(), QpidProperties.getReleaseVersion());
+ ServerInformationMBean sysInfoMBean = new ServerInformationMBean(config);
sysInfoMBean.register();
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java Tue Mar 8 00:14:59 2011
@@ -767,6 +767,36 @@ public class ServerConfiguration extends
DEFAULT_HOUSEKEEPING_PERIOD));
}
+ public long getStatisticsSamplePeriod()
+ {
+ return getConfig().getLong("statistics.sample.period", 5000L);
+ }
+
+ public boolean isStatisticsGenerationBrokerEnabled()
+ {
+ return getConfig().getBoolean("statistics.generation.broker", false);
+ }
+
+ public boolean isStatisticsGenerationVirtualhostsEnabled()
+ {
+ return getConfig().getBoolean("statistics.generation.virtualhosts", false);
+ }
+
+ public boolean isStatisticsGenerationConnectionsEnabled()
+ {
+ return getConfig().getBoolean("statistics.generation.connections", false);
+ }
+
+ public long getStatisticsReportingPeriod()
+ {
+ return getConfig().getLong("statistics.reporting.period", 0L);
+ }
+
+ public boolean isStatisticsReportResetEnabled()
+ {
+ return getConfig().getBoolean("statistics.reporting.reset", false);
+ }
+
public NetworkDriverConfiguration getNetworkConfiguration()
{
return new NetworkDriverConfiguration()
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/information/management/ServerInformationMBean.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/information/management/ServerInformationMBean.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/information/management/ServerInformationMBean.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/information/management/ServerInformationMBean.java Tue Mar 8 00:14:59 2011
@@ -22,9 +22,11 @@ package org.apache.qpid.server.informati
import java.io.IOException;
+import org.apache.qpid.common.QpidProperties;
import org.apache.qpid.management.common.mbeans.ServerInformation;
import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
import org.apache.qpid.server.management.AMQManagedObject;
+import org.apache.qpid.server.registry.ApplicationRegistry;
import javax.management.JMException;
@@ -34,12 +36,15 @@ public class ServerInformationMBean exte
{
private String buildVersion;
private String productVersion;
+ private ApplicationRegistry registry;
- public ServerInformationMBean(String buildVersion, String productVersion) throws JMException
+ public ServerInformationMBean(ApplicationRegistry applicationRegistry) throws JMException
{
super(ServerInformation.class, ServerInformation.TYPE);
- this.buildVersion = buildVersion;
- this.productVersion = productVersion;
+
+ registry = applicationRegistry;
+ buildVersion = QpidProperties.getBuildVersion();
+ productVersion = QpidProperties.getReleaseVersion();
}
public String getObjectInstanceName()
@@ -67,5 +72,75 @@ public class ServerInformationMBean exte
return productVersion;
}
+
+ public void resetStatistics() throws Exception
+ {
+ registry.resetStatistics();
+ }
+
+ public double getPeakMessageDeliveryRate()
+ {
+ return registry.getMessageDeliveryStatistics().getPeak();
+ }
+
+ public double getPeakDataDeliveryRate()
+ {
+ return registry.getDataDeliveryStatistics().getPeak();
+ }
+
+ public double getMessageDeliveryRate()
+ {
+ return registry.getMessageDeliveryStatistics().getRate();
+ }
+
+ public double getDataDeliveryRate()
+ {
+ return registry.getDataDeliveryStatistics().getRate();
+ }
+
+ public long getTotalMessagesDelivered()
+ {
+ return registry.getMessageDeliveryStatistics().getTotal();
+ }
+
+ public long getTotalDataDelivered()
+ {
+ return registry.getDataDeliveryStatistics().getTotal();
+ }
+
+ public double getPeakMessageReceiptRate()
+ {
+ return registry.getMessageReceiptStatistics().getPeak();
+ }
+
+ public double getPeakDataReceiptRate()
+ {
+ return registry.getDataReceiptStatistics().getPeak();
+ }
+
+ public double getMessageReceiptRate()
+ {
+ return registry.getMessageReceiptStatistics().getRate();
+ }
+
+ public double getDataReceiptRate()
+ {
+ return registry.getDataReceiptStatistics().getRate();
+ }
+
+ public long getTotalMessagesReceived()
+ {
+ return registry.getMessageReceiptStatistics().getTotal();
+ }
+
+ public long getTotalDataReceived()
+ {
+ return registry.getDataReceiptStatistics().getTotal();
+ }
+
+ public boolean isStatisticsEnabled()
+ {
+ return registry.isStatisticsEnabled();
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Broker_logmessages.properties
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Broker_logmessages.properties?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Broker_logmessages.properties (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/Broker_logmessages.properties Tue Mar 8 00:14:59 2011
@@ -32,4 +32,7 @@ STOPPED = BRK-1005 : Stopped
# 0 - path
CONFIG = BRK-1006 : Using configuration : {0}
# 0 - path
-LOG_CONFIG = BRK-1007 : Using logging configuration : {0}
\ No newline at end of file
+LOG_CONFIG = BRK-1007 : Using logging configuration : {0}
+
+STATS_DATA = BRK-1008 : {0,choice,0#delivered|1#received} : {1,number,#.###} kB/s peak : {2,number,#} bytes total
+STATS_MSGS = BRK-1009 : {0,choice,0#delivered|1#received} : {1,number,#.###} msg/s peak : {2,number,#} msgs total
\ No newline at end of file
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/VirtualHost_logmessages.properties
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/VirtualHost_logmessages.properties?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/VirtualHost_logmessages.properties (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/logging/messages/VirtualHost_logmessages.properties Tue Mar 8 00:14:59 2011
@@ -20,4 +20,7 @@
#
# 0 - name
CREATED = VHT-1001 : Created : {0}
-CLOSED = VHT-1002 : Closed
\ No newline at end of file
+CLOSED = VHT-1002 : Closed
+
+STATS_DATA = VHT-1003 : {0} : {1,choice,0#delivered|1#received} : {2,number,#.###} kB/s peak : {3,number,#} bytes total
+STATS_MSGS = VHT-1004 : {0} : {1,choice,0#delivered|1#received} : {2,number,#.###} msg/s peak : {3,number,#} msgs total`
\ No newline at end of file
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java Tue Mar 8 00:14:59 2011
@@ -26,8 +26,9 @@ import java.util.UUID;
import org.apache.qpid.AMQException;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.stats.StatisticsGatherer;
-public interface AMQConnectionModel
+public interface AMQConnectionModel extends StatisticsGatherer
{
/**
* get a unique id for this connection.
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java Tue Mar 8 00:14:59 2011
@@ -91,6 +91,7 @@ import org.apache.qpid.server.output.Pro
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.state.AMQState;
import org.apache.qpid.server.state.AMQStateManager;
+import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
import org.apache.qpid.transport.NetworkDriver;
@@ -171,6 +172,10 @@ public class AMQProtocolEngine implement
private final UUID _id;
private final ConfigStore _configStore;
private long _createTime = System.currentTimeMillis();
+
+ private ApplicationRegistry _registry;
+ private boolean _statisticsEnabled = false;
+ private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
public ManagedObject getManagedObject()
{
@@ -194,9 +199,10 @@ public class AMQProtocolEngine implement
_configStore = virtualHostRegistry.getConfigStore();
_id = _configStore.createId();
-
_actor.message(ConnectionMessages.OPEN(null, null, false, false));
+ _registry = virtualHostRegistry.getApplicationRegistry();
+ initialiseStatistics();
}
private AMQProtocolSessionMBean createMBean() throws JMException
@@ -1282,5 +1288,74 @@ public class AMQProtocolEngine implement
public LogSubject getLogSubject()
{
return _logSubject;
- }
+ }
+
+ public void registerMessageDelivered(long messageSize)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesDelivered.registerEvent(1L);
+ _dataDelivered.registerEvent(messageSize);
+ }
+ _virtualHost.registerMessageDelivered(messageSize);
+ }
+
+ public void registerMessageReceived(long messageSize, long timestamp)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesReceived.registerEvent(1L, timestamp);
+ _dataReceived.registerEvent(messageSize, timestamp);
+ }
+ _virtualHost.registerMessageReceived(messageSize, timestamp);
+ }
+
+ public StatisticsCounter getMessageReceiptStatistics()
+ {
+ return _messagesReceived;
+ }
+
+ public StatisticsCounter getDataReceiptStatistics()
+ {
+ return _dataReceived;
+ }
+
+ public StatisticsCounter getMessageDeliveryStatistics()
+ {
+ return _messagesDelivered;
+ }
+
+ public StatisticsCounter getDataDeliveryStatistics()
+ {
+ return _dataDelivered;
+ }
+
+ public void resetStatistics()
+ {
+ _messagesDelivered.reset();
+ _dataDelivered.reset();
+ _messagesReceived.reset();
+ _dataReceived.reset();
+ }
+
+ public void initialiseStatistics()
+ {
+ setStatisticsEnabled(!StatisticsCounter.DISABLE_STATISTICS &&
+ _registry.getConfiguration().isStatisticsGenerationConnectionsEnabled());
+
+ _messagesDelivered = new StatisticsCounter("messages-delivered-" + getSessionID());
+ _dataDelivered = new StatisticsCounter("data-delivered-" + getSessionID());
+ _messagesReceived = new StatisticsCounter("messages-received-" + getSessionID());
+ _dataReceived = new StatisticsCounter("data-received-" + getSessionID());
+ }
+
+ public boolean isStatisticsEnabled()
+ {
+ return _statisticsEnabled;
+ }
+
+ public void setStatisticsEnabled(boolean enabled)
+ {
+ _statisticsEnabled = enabled;
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBean.java Tue Mar 8 00:14:59 2011
@@ -37,19 +37,8 @@
*/
package org.apache.qpid.server.protocol;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ConnectionCloseBody;
-import org.apache.qpid.framing.MethodRegistry;
-import org.apache.qpid.management.common.mbeans.ManagedConnection;
-import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
-import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
-import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.actors.ManagementActor;
-import org.apache.qpid.server.management.AMQManagedObject;
-import org.apache.qpid.server.management.ManagedObject;
+import java.util.Date;
+import java.util.List;
import javax.management.JMException;
import javax.management.MBeanException;
@@ -67,8 +56,20 @@ import javax.management.openmbean.Simple
import javax.management.openmbean.TabularData;
import javax.management.openmbean.TabularDataSupport;
import javax.management.openmbean.TabularType;
-import java.util.Date;
-import java.util.List;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.ConnectionCloseBody;
+import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.management.common.mbeans.ManagedConnection;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanConstructor;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanDescription;
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.actors.ManagementActor;
+import org.apache.qpid.server.management.AMQManagedObject;
+import org.apache.qpid.server.management.ManagedObject;
/**
* This MBean class implements the management interface. In order to make more attributes, operations and notifications
@@ -339,4 +340,78 @@ public class AMQProtocolSessionMBean ext
_broadcaster.sendNotification(n);
}
-} // End of MBean class
+ public void resetStatistics() throws Exception
+ {
+ _protocolSession.resetStatistics();
+ }
+
+ public double getPeakMessageDeliveryRate()
+ {
+ return _protocolSession.getMessageDeliveryStatistics().getPeak();
+ }
+
+ public double getPeakDataDeliveryRate()
+ {
+ return _protocolSession.getDataDeliveryStatistics().getPeak();
+ }
+
+ public double getMessageDeliveryRate()
+ {
+ return _protocolSession.getMessageDeliveryStatistics().getRate();
+ }
+
+ public double getDataDeliveryRate()
+ {
+ return _protocolSession.getDataDeliveryStatistics().getRate();
+ }
+
+ public long getTotalMessagesDelivered()
+ {
+ return _protocolSession.getMessageDeliveryStatistics().getTotal();
+ }
+
+ public long getTotalDataDelivered()
+ {
+ return _protocolSession.getDataDeliveryStatistics().getTotal();
+ }
+
+ public double getPeakMessageReceiptRate()
+ {
+ return _protocolSession.getMessageReceiptStatistics().getPeak();
+ }
+
+ public double getPeakDataReceiptRate()
+ {
+ return _protocolSession.getDataReceiptStatistics().getPeak();
+ }
+
+ public double getMessageReceiptRate()
+ {
+ return _protocolSession.getMessageReceiptStatistics().getRate();
+ }
+
+ public double getDataReceiptRate()
+ {
+ return _protocolSession.getDataReceiptStatistics().getRate();
+ }
+
+ public long getTotalMessagesReceived()
+ {
+ return _protocolSession.getMessageReceiptStatistics().getTotal();
+ }
+
+ public long getTotalDataReceived()
+ {
+ return _protocolSession.getDataReceiptStatistics().getTotal();
+ }
+
+ public boolean isStatisticsEnabled()
+ {
+ return _protocolSession.isStatisticsEnabled();
+ }
+
+ public void setStatisticsEnabled(boolean enabled)
+ {
+ _protocolSession.setStatisticsEnabled(enabled);
+ }
+}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java Tue Mar 8 00:14:59 2011
@@ -23,6 +23,8 @@ package org.apache.qpid.server.registry;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
import java.util.UUID;
import org.apache.commons.configuration.ConfigurationException;
@@ -41,11 +43,12 @@ import org.apache.qpid.server.configurat
import org.apache.qpid.server.logging.CompositeStartupMessageLogger;
import org.apache.qpid.server.logging.Log4jMessageLogger;
import org.apache.qpid.server.logging.RootMessageLogger;
-import org.apache.qpid.server.logging.AbstractRootMessageLogger;
import org.apache.qpid.server.logging.SystemOutMessageLogger;
+import org.apache.qpid.server.logging.actors.AbstractActor;
import org.apache.qpid.server.logging.actors.BrokerActor;
import org.apache.qpid.server.logging.actors.CurrentActor;
import org.apache.qpid.server.logging.messages.BrokerMessages;
+import org.apache.qpid.server.logging.messages.VirtualHostMessages;
import org.apache.qpid.server.management.ManagedObjectRegistry;
import org.apache.qpid.server.management.NoopManagedObjectRegistry;
import org.apache.qpid.server.plugins.PluginManager;
@@ -54,6 +57,7 @@ import org.apache.qpid.server.security.a
import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager;
import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager;
+import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.transport.QpidAcceptor;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.virtualhost.VirtualHostImpl;
@@ -104,6 +108,10 @@ public abstract class ApplicationRegistr
private ConfigStore _configStore;
protected String _registryName;
+
+ private Timer _reportingTimer;
+ private boolean _statisticsEnabled = false;
+ private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
static
{
@@ -294,6 +302,8 @@ public abstract class ApplicationRegistr
try
{
initialiseVirtualHosts();
+ initialiseStatistics();
+ initialiseStatisticsReporting();
}
finally
{
@@ -320,6 +330,72 @@ public abstract class ApplicationRegistr
{
_managedObjectRegistry = new NoopManagedObjectRegistry();
}
+
+ public void initialiseStatisticsReporting()
+ {
+ long report = _configuration.getStatisticsReportingPeriod() * 1000; // convert to ms
+ final boolean broker = _configuration.isStatisticsGenerationBrokerEnabled();
+ final boolean virtualhost = _configuration.isStatisticsGenerationVirtualhostsEnabled();
+ final boolean reset = _configuration.isStatisticsReportResetEnabled();
+
+ /* add a timer task to report statistics if generation is enabled for broker or virtualhosts */
+ if (report > 0L && (broker || virtualhost))
+ {
+ _reportingTimer = new Timer("Statistics-Reporting", true);
+
+ class StatisticsReportingTask extends TimerTask
+ {
+ private final int DELIVERED = 0;
+ private final int RECEIVED = 1;
+
+ public void run()
+ {
+ CurrentActor.set(new AbstractActor(ApplicationRegistry.getInstance().getRootMessageLogger()) {
+ public String getLogMessage()
+ {
+ return "[" + Thread.currentThread().getName() + "] ";
+ }
+ });
+
+ if (broker)
+ {
+ CurrentActor.get().message(BrokerMessages.STATS_DATA(DELIVERED, _dataDelivered.getPeak() / 1024.0, _dataDelivered.getTotal()));
+ CurrentActor.get().message(BrokerMessages.STATS_MSGS(DELIVERED, _messagesDelivered.getPeak(), _messagesDelivered.getTotal()));
+ CurrentActor.get().message(BrokerMessages.STATS_DATA(RECEIVED, _dataReceived.getPeak() / 1024.0, _dataReceived.getTotal()));
+ CurrentActor.get().message(BrokerMessages.STATS_MSGS(RECEIVED, _messagesReceived.getPeak(), _messagesReceived.getTotal()));
+ }
+
+ if (virtualhost)
+ {
+ for (VirtualHost vhost : getVirtualHostRegistry().getVirtualHosts())
+ {
+ String name = vhost.getName();
+ StatisticsCounter dataDelivered = vhost.getDataDeliveryStatistics();
+ StatisticsCounter messagesDelivered = vhost.getMessageDeliveryStatistics();
+ StatisticsCounter dataReceived = vhost.getDataReceiptStatistics();
+ StatisticsCounter messagesReceived = vhost.getMessageReceiptStatistics();
+
+ CurrentActor.get().message(VirtualHostMessages.STATS_DATA(name, DELIVERED, dataDelivered.getPeak() / 1024.0, dataDelivered.getTotal()));
+ CurrentActor.get().message(VirtualHostMessages.STATS_MSGS(name, DELIVERED, messagesDelivered.getPeak(), messagesDelivered.getTotal()));
+ CurrentActor.get().message(VirtualHostMessages.STATS_DATA(name, RECEIVED, dataReceived.getPeak() / 1024.0, dataReceived.getTotal()));
+ CurrentActor.get().message(VirtualHostMessages.STATS_MSGS(name, RECEIVED, messagesReceived.getPeak(), messagesReceived.getTotal()));
+ }
+ }
+
+ if (reset)
+ {
+ resetStatistics();
+ }
+
+ CurrentActor.remove();
+ }
+ }
+
+ _reportingTimer.scheduleAtFixedRate(new StatisticsReportingTask(),
+ report / 2,
+ report);
+ }
+ }
public static IApplicationRegistry getInstance()
{
@@ -369,6 +445,12 @@ public abstract class ApplicationRegistr
{
_logger.info("Shutting down ApplicationRegistry:" + this);
}
+
+ //Stop Statistics Reporting
+ if (_reportingTimer != null)
+ {
+ _reportingTimer.cancel();
+ }
//Stop incoming connections
unbind();
@@ -498,4 +580,76 @@ public abstract class ApplicationRegistr
getBroker().addVirtualHost(virtualHost);
return virtualHost;
}
+
+ public void registerMessageDelivered(long messageSize)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesDelivered.registerEvent(1L);
+ _dataDelivered.registerEvent(messageSize);
+ }
+ }
+
+ public void registerMessageReceived(long messageSize, long timestamp)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesReceived.registerEvent(1L, timestamp);
+ _dataReceived.registerEvent(messageSize, timestamp);
+ }
+ }
+
+ public StatisticsCounter getMessageReceiptStatistics()
+ {
+ return _messagesReceived;
+ }
+
+ public StatisticsCounter getDataReceiptStatistics()
+ {
+ return _dataReceived;
+ }
+
+ public StatisticsCounter getMessageDeliveryStatistics()
+ {
+ return _messagesDelivered;
+ }
+
+ public StatisticsCounter getDataDeliveryStatistics()
+ {
+ return _dataDelivered;
+ }
+
+ public void resetStatistics()
+ {
+ _messagesDelivered.reset();
+ _dataDelivered.reset();
+ _messagesReceived.reset();
+ _dataReceived.reset();
+
+ for (VirtualHost vhost : _virtualHostRegistry.getVirtualHosts())
+ {
+ vhost.resetStatistics();
+ }
+ }
+
+ public void initialiseStatistics()
+ {
+ setStatisticsEnabled(!StatisticsCounter.DISABLE_STATISTICS &&
+ getConfiguration().isStatisticsGenerationBrokerEnabled());
+
+ _messagesDelivered = new StatisticsCounter("messages-delivered");
+ _dataDelivered = new StatisticsCounter("bytes-delivered");
+ _messagesReceived = new StatisticsCounter("messages-received");
+ _dataReceived = new StatisticsCounter("bytes-received");
+ }
+
+ public boolean isStatisticsEnabled()
+ {
+ return _statisticsEnabled;
+ }
+
+ public void setStatisticsEnabled(boolean enabled)
+ {
+ _statisticsEnabled = enabled;
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/registry/IApplicationRegistry.java Tue Mar 8 00:14:59 2011
@@ -35,11 +35,12 @@ import org.apache.qpid.server.plugins.Pl
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.auth.database.PrincipalDatabaseManager;
import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
+import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.transport.QpidAcceptor;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
-public interface IApplicationRegistry
+public interface IApplicationRegistry extends StatisticsGatherer
{
/**
* Initialise the application registry. All initialisation must be done in this method so that any components
@@ -97,4 +98,6 @@ public interface IApplicationRegistry
ConfigStore getConfigStore();
void setConfigStore(ConfigStore store);
+
+ void initialiseStatisticsReporting();
}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsCounter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsCounter.java?rev=1079043&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsCounter.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsCounter.java Tue Mar 8 00:14:59 2011
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.qpid.server.stats;
+
+import java.util.Date;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class collects statistics and counts the total, rate per second and
+ * peak rate per second values for the events that are registered with it.
+ */
+public class StatisticsCounter
+{
+ private static final Logger _log = LoggerFactory.getLogger(StatisticsCounter.class);
+
+ public static final long DEFAULT_SAMPLE_PERIOD = Long.getLong("qpid.statistics.samplePeriod", 2000L); // 2s
+ public static final boolean DISABLE_STATISTICS = Boolean.getBoolean("qpid.statistics.disable");
+
+ private static final String COUNTER = "counter";
+ private static final AtomicLong _counterIds = new AtomicLong(0L);
+
+ private long _peak = 0L;
+ private long _total = 0L;
+ private long _temp = 0L;
+ private long _last = 0L;
+ private long _rate = 0L;
+
+ private long _start;
+
+ private final long _period;
+ private final String _name;
+
+ public StatisticsCounter()
+ {
+ this(COUNTER);
+ }
+
+ public StatisticsCounter(String name)
+ {
+ this(name, DEFAULT_SAMPLE_PERIOD);
+ }
+
+ public StatisticsCounter(String name, long period)
+ {
+ _period = period;
+ _name = name + "-" + + _counterIds.incrementAndGet();
+ reset();
+ }
+
+ public void registerEvent()
+ {
+ registerEvent(1L);
+ }
+
+ public void registerEvent(long value)
+ {
+ registerEvent(value, System.currentTimeMillis());
+ }
+
+ public void registerEvent(long value, long timestamp)
+ {
+ if (DISABLE_STATISTICS)
+ {
+ return;
+ }
+
+ long thisSample = (timestamp / _period);
+ synchronized (this)
+ {
+ if (thisSample > _last)
+ {
+ _last = thisSample;
+ _rate = _temp;
+ _temp = 0L;
+ if (_rate > _peak)
+ {
+ _peak = _rate;
+ }
+ }
+
+ _total += value;
+ _temp += value;
+ }
+ }
+
+ /**
+ * Update the current rate and peak - may reset rate to zero if a new
+ * sample period has started.
+ */
+ private void update()
+ {
+ registerEvent(0L, System.currentTimeMillis());
+ }
+
+ /**
+ * Reset
+ */
+ public void reset()
+ {
+ _log.info("Resetting statistics for counter: " + _name);
+ _peak = 0L;
+ _rate = 0L;
+ _total = 0L;
+ _start = System.currentTimeMillis();
+ _last = _start / _period;
+ }
+
+ public double getPeak()
+ {
+ update();
+ return (double) _peak / ((double) _period / 1000.0d);
+ }
+
+ public double getRate()
+ {
+ update();
+ return (double) _rate / ((double) _period / 1000.0d);
+ }
+
+ public long getTotal()
+ {
+ return _total;
+ }
+
+ public long getStart()
+ {
+ return _start;
+ }
+
+ public Date getStartTime()
+ {
+ return new Date(_start);
+ }
+
+ public String getName()
+ {
+ return _name;
+ }
+
+ public long getPeriod()
+ {
+ return _period;
+ }
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsGatherer.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsGatherer.java?rev=1079043&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsGatherer.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/stats/StatisticsGatherer.java Tue Mar 8 00:14:59 2011
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.qpid.server.stats;
+
+/**
+ * This interface is to be implemented by any broker business object that
+ * wishes to gather statistics about messages delivered through it.
+ *
+ * These statistics are exposed using a separate JMX Mbean interface, which
+ * calls these methods to retrieve the underlying {@link StatisticsCounter}s
+ * and return their attributes. This interface gives a standard way for
+ * parts of the broker to set up and configure statistics generation.
+ * <p>
+ * When creating these objects, there should be a parent/child relationship
+ * between them, such that the lowest level gatherer can record staticics if
+ * enabled, and pass on the notification to the parent object to allow higher
+ * level aggregation. When resetting statistics, this works in the opposite
+ * direction, with higher level gatherers also resetting all of their children.
+ */
+public interface StatisticsGatherer
+{
+ /**
+ * Initialise the statistics gathering for this object.
+ *
+ * This method is responsible for creating any {@link StatisticsCounter}
+ * objects and for determining whether statistics generation should be
+ * enabled, by checking broker and system configuration.
+ *
+ * @see StatisticsCounter#DISABLE_STATISTICS
+ */
+ void initialiseStatistics();
+
+ /**
+ * This method is responsible for registering the receipt of a message
+ * with the counters, and also for passing this notification to any parent
+ * {@link StatisticsGatherer}s. If statistics generation is not enabled,
+ * then this method should simple delegate to the parent gatherer.
+ *
+ * @param messageSize the size in bytes of the delivered message
+ * @param timestamp the time the message was delivered
+ */
+ void registerMessageReceived(long messageSize, long timestamp);
+
+ /**
+ * This method is responsible for registering the delivery of a message
+ * with the counters. Message delivery is recorded by the counter using
+ * the current system time, as opposed to the message timestamp.
+ *
+ * @param messageSize the size in bytes of the delivered message
+ * @see #registerMessageReceived(long, long)
+ */
+ void registerMessageDelivered(long messageSize);
+
+ /**
+ * Gives access to the {@link StatisticsCounter} that is used to count
+ * delivered message statistics.
+ *
+ * @return the {@link StatisticsCounter} that counts delivered messages
+ */
+ StatisticsCounter getMessageDeliveryStatistics();
+
+ /**
+ * Gives access to the {@link StatisticsCounter} that is used to count
+ * received message statistics.
+ *
+ * @return the {@link StatisticsCounter} that counts received messages
+ */
+ StatisticsCounter getMessageReceiptStatistics();
+
+ /**
+ * Gives access to the {@link StatisticsCounter} that is used to count
+ * delivered message size statistics.
+ *
+ * @return the {@link StatisticsCounter} that counts delivered bytes
+ */
+ StatisticsCounter getDataDeliveryStatistics();
+
+ /**
+ * Gives access to the {@link StatisticsCounter} that is used to count
+ * received message size statistics.
+ *
+ * @return the {@link StatisticsCounter} that counts received bytes
+ */
+ StatisticsCounter getDataReceiptStatistics();
+
+ /**
+ * Reset the counters for this, and any child {@link StatisticsGatherer}s.
+ */
+ void resetStatistics();
+
+ /**
+ * Check if this object has statistics generation enabled.
+ *
+ * @return true if statistics generation is enabled
+ */
+ boolean isStatisticsEnabled();
+
+ /**
+ * Enable or disable statistics generation for this object.
+ */
+ void setStatisticsEnabled(boolean enabled);
+}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java Tue Mar 8 00:14:59 2011
@@ -38,6 +38,8 @@ import org.apache.qpid.server.logging.ac
import org.apache.qpid.server.logging.messages.ConnectionMessages;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.virtualhost.VirtualHost;
import org.apache.qpid.transport.Connection;
import org.apache.qpid.transport.ConnectionCloseCode;
@@ -54,6 +56,10 @@ public class ServerConnection extends Co
private AtomicBoolean _logClosed = new AtomicBoolean(false);
private LogActor _actor = GenericActor.getInstance(this);
+ private ApplicationRegistry _registry;
+ private boolean _statisticsEnabled = false;
+ private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
+
public ServerConnection()
{
@@ -121,6 +127,8 @@ public class ServerConnection extends Co
{
_virtualHost = virtualHost;
_virtualHost.getConnectionRegistry().registerConnection(this);
+
+ initialiseStatistics();
}
public void setConnectionConfig(final ConnectionConfig config)
@@ -258,4 +266,73 @@ public class ServerConnection extends Co
}
return sessions;
}
+
+ public void registerMessageDelivered(long messageSize)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesDelivered.registerEvent(1L);
+ _dataDelivered.registerEvent(messageSize);
+ }
+ _virtualHost.registerMessageDelivered(messageSize);
+ }
+
+ public void registerMessageReceived(long messageSize, long timestamp)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesReceived.registerEvent(1L, timestamp);
+ _dataReceived.registerEvent(messageSize, timestamp);
+ }
+ _virtualHost.registerMessageReceived(messageSize, timestamp);
+ }
+
+ public StatisticsCounter getMessageReceiptStatistics()
+ {
+ return _messagesReceived;
+ }
+
+ public StatisticsCounter getDataReceiptStatistics()
+ {
+ return _dataReceived;
+ }
+
+ public StatisticsCounter getMessageDeliveryStatistics()
+ {
+ return _messagesDelivered;
+ }
+
+ public StatisticsCounter getDataDeliveryStatistics()
+ {
+ return _dataDelivered;
+ }
+
+ public void resetStatistics()
+ {
+ _messagesDelivered.reset();
+ _dataDelivered.reset();
+ _messagesReceived.reset();
+ _dataReceived.reset();
+ }
+
+ public void initialiseStatistics()
+ {
+ setStatisticsEnabled(!StatisticsCounter.DISABLE_STATISTICS &&
+ _virtualHost.getApplicationRegistry().getConfiguration().isStatisticsGenerationConnectionsEnabled());
+
+ _messagesDelivered = new StatisticsCounter("messages-delivered-" + getConnectionId());
+ _dataDelivered = new StatisticsCounter("data-delivered-" + getConnectionId());
+ _messagesReceived = new StatisticsCounter("messages-received-" + getConnectionId());
+ _dataReceived = new StatisticsCounter("data-received-" + getConnectionId());
+ }
+
+ public boolean isStatisticsEnabled()
+ {
+ return _statisticsEnabled;
+ }
+
+ public void setStatisticsEnabled(boolean enabled)
+ {
+ _statisticsEnabled = enabled;
+ }
}
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java Tue Mar 8 00:14:59 2011
@@ -20,26 +20,28 @@
*/
package org.apache.qpid.server.transport;
-import org.apache.qpid.transport.*;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.actors.GenericActor;
-import org.apache.qpid.common.ClientProperties;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+
import org.apache.qpid.protocol.ProtocolEngine;
-import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.registry.ApplicationRegistry;
import org.apache.qpid.server.registry.IApplicationRegistry;
+import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.virtualhost.VirtualHost;
-
-import javax.security.sasl.SaslServer;
-import javax.security.sasl.SaslException;
-import java.util.*;
+import org.apache.qpid.transport.*;
public class ServerConnectionDelegate extends ServerDelegate
{
private String _localFQDN;
private final IApplicationRegistry _appRegistry;
-
public ServerConnectionDelegate(IApplicationRegistry appRegistry, String localFQDN)
{
this(new HashMap<String,Object>(Collections.singletonMap("qpid.federation_tag",appRegistry.getBroker().getFederationTag())), Collections.singletonList((Object)"en_US"), appRegistry, localFQDN);
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java Tue Mar 8 00:14:59 2011
@@ -166,6 +166,7 @@ public class ServerSession extends Sessi
public void enqueue(final ServerMessage message, final ArrayList<? extends BaseQueue> queues)
{
+ getConnectionModel().registerMessageReceived(message.getSize(), message.getArrivalTime());
_transaction.enqueue(queues,message, new ServerTransaction.Action()
{
@@ -202,6 +203,7 @@ public class ServerSession extends Sessi
Runnable postIdSettingAction)
{
invoke(xfr, postIdSettingAction);
+ getConnectionModel().registerMessageDelivered(xfr.getBodySize());
}
public void onMessageDispositionChange(MessageTransfer xfr, MessageDispositionChangeListener acceptListener)
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java Tue Mar 8 00:14:59 2011
@@ -20,30 +20,28 @@
*/
package org.apache.qpid.server.virtualhost;
+import java.util.UUID;
+
import org.apache.qpid.common.Closeable;
+import org.apache.qpid.server.binding.BindingFactory;
+import org.apache.qpid.server.configuration.ConfigStore;
+import org.apache.qpid.server.configuration.VirtualHostConfig;
+import org.apache.qpid.server.configuration.VirtualHostConfiguration;
import org.apache.qpid.server.connection.IConnectionRegistry;
+import org.apache.qpid.server.exchange.ExchangeFactory;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
import org.apache.qpid.server.federation.BrokerLink;
-import org.apache.qpid.server.configuration.VirtualHostConfiguration;
-import org.apache.qpid.server.configuration.VirtualHostConfig;
-import org.apache.qpid.server.configuration.ConfigStore;
+import org.apache.qpid.server.management.ManagedObject;
import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.exchange.ExchangeRegistry;
-import org.apache.qpid.server.exchange.ExchangeFactory;
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.TransactionLog;
-import org.apache.qpid.server.store.DurableConfigurationStore;
+import org.apache.qpid.server.registry.IApplicationRegistry;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
-import org.apache.qpid.server.management.ManagedObject;
-import org.apache.qpid.server.registry.IApplicationRegistry;
-import org.apache.qpid.server.binding.BindingFactory;
-
-import java.util.List;
-import java.util.UUID;
-import java.util.TimerTask;
-import java.util.concurrent.FutureTask;
+import org.apache.qpid.server.stats.StatisticsGatherer;
+import org.apache.qpid.server.store.DurableConfigurationStore;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.TransactionLog;
-public interface VirtualHost extends DurableConfigurationStore.Source, VirtualHostConfig, Closeable
+public interface VirtualHost extends DurableConfigurationStore.Source, VirtualHostConfig, Closeable, StatisticsGatherer
{
IConnectionRegistry getConnectionRegistry();
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostImpl.java Tue Mar 8 00:14:59 2011
@@ -73,6 +73,7 @@ import org.apache.qpid.server.registry.I
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager;
+import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.store.ConfigurationRecoveryHandler;
import org.apache.qpid.server.store.DurableConfigurationStore;
import org.apache.qpid.server.store.MessageStore;
@@ -112,6 +113,8 @@ public class VirtualHostImpl implements
private BrokerConfig _broker;
private UUID _id;
+ private boolean _statisticsEnabled = false;
+ private StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
private final long _createTime = System.currentTimeMillis();
private final ConcurrentHashMap<BrokerLink,BrokerLink> _links = new ConcurrentHashMap<BrokerLink, BrokerLink>();
@@ -250,6 +253,8 @@ public class VirtualHostImpl implements
_brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean);
_brokerMBean.register();
initialiseHouseKeeping(hostConfig.getHousekeepingExpiredMessageCheckPeriod());
+
+ initialiseStatistics();
}
private void initialiseHouseKeeping(long period)
@@ -639,6 +644,80 @@ public class VirtualHostImpl implements
{
return _bindingFactory;
}
+
+ public void registerMessageDelivered(long messageSize)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesDelivered.registerEvent(1L);
+ _dataDelivered.registerEvent(messageSize);
+ }
+ _appRegistry.registerMessageDelivered(messageSize);
+ }
+
+ public void registerMessageReceived(long messageSize, long timestamp)
+ {
+ if (isStatisticsEnabled())
+ {
+ _messagesReceived.registerEvent(1L, timestamp);
+ _dataReceived.registerEvent(messageSize, timestamp);
+ }
+ _appRegistry.registerMessageReceived(messageSize, timestamp);
+ }
+
+ public StatisticsCounter getMessageReceiptStatistics()
+ {
+ return _messagesReceived;
+ }
+
+ public StatisticsCounter getDataReceiptStatistics()
+ {
+ return _dataReceived;
+ }
+
+ public StatisticsCounter getMessageDeliveryStatistics()
+ {
+ return _messagesDelivered;
+ }
+
+ public StatisticsCounter getDataDeliveryStatistics()
+ {
+ return _dataDelivered;
+ }
+
+ public void resetStatistics()
+ {
+ _messagesDelivered.reset();
+ _dataDelivered.reset();
+ _messagesReceived.reset();
+ _dataReceived.reset();
+
+ for (AMQConnectionModel connection : _connectionRegistry.getConnections())
+ {
+ connection.resetStatistics();
+ }
+ }
+
+ public void initialiseStatistics()
+ {
+ setStatisticsEnabled(!StatisticsCounter.DISABLE_STATISTICS &&
+ _appRegistry.getConfiguration().isStatisticsGenerationVirtualhostsEnabled());
+
+ _messagesDelivered = new StatisticsCounter("messages-delivered-" + getName());
+ _dataDelivered = new StatisticsCounter("bytes-delivered-" + getName());
+ _messagesReceived = new StatisticsCounter("messages-received-" + getName());
+ _dataReceived = new StatisticsCounter("bytes-received-" + getName());
+ }
+
+ public boolean isStatisticsEnabled()
+ {
+ return _statisticsEnabled;
+ }
+
+ public void setStatisticsEnabled(boolean enabled)
+ {
+ _statisticsEnabled = enabled;
+ }
public void createBrokerConnection(final String transport,
final String host,
Added: qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/stats/StatisticsCounterTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/stats/StatisticsCounterTest.java?rev=1079043&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/stats/StatisticsCounterTest.java (added)
+++ qpid/trunk/qpid/java/broker/src/test/java/org/apache/qpid/server/stats/StatisticsCounterTest.java Tue Mar 8 00:14:59 2011
@@ -0,0 +1,144 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.qpid.server.stats;
+
+import junit.framework.TestCase;
+
+/**
+ * Unit tests for the {@link StatisticsCounter} class.
+ */
+public class StatisticsCounterTest extends TestCase
+{
+ /**
+ * Check that statistics counters are created correctly.
+ */
+ public void testCreate()
+ {
+ long before = System.currentTimeMillis();
+ StatisticsCounter counter = new StatisticsCounter("name", 1234L);
+ long after = System.currentTimeMillis();
+
+ assertTrue(before <= counter.getStart());
+ assertTrue(after >= counter.getStart());
+ assertTrue(counter.getName().startsWith("name-"));
+ assertEquals(1234L, counter.getPeriod());
+ }
+
+ /**
+ * Check that totals add up correctly.
+ */
+ public void testTotal()
+ {
+ StatisticsCounter counter = new StatisticsCounter("test", 1000L);
+ long start = counter.getStart();
+ for (int i = 0; i < 100; i++)
+ {
+ counter.registerEvent(i, start + i);
+ }
+ assertEquals(99 * 50, counter.getTotal()); // cf. Gauss
+ }
+
+ /**
+ * Test totals add up correctly even when messages are delivered
+ * out-of-order.
+ */
+ public void testTotalOutOfOrder()
+ {
+ StatisticsCounter counter = new StatisticsCounter("test", 1000L);
+ long start = counter.getStart();
+ assertEquals(0, counter.getTotal());
+ counter.registerEvent(10, start + 2500);
+ assertEquals(10, counter.getTotal());
+ counter.registerEvent(20, start + 1500);
+ assertEquals(30, counter.getTotal());
+ counter.registerEvent(10, start + 500);
+ assertEquals(40, counter.getTotal());
+ }
+
+ /**
+ * Test that the peak rate is reported correctly.
+ */
+ public void testPeak() throws Exception
+ {
+ StatisticsCounter counter = new StatisticsCounter("test", 1000L);
+ long start = counter.getStart();
+ assertEquals(0.0, counter.getPeak());
+ Thread.sleep(500);
+ counter.registerEvent(1000, start + 500);
+ Thread.sleep(1000);
+ assertEquals(1000.0, counter.getPeak());
+ counter.registerEvent(2000, start + 1500);
+ Thread.sleep(1000);
+ assertEquals(2000.0, counter.getPeak());
+ counter.registerEvent(1000, start + 2500);
+ Thread.sleep(1000);
+ assertEquals(2000.0, counter.getPeak());
+ }
+
+ /**
+ * Test that peak rate is reported correctly for out-of-order messages,
+ * and the total is also unaffected.
+ */
+ public void testPeakOutOfOrder() throws Exception
+ {
+ StatisticsCounter counter = new StatisticsCounter("test", 1000L);
+ long start = counter.getStart();
+ assertEquals(0.0, counter.getPeak());
+ counter.registerEvent(1000, start + 2500);
+ Thread.sleep(1500);
+ assertEquals(0.0, counter.getPeak());
+ counter.registerEvent(2000, start + 1500);
+ Thread.sleep(1000L);
+ assertEquals(0.0, counter.getPeak());
+ counter.registerEvent(1000, start + 500);
+ Thread.sleep(1500);
+ assertEquals(4000.0, counter.getPeak());
+ Thread.sleep(2000);
+ assertEquals(4000.0, counter.getPeak());
+ counter.registerEvent(1000, start + 500);
+ assertEquals(4000.0, counter.getPeak());
+ Thread.sleep(2000);
+ counter.registerEvent(1000);
+ assertEquals(4000.0, counter.getPeak());
+ assertEquals(6000, counter.getTotal());
+ }
+
+ /**
+ * Test the current rate is generated correctly.
+ */
+ public void testRate() throws Exception
+ {
+ StatisticsCounter counter = new StatisticsCounter("test", 1000L);
+ assertEquals(0.0, counter.getRate());
+ Thread.sleep(500);
+ counter.registerEvent(1000);
+ Thread.sleep(1000);
+ assertEquals(1000.0, counter.getRate());
+ counter.registerEvent(2000);
+ Thread.sleep(1000);
+ assertEquals(2000.0, counter.getRate());
+ counter.registerEvent(1000);
+ Thread.sleep(1000);
+ assertEquals(1000.0, counter.getRate());
+ Thread.sleep(1000);
+ assertEquals(0.0, counter.getRate());
+ }
+}
Modified: qpid/trunk/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java (original)
+++ qpid/trunk/qpid/java/broker/src/velocity/java/org/apache/qpid/server/logging/GenerateLogMessages.java Tue Mar 8 00:14:59 2011
@@ -481,7 +481,7 @@ public class GenerateLogMessages
// Only check the text inside the braces '{}'
int typeIndexEnd = parametersString[index].indexOf("}", typeIndex);
String typeString = parametersString[index].substring(typeIndex, typeIndexEnd);
- if (typeString.contains("number"))
+ if (typeString.contains("number") || typeString.contains("choice"))
{
type = "Number";
}
Modified: qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java (original)
+++ qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedBroker.java Tue Mar 8 00:14:59 2011
@@ -36,8 +36,8 @@ import org.apache.qpid.management.common
* The ManagedBroker is the management interface to expose management
* features of the Broker.
*
- * @author Bhupendra Bhardwaj
- * @version 0.1
+ * @version Qpid JMX API 2.1
+ * @since Qpid JMX API 1.3
*/
public interface ManagedBroker
{
@@ -131,4 +131,118 @@ public interface ManagedBroker
impact= MBeanOperationInfo.ACTION)
void deleteQueue(@MBeanOperationParameter(name= ManagedQueue.TYPE, description="Queue Name")String queueName)
throws IOException, JMException, MBeanException;
+
+ /**
+ * Resets all message and data statistics for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanOperation(name="resetStatistics",
+ description="Resets all message and data statistics for the virtual host",
+ impact= MBeanOperationInfo.ACTION)
+ void resetStatistics() throws Exception;
+
+ /**
+ * Peak rate of messages delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakMessageDeliveryRate", description=TYPE + " Peak Message Delivery Rate")
+ double getPeakMessageDeliveryRate();
+
+ /**
+ * Peak rate of bytes delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakDataDeliveryRate", description=TYPE + " Peak Data Delivery Rate")
+ double getPeakDataDeliveryRate();
+
+ /**
+ * Rate of messages delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="MessageDeliveryRate", description=TYPE + " Message Delivery Rate")
+ double getMessageDeliveryRate();
+
+ /**
+ * Rate of bytes delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="DataDeliveryRate", description=TYPE + " Data Delivery Rate")
+ double getDataDeliveryRate();
+
+ /**
+ * Total count of messages delivered for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalMessagesDelivered", description=TYPE + " Total Messages Delivered")
+ long getTotalMessagesDelivered();
+
+ /**
+ * Total count of bytes for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalDataDelivered", description=TYPE + " Total Data Delivered")
+ long getTotalDataDelivered();
+
+ /**
+ * Peak rate of messages received per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakMessageReceiptRate", description=TYPE + " Peak Message Receipt Rate")
+ double getPeakMessageReceiptRate();
+
+ /**
+ * Peak rate of bytes received per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakDataReceiptRate", description=TYPE + " Peak Data Receipt Rate")
+ double getPeakDataReceiptRate();
+
+ /**
+ * Rate of messages received per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="MessageReceiptRate", description=TYPE + " Message Receipt Rate")
+ double getMessageReceiptRate();
+
+ /**
+ * Rate of bytes received per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="DataReceiptRate", description=TYPE + " Data Receipt Rate")
+ double getDataReceiptRate();
+
+ /**
+ * Total count of messages received for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalMessagesReceived", description=TYPE + " Total Messages Received")
+ long getTotalMessagesReceived();
+
+ /**
+ * Total count of bytes received for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalDataReceived", description=TYPE + " Total Data Received")
+ long getTotalDataReceived();
+
+ /**
+ * Is statistics collection enabled for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="StatisticsEnabled", description=TYPE + " Statistics Enabled")
+ boolean isStatisticsEnabled();
}
Modified: qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java (original)
+++ qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedConnection.java Tue Mar 8 00:14:59 2011
@@ -37,8 +37,9 @@ import org.apache.qpid.management.common
/**
* The management interface exposed to allow management of Connections.
- * @author Bhupendra Bhardwaj
- * @version 0.1
+ *
+ * @version Qpid JMX API 2.1
+ * @since Qpid JMX API 1.3
*/
public interface ManagedConnection
{
@@ -145,4 +146,120 @@ public interface ManagedConnection
description="Closes this connection and all related channels",
impact= MBeanOperationInfo.ACTION)
void closeConnection() throws Exception;
+
+ /**
+ * Resets message and data statistics for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanOperation(name="resetStatistics",
+ description="Resets message and data statistics for this connection",
+ impact= MBeanOperationInfo.ACTION)
+ void resetStatistics() throws Exception;
+
+ /**
+ * Peak rate of messages delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakMessageDeliveryRate", description=TYPE + " Peak Message Delivery Rate")
+ double getPeakMessageDeliveryRate();
+
+ /**
+ * Peak rate of bytes delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakDataDeliveryRate", description=TYPE + " Peak Data Delivery Rate")
+ double getPeakDataDeliveryRate();
+
+ /**
+ * Rate of messages delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="MessageDeliveryRate", description=TYPE + " Message Delivery Rate")
+ double getMessageDeliveryRate();
+
+ /**
+ * Rate of bytes delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="DataDeliveryRate", description=TYPE + " Data Delivery Rate")
+ double getDataDeliveryRate();
+
+ /**
+ * Total count of messages delivered for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalMessagesDelivered", description=TYPE + " Total Messages Delivered")
+ long getTotalMessagesDelivered();
+
+ /**
+ * Total count of bytes for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalDataDelivered", description=TYPE + " Total Data Delivered")
+ long getTotalDataDelivered();
+
+ /**
+ * Peak rate of messages received per second for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakMessageReceiptRate", description=TYPE + " Peak Message Receipt Rate")
+ double getPeakMessageReceiptRate();
+
+ /**
+ * Peak rate of bytes received per second for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakDataReceiptRate", description=TYPE + " Peak Data Receipt Rate")
+ double getPeakDataReceiptRate();
+
+ /**
+ * Rate of messages received per second for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="MessageReceiptRate", description=TYPE + " Message Receipt Rate")
+ double getMessageReceiptRate();
+
+ /**
+ * Rate of bytes received per second for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="DataReceiptRate", description=TYPE + " Data Receipt Rate")
+ double getDataReceiptRate();
+
+ /**
+ * Total count of messages received for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalMessagesReceived", description=TYPE + " Total Messages Received")
+ long getTotalMessagesReceived();
+
+ /**
+ * Total count of bytes received for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalDataReceived", description=TYPE + " Total Data Received")
+ long getTotalDataReceived();
+
+ /**
+ * Is statistics collection enabled for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="StatisticsEnabled", description=TYPE + " Statistics Enabled")
+ boolean isStatisticsEnabled();
+
+ void setStatisticsEnabled(boolean enabled);
}
Modified: qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java?rev=1079043&r1=1079042&r2=1079043&view=diff
==============================================================================
--- qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java (original)
+++ qpid/trunk/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java Tue Mar 8 00:14:59 2011
@@ -22,10 +22,15 @@ package org.apache.qpid.management.commo
import java.io.IOException;
+import javax.management.MBeanOperationInfo;
+
import org.apache.qpid.management.common.mbeans.annotations.MBeanAttribute;
+import org.apache.qpid.management.common.mbeans.annotations.MBeanOperation;
/**
* Interface for the ServerInformation MBean
+ *
+ * @version Qpid JMX API 2.1
* @since Qpid JMX API 1.3
*/
public interface ServerInformation
@@ -80,4 +85,118 @@ public interface ServerInformation
@MBeanAttribute(name="ProductVersion",
description = "The product version string")
String getProductVersion() throws IOException;
+
+ /**
+ * Resets all message and data statistics for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanOperation(name="resetStatistics",
+ description="Resets all message and data statistics for the broker",
+ impact= MBeanOperationInfo.ACTION)
+ void resetStatistics() throws Exception;
+
+ /**
+ * Peak rate of messages delivered per second for the virtual host.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakMessageDeliveryRate", description=TYPE + " Peak Message Delivery Rate")
+ double getPeakMessageDeliveryRate();
+
+ /**
+ * Peak rate of bytes delivered per second for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakDataDeliveryRate", description=TYPE + " Peak Data Delivery Rate")
+ double getPeakDataDeliveryRate();
+
+ /**
+ * Rate of messages delivered per second for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="MessageDeliveryRate", description=TYPE + " Message Delivery Rate")
+ double getMessageDeliveryRate();
+
+ /**
+ * Rate of bytes delivered per second for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="DataDeliveryRate", description=TYPE + " Data Delivery Rate")
+ double getDataDeliveryRate();
+
+ /**
+ * Total count of messages delivered for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalMessagesDelivered", description=TYPE + " Total Messages Delivered")
+ long getTotalMessagesDelivered();
+
+ /**
+ * Total count of bytes for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalDataDelivered", description=TYPE + " Total Data Delivered")
+ long getTotalDataDelivered();
+
+ /**
+ * Peak rate of messages received per second for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakMessageReceiptRate", description=TYPE + " Peak Message Receipt Rate")
+ double getPeakMessageReceiptRate();
+
+ /**
+ * Peak rate of bytes received per second for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="PeakDataReceiptRate", description=TYPE + " Peak Data Receipt Rate")
+ double getPeakDataReceiptRate();
+
+ /**
+ * Rate of messages received per second for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="MessageReceiptRate", description=TYPE + " Message Receipt Rate")
+ double getMessageReceiptRate();
+
+ /**
+ * Rate of bytes received per second for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="DataReceiptRate", description=TYPE + " Data Receipt Rate")
+ double getDataReceiptRate();
+
+ /**
+ * Total count of messages received for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalMessagesReceived", description=TYPE + " Total Messages Received")
+ long getTotalMessagesReceived();
+
+ /**
+ * Total count of bytes received for the broker.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="TotalDataReceived", description=TYPE + " Total Data Received")
+ long getTotalDataReceived();
+
+ /**
+ * Is statistics collection enabled for this connection.
+ *
+ * @since Qpid JMX API 2.1
+ */
+ @MBeanAttribute(name="StatisticsEnabled", description=TYPE + " Statistics Enabled")
+ boolean isStatisticsEnabled();
}
Added: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java?rev=1079043&view=auto
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java (added)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java Tue Mar 8 00:14:59 2011
@@ -0,0 +1,102 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.qpid.management.jmx;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.jms.Connection;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.management.common.mbeans.ManagedBroker;
+import org.apache.qpid.management.common.mbeans.ManagedConnection;
+
+/**
+ * Test enabling generation of message statistics on a per-connection basis.
+ */
+public class MessageConnectionStatisticsTest extends MessageStatisticsTestCase
+{
+ public void configureStatistics() throws Exception
+ {
+ // no statistics generation configured
+ }
+
+ /**
+ * Test statistics on a single connection
+ */
+ public void testEnablingStatisticsPerConnection() throws Exception
+ {
+ ManagedBroker vhost = _jmxUtils.getManagedBroker("test");
+
+ sendUsing(_test, 5, 200);
+ Thread.sleep(1000);
+
+ List<String> addresses = new ArrayList<String>();
+ for (ManagedConnection mc : _jmxUtils.getManagedConnections("test"))
+ {
+ assertEquals("Incorrect connection total", 0, mc.getTotalMessagesReceived());
+ assertEquals("Incorrect connection data", 0, mc.getTotalDataReceived());
+ assertFalse("Connection statistics should not be enabled", mc.isStatisticsEnabled());
+
+ addresses.add(mc.getRemoteAddress());
+ }
+ assertEquals("Incorrect vhost total", 0, vhost.getTotalMessagesReceived());
+ assertEquals("Incorrect vhost data", 0, vhost.getTotalDataReceived());
+
+ Connection test = new AMQConnection(_brokerUrl, USER, USER, "clientid", "test");
+ test.start();
+ for (ManagedConnection mc : _jmxUtils.getManagedConnections("test"))
+ {
+ if (addresses.contains(mc.getRemoteAddress()))
+ {
+ continue;
+ }
+ mc.setStatisticsEnabled(true);
+ assertEquals("Incorrect connection total", 0, mc.getTotalMessagesReceived());
+ assertEquals("Incorrect connection data", 0, mc.getTotalDataReceived());
+ }
+
+ sendUsing(test, 5, 200);
+ sendUsing(_test, 5, 200);
+ Thread.sleep(1000);
+
+ for (ManagedConnection mc : _jmxUtils.getManagedConnections("test"))
+ {
+ if (addresses.contains(mc.getRemoteAddress()))
+ {
+ assertEquals("Incorrect connection total", 0, mc.getTotalMessagesReceived());
+ assertEquals("Incorrect connection data", 0, mc.getTotalDataReceived());
+ assertFalse("Connection statistics should not be enabled", mc.isStatisticsEnabled());
+ }
+ else
+ {
+ assertEquals("Incorrect connection total", 5, mc.getTotalMessagesReceived());
+ assertEquals("Incorrect connection data", 1000, mc.getTotalDataReceived());
+ assertTrue("Connection statistics should be enabled", mc.isStatisticsEnabled());
+ }
+ }
+ assertEquals("Incorrect vhost total", 0, vhost.getTotalMessagesReceived());
+ assertEquals("Incorrect vhost data", 0, vhost.getTotalDataReceived());
+ assertFalse("Vhost statistics should not be enabled", vhost.isStatisticsEnabled());
+
+ test.close();
+ }
+}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org