diff options
author | Andrew Donald Kennedy <grkvlt@apache.org> | 2011-03-08 00:14:59 +0000 |
---|---|---|
committer | Andrew Donald Kennedy <grkvlt@apache.org> | 2011-03-08 00:14:59 +0000 |
commit | b387fc19a0dc062e95659ad34360ea942790e49e (patch) | |
tree | ab963e63e33e0500ff93c2065b2f5e00c99a7330 /qpid/java/systests | |
parent | 5c797b1f2ebce8b79f118dc64cd0c1b3b9efd23c (diff) | |
download | qpid-python-b387fc19a0dc062e95659ad34360ea942790e49e.tar.gz |
QPID-2984: Add statistics generation for broker message delivery
Port of QPID-2932 changes from 0.5.x-dev branch to trunk.
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1079043 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/java/systests')
7 files changed, 906 insertions, 0 deletions
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java new file mode 100644 index 0000000000..9839c6e475 --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageConnectionStatisticsTest.java @@ -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(); + } +} diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java new file mode 100644 index 0000000000..df8c6e74cd --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsConfigurationTest.java @@ -0,0 +1,177 @@ +/* + * + * 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 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 MessageStatisticsConfigurationTest extends MessageStatisticsTestCase +{ + public void configureStatistics() throws Exception + { + setConfigurationProperty("statistics.generation.broker", Boolean.toString(getName().contains("Broker"))); + setConfigurationProperty("statistics.generation.virtualhosts", Boolean.toString(getName().contains("Virtualhost"))); + setConfigurationProperty("statistics.generation.connections", Boolean.toString(getName().contains("Connection"))); + } + + /** + * Just broker statistics. + */ + public void testGenerateBrokerStatistics() throws Exception + { + sendUsing(_test, 5, 200); + Thread.sleep(1000); + + 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()); + } + + ManagedBroker vhost = _jmxUtils.getManagedBroker("test"); + assertEquals("Incorrect vhost data", 0, vhost.getTotalMessagesReceived()); + assertEquals("Incorrect vhost data", 0, vhost.getTotalDataReceived()); + assertFalse("Vhost statistics should not be enabled", vhost.isStatisticsEnabled()); + + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total messages", 5, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server total data", 1000, _jmxUtils.getServerInformation().getTotalDataReceived()); + assertTrue("Server statistics should be enabled", _jmxUtils.getServerInformation().isStatisticsEnabled()); + } + } + + /** + * Just virtualhost statistics. + */ + public void testGenerateVirtualhostStatistics() throws Exception + { + sendUsing(_test, 5, 200); + Thread.sleep(1000); + + 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()); + } + + ManagedBroker vhost = _jmxUtils.getManagedBroker("test"); + assertEquals("Incorrect vhost data", 5, vhost.getTotalMessagesReceived()); + assertEquals("Incorrect vhost data", 1000, vhost.getTotalDataReceived()); + assertTrue("Vhost statistics should be enabled", vhost.isStatisticsEnabled()); + + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total messages", 0, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server total data", 0, _jmxUtils.getServerInformation().getTotalDataReceived()); + assertFalse("Server statistics should not be enabled", _jmxUtils.getServerInformation().isStatisticsEnabled()); + } + } + + /** + * Just connection statistics. + */ + public void testGenerateConnectionStatistics() throws Exception + { + sendUsing(_test, 5, 200); + Thread.sleep(1000); + + for (ManagedConnection mc : _jmxUtils.getManagedConnections("test")) + { + assertEquals("Incorrect connection total", 5, mc.getTotalMessagesReceived()); + assertEquals("Incorrect connection data", 1000, mc.getTotalDataReceived()); + assertTrue("Connection statistics should be enabled", mc.isStatisticsEnabled()); + } + + ManagedBroker vhost = _jmxUtils.getManagedBroker("test"); + assertEquals("Incorrect vhost data", 0, vhost.getTotalMessagesReceived()); + assertEquals("Incorrect vhost data", 0, vhost.getTotalDataReceived()); + assertFalse("Vhost statistics should not be enabled", vhost.isStatisticsEnabled()); + + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total messages", 0, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server total data", 0, _jmxUtils.getServerInformation().getTotalDataReceived()); + assertFalse("Server statistics should not be enabled", _jmxUtils.getServerInformation().isStatisticsEnabled()); + } + } + + /** + * Both broker and virtualhost statistics. + */ + public void testGenerateBrokerAndVirtualhostStatistics() throws Exception + { + sendUsing(_test, 5, 200); + Thread.sleep(1000); + + 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()); + } + + ManagedBroker vhost = _jmxUtils.getManagedBroker("test"); + assertEquals("Incorrect vhost data", 5, vhost.getTotalMessagesReceived()); + assertEquals("Incorrect vhost data", 1000, vhost.getTotalDataReceived()); + assertTrue("Vhost statistics should be enabled", vhost.isStatisticsEnabled()); + + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total messages", 5, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server total data", 1000, _jmxUtils.getServerInformation().getTotalDataReceived()); + assertTrue("Server statistics should be enabled", _jmxUtils.getServerInformation().isStatisticsEnabled()); + } + } + + /** + * Broker, virtualhost and connection statistics. + */ + public void testGenerateBrokerVirtualhostAndConnectionStatistics() throws Exception + { + sendUsing(_test, 5, 200); + Thread.sleep(1000); + + for (ManagedConnection mc : _jmxUtils.getManagedConnections("test")) + { + assertEquals("Incorrect connection total", 5, mc.getTotalMessagesReceived()); + assertEquals("Incorrect connection data", 1000, mc.getTotalDataReceived()); + assertTrue("Connection statistics should be enabled", mc.isStatisticsEnabled()); + } + + ManagedBroker vhost = _jmxUtils.getManagedBroker("test"); + assertEquals("Incorrect vhost data", 5, vhost.getTotalMessagesReceived()); + assertEquals("Incorrect vhost data", 1000, vhost.getTotalDataReceived()); + assertTrue("Vhost statistics should be enabled", vhost.isStatisticsEnabled()); + + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total messages", 5, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server total data", 1000, _jmxUtils.getServerInformation().getTotalDataReceived()); + assertTrue("Server statistics should be enabled", _jmxUtils.getServerInformation().isStatisticsEnabled()); + } + } +} diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsDeliveryTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsDeliveryTest.java new file mode 100644 index 0000000000..e657856d0e --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsDeliveryTest.java @@ -0,0 +1,110 @@ +/* + * + * 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 javax.jms.Message; +import javax.jms.MessageConsumer; +import javax.jms.Session; + +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.management.common.mbeans.ManagedBroker; +import org.apache.qpid.management.common.mbeans.ManagedConnection; + +/** + * Test statistics for delivery and receipt. + */ +public class MessageStatisticsDeliveryTest extends MessageStatisticsTestCase +{ + public void configureStatistics() throws Exception + { + setConfigurationProperty("statistics.generation.broker", "true"); + setConfigurationProperty("statistics.generation.virtualhosts", "true"); + setConfigurationProperty("statistics.generation.connections", "true"); + } + + public void testDeliveryAndReceiptStatistics() 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 delivery total", 0, mc.getTotalMessagesDelivered()); + assertEquals("Incorrect connection delivery data", 0, mc.getTotalDataDelivered()); + assertEquals("Incorrect connection receipt total", 5, mc.getTotalMessagesReceived()); + assertEquals("Incorrect connection receipt data", 1000, mc.getTotalDataReceived()); + + addresses.add(mc.getRemoteAddress()); + } + + assertEquals("Incorrect vhost delivery total", 0, vhost.getTotalMessagesDelivered()); + assertEquals("Incorrect vhost delivery data", 0, vhost.getTotalDataDelivered()); + assertEquals("Incorrect vhost receipt total", 5, vhost.getTotalMessagesReceived()); + assertEquals("Incorrect vhost receipt data", 1000, vhost.getTotalDataReceived()); + + Connection test = new AMQConnection(_brokerUrl, USER, USER, "clientid", "test"); + test.start(); + receiveUsing(test, 5); + + for (ManagedConnection mc : _jmxUtils.getManagedConnections("test")) + { + if (addresses.contains(mc.getRemoteAddress())) + { + assertEquals("Incorrect connection delivery total", 0, mc.getTotalMessagesDelivered()); + assertEquals("Incorrect connection delivery data", 0, mc.getTotalDataDelivered()); + assertEquals("Incorrect connection receipt total", 5, mc.getTotalMessagesReceived()); + assertEquals("Incorrect connection receipt data", 1000, mc.getTotalDataReceived()); + } + else + { + assertEquals("Incorrect connection delivery total", 5, mc.getTotalMessagesDelivered()); + assertEquals("Incorrect connection delivery data", 1000, mc.getTotalDataDelivered()); + assertEquals("Incorrect connection receipt total", 0, mc.getTotalMessagesReceived()); + assertEquals("Incorrect connection receipt data", 0, mc.getTotalDataReceived()); + } + } + assertEquals("Incorrect vhost delivery total", 5, vhost.getTotalMessagesDelivered()); + assertEquals("Incorrect vhost delivery data", 1000, vhost.getTotalDataDelivered()); + assertEquals("Incorrect vhost receipt total", 5, vhost.getTotalMessagesReceived()); + assertEquals("Incorrect vhost receipt data", 1000, vhost.getTotalDataReceived()); + + test.close(); + } + + protected void receiveUsing(Connection con, int number) throws Exception + { + Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE); + createQueue(session); + MessageConsumer consumer = session.createConsumer(_queue); + for (int i = 0; i < number; i++) + { + Message msg = consumer.receive(100); + assertNotNull("Message " + i + " was not received", msg); + } + } +} diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsReportingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsReportingTest.java new file mode 100644 index 0000000000..180440c0d6 --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsReportingTest.java @@ -0,0 +1,90 @@ +/* + * + * 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.List; + +import org.apache.qpid.util.LogMonitor; + +/** + * Test generation of message statistics reporting. + */ +public class MessageStatisticsReportingTest extends MessageStatisticsTestCase +{ + protected LogMonitor _monitor; + + public void configureStatistics() throws Exception + { + setConfigurationProperty("statistics.generation.broker", "true"); + setConfigurationProperty("statistics.generation.virtualhosts", "true"); + + if (getName().equals("testEnabledStatisticsReporting")) + { + setConfigurationProperty("statistics.reporting.period", "10"); + } + + _monitor = new LogMonitor(_outputFile); + } + + /** + * Test enabling reporting. + */ + public void testEnabledStatisticsReporting() throws Exception + { + sendUsing(_test, 10, 100); + sendUsing(_dev, 20, 100); + sendUsing(_local, 15, 100); + + Thread.sleep(10 * 1000); // 15s + + List<String> brokerStatsData = _monitor.findMatches("BRK-1008"); + List<String> brokerStatsMessages = _monitor.findMatches("BRK-1009"); + List<String> vhostStatsData = _monitor.findMatches("VHT-1003"); + List<String> vhostStatsMessages = _monitor.findMatches("VHT-1004"); + + assertEquals("Incorrect number of broker data stats log messages", 2, brokerStatsData.size()); + assertEquals("Incorrect number of broker message stats log messages", 2, brokerStatsMessages.size()); + assertEquals("Incorrect number of virtualhost data stats log messages", 6, vhostStatsData.size()); + assertEquals("Incorrect number of virtualhost message stats log messages", 6, vhostStatsMessages.size()); + } + + /** + * Test not enabling reporting. + */ + public void testNotEnabledStatisticsReporting() throws Exception + { + sendUsing(_test, 10, 100); + sendUsing(_dev, 20, 100); + sendUsing(_local, 15, 100); + + Thread.sleep(10 * 1000); // 15s + + List<String> brokerStatsData = _monitor.findMatches("BRK-1008"); + List<String> brokerStatsMessages = _monitor.findMatches("BRK-1009"); + List<String> vhostStatsData = _monitor.findMatches("VHT-1003"); + List<String> vhostStatsMessages = _monitor.findMatches("VHT-1004"); + + assertEquals("Incorrect number of broker data stats log messages", 0, brokerStatsData.size()); + assertEquals("Incorrect number of broker message stats log messages", 0, brokerStatsMessages.size()); + assertEquals("Incorrect number of virtualhost data stats log messages", 0, vhostStatsData.size()); + assertEquals("Incorrect number of virtualhost message stats log messages", 0, vhostStatsMessages.size()); + } +} diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java new file mode 100644 index 0000000000..50ca51b18a --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTest.java @@ -0,0 +1,233 @@ +/* + * + * 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 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 generation of message statistics. + */ +public class MessageStatisticsTest extends MessageStatisticsTestCase +{ + public void configureStatistics() throws Exception + { + setConfigurationProperty("statistics.generation.broker", "true"); + setConfigurationProperty("statistics.generation.virtualhosts", "true"); + setConfigurationProperty("statistics.generation.connections", "true"); + } + + /** + * Test message totals. + */ + public void testMessageTotals() throws Exception + { + sendUsing(_test, 10, 100); + sendUsing(_dev, 20, 100); + sendUsing(_local, 5, 100); + sendUsing(_local, 5, 100); + sendUsing(_local, 5, 100); + Thread.sleep(2000); + + ManagedBroker test = _jmxUtils.getManagedBroker("test"); + ManagedBroker dev = _jmxUtils.getManagedBroker("development"); + ManagedBroker local = _jmxUtils.getManagedBroker("localhost"); + + if (!isBroker010()) + { + long total = 0; + long data = 0; + for (ManagedConnection mc : _jmxUtils.getAllManagedConnections()) + { + total += mc.getTotalMessagesReceived(); + data += mc.getTotalDataReceived(); + } + assertEquals("Incorrect connection total", 45, total); + assertEquals("Incorrect connection data", 4500, data); + } + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total", 45, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server data", 4500, _jmxUtils.getServerInformation().getTotalDataReceived()); + } + + if (!isBroker010()) + { + long testTotal = 0; + long testData = 0; + for (ManagedConnection mc : _jmxUtils.getManagedConnections("test")) + { + testTotal += mc.getTotalMessagesReceived(); + testData += mc.getTotalDataReceived(); + } + assertEquals("Incorrect test connection total", 10, testTotal); + assertEquals("Incorrect test connection data", 1000, testData); + } + assertEquals("Incorrect test vhost total", 10, test.getTotalMessagesReceived()); + assertEquals("Incorrect test vhost data", 1000, test.getTotalDataReceived()); + + if (!isBroker010()) + { + long devTotal = 0; + long devData = 0; + for (ManagedConnection mc : _jmxUtils.getManagedConnections("development")) + { + devTotal += mc.getTotalMessagesReceived(); + devData += mc.getTotalDataReceived(); + } + assertEquals("Incorrect test connection total", 20, devTotal); + assertEquals("Incorrect test connection data", 2000, devData); + } + assertEquals("Incorrect development total", 20, dev.getTotalMessagesReceived()); + assertEquals("Incorrect development data", 2000, dev.getTotalDataReceived()); + + if (!isBroker010()) + { + long localTotal = 0; + long localData = 0; + for (ManagedConnection mc : _jmxUtils.getManagedConnections("localhost")) + { + localTotal += mc.getTotalMessagesReceived(); + localData += mc.getTotalDataReceived(); + } + assertEquals("Incorrect test connection total", 15, localTotal); + assertEquals("Incorrect test connection data", 1500, localData); + } + assertEquals("Incorrect localhost total", 15, local.getTotalMessagesReceived()); + assertEquals("Incorrect localhost data", 1500, local.getTotalDataReceived()); + } + + /** + * Test message totals when a connection is closed. + */ + public void testMessageTotalsWithClosedConnections() throws Exception + { + Connection temp = new AMQConnection(_brokerUrl, USER, USER, "clientid", "test"); + temp.start(); + + sendUsing(_test, 10, 100); + sendUsing(temp, 10, 100); + sendUsing(_test, 10, 100); + Thread.sleep(2000); + + temp.close(); + + ManagedBroker test = _jmxUtils.getManagedBroker("test"); + + if (!isBroker010()) + { + long total = 0; + long data = 0; + for (ManagedConnection mc : _jmxUtils.getAllManagedConnections()) + { + total += mc.getTotalMessagesReceived(); + data += mc.getTotalDataReceived(); + } + assertEquals("Incorrect active connection total", 20, total); + assertEquals("Incorrect active connection data", 2000, data); + } + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total", 30, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server data", 3000, _jmxUtils.getServerInformation().getTotalDataReceived()); + } + + if (!isBroker010()) + { + long testTotal = 0; + long testData = 0; + for (ManagedConnection mc : _jmxUtils.getManagedConnections("test")) + { + testTotal += mc.getTotalMessagesReceived(); + testData += mc.getTotalDataReceived(); + } + assertEquals("Incorrect test active connection total", 20, testTotal); + assertEquals("Incorrect test active connection data", 20 * 100, testData); + } + assertEquals("Incorrect test vhost total", 30, test.getTotalMessagesReceived()); + assertEquals("Incorrect test vhost data", 30 * 100, test.getTotalDataReceived()); + } + + /** + * Test message peak rate generation. + */ + public void testMessagePeakRates() throws Exception + { + sendUsing(_test, 2, 10); + Thread.sleep(10000); + sendUsing(_dev, 4, 10); + Thread.sleep(10000); + + ManagedBroker test = _jmxUtils.getManagedBroker("test"); + ManagedBroker dev = _jmxUtils.getManagedBroker("development"); + + assertApprox("Incorrect test vhost peak messages", 0.2d, 1.0d, test.getPeakMessageReceiptRate()); + assertApprox("Incorrect test vhost peak data", 0.2d, 10.0d, test.getPeakDataReceiptRate()); + assertApprox("Incorrect dev vhost peak messages", 0.2d, 2.0d, dev.getPeakMessageReceiptRate()); + assertApprox("Incorrect dev vhost peak data", 0.2d, 20.0d, dev.getPeakDataReceiptRate()); + + if (!_broker.equals(VM)) + { + assertApprox("Incorrect server peak messages", 0.2d, 2.0d, _jmxUtils.getServerInformation().getPeakMessageReceiptRate()); + assertApprox("Incorrect server peak data", 0.2d, 20.0d, _jmxUtils.getServerInformation().getPeakDataReceiptRate()); + } + } + + /** + * Test message totals when a vhost has its statistics reset + */ + public void testMessageTotalVhostReset() throws Exception + { + sendUsing(_test, 10, 10); + sendUsing(_dev, 10, 10); + Thread.sleep(2000); + + ManagedBroker test = _jmxUtils.getManagedBroker("test"); + ManagedBroker dev = _jmxUtils.getManagedBroker("development"); + + assertEquals("Incorrect test vhost total messages", 10, test.getTotalMessagesReceived()); + assertEquals("Incorrect test vhost total data", 100, test.getTotalDataReceived()); + assertEquals("Incorrect dev vhost total messages", 10, dev.getTotalMessagesReceived()); + assertEquals("Incorrect dev vhost total data", 100, dev.getTotalDataReceived()); + + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total messages", 20, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server total data", 200, _jmxUtils.getServerInformation().getTotalDataReceived()); + } + + test.resetStatistics(); + + assertEquals("Incorrect test vhost total messages", 0, test.getTotalMessagesReceived()); + assertEquals("Incorrect test vhost total data", 0, test.getTotalDataReceived()); + assertEquals("Incorrect dev vhost total messages", 10, dev.getTotalMessagesReceived()); + assertEquals("Incorrect dev vhost total data", 100, dev.getTotalDataReceived()); + + if (!_broker.equals(VM)) + { + assertEquals("Incorrect server total messages", 20, _jmxUtils.getServerInformation().getTotalMessagesReceived()); + assertEquals("Incorrect server total data", 200, _jmxUtils.getServerInformation().getTotalDataReceived()); + } + } +} diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTestCase.java b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTestCase.java new file mode 100644 index 0000000000..a5b3aa283c --- /dev/null +++ b/qpid/java/systests/src/main/java/org/apache/qpid/management/jmx/MessageStatisticsTestCase.java @@ -0,0 +1,128 @@ +/* + * + * 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 javax.jms.Connection; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.apache.qpid.AMQException; +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQDestination; +import org.apache.qpid.client.AMQQueue; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.exchange.ExchangeDefaults; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.test.utils.JMXTestUtils; +import org.apache.qpid.test.utils.QpidBrokerTestCase; + +/** + * Test generation of message statistics. + */ +public abstract class MessageStatisticsTestCase extends QpidBrokerTestCase +{ + protected static final String USER = "admin"; + + protected JMXTestUtils _jmxUtils; + protected Connection _test, _dev, _local; + protected String _queueName = "statistics"; + protected Destination _queue; + protected String _brokerUrl; + + @Override + public void setUp() throws Exception + { + _jmxUtils = new JMXTestUtils(this, USER, USER); + _jmxUtils.setUp(); + + configureStatistics(); + + super.setUp(); + + _brokerUrl = getBroker().toString(); + _test = new AMQConnection(_brokerUrl, USER, USER, "clientid", "test"); + _dev = new AMQConnection(_brokerUrl, USER, USER, "clientid", "development"); + _local = new AMQConnection(_brokerUrl, USER, USER, "clientid", "localhost"); + + _test.start(); + _dev.start(); + _local.start(); + + _jmxUtils.open(); + } + + protected void createQueue(Session session) throws AMQException, JMSException + { + _queue = new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, _queueName); + if (!((AMQSession<?,?>) session).isQueueBound((AMQDestination) _queue)) + { + ((AMQSession<?,?>) session).createQueue(new AMQShortString(_queueName), false, true, false, null); + ((AMQSession<?,?>) session).declareAndBind((AMQDestination) new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, _queueName)); + } + } + + + @Override + public void tearDown() throws Exception + { + _jmxUtils.close(); + + _test.close(); + _dev.close(); + _local.close(); + + super.tearDown(); + } + + /** + * Configure statistics generation properties on the broker. + */ + public abstract void configureStatistics() throws Exception; + + protected void sendUsing(Connection con, int number, int size) throws Exception + { + Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE); + createQueue(session); + MessageProducer producer = session.createProducer(_queue); + String content = new String(new byte[size]); + TextMessage msg = session.createTextMessage(content); + for (int i = 0; i < number; i++) + { + producer.send(msg); + } + } + + /** + * Asserts that the actual value is within the expected value plus or + * minus the given error. + */ + public void assertApprox(String message, double error, double expected, double actual) + { + double min = expected * (1.0d - error); + double max = expected * (1.0d + error); + String assertion = String.format("%s: expected %f +/- %d%%, actual %f", + message, expected, (int) (error * 100.0d), actual); + assertTrue(assertion, actual > min && actual < max); + } +} diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/JMXTestUtils.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/JMXTestUtils.java index 3fece2130a..9f6963643a 100644 --- a/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/JMXTestUtils.java +++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/JMXTestUtils.java @@ -21,6 +21,8 @@ package org.apache.qpid.test.utils; import java.io.IOException; +import java.util.ArrayList; +import java.util.List; import java.util.Set; import javax.management.JMException; @@ -31,14 +33,18 @@ import javax.management.ObjectName; import javax.management.MalformedObjectNameException; import javax.management.remote.JMXConnector; +import junit.framework.TestCase; + import org.apache.commons.configuration.ConfigurationException; import org.apache.qpid.commands.objects.AllObjects; import org.apache.qpid.management.common.JMXConnnectionFactory; import org.apache.qpid.management.common.mbeans.ManagedBroker; +import org.apache.qpid.management.common.mbeans.ManagedConnection; import org.apache.qpid.management.common.mbeans.ManagedExchange; import org.apache.qpid.management.common.mbeans.LoggingManagement; import org.apache.qpid.management.common.mbeans.ConfigurationManagement; import org.apache.qpid.management.common.mbeans.ManagedQueue; +import org.apache.qpid.management.common.mbeans.ServerInformation; import org.apache.qpid.management.common.mbeans.UserManagement; /** @@ -326,6 +332,16 @@ public class JMXTestUtils return MBeanServerInvocationHandler.newProxyInstance(_mbsc, objectName, managedClass, false); } + public <T> List<T> getManagedObjectList(Class<T> managedClass, Set<ObjectName> objectNames) + { + List<T> objects = new ArrayList<T>(); + for (ObjectName name : objectNames) + { + objects.add(getManagedObject(managedClass, name)); + } + return objects; + } + public ManagedBroker getManagedBroker(String virtualHost) { return getManagedObject(ManagedBroker.class, getVirtualHostManagerObjectName(virtualHost)); @@ -360,4 +376,54 @@ public class JMXTestUtils ObjectName objectName = new ObjectName("org.apache.qpid:type=UserManagement,name=UserManagement"); return getManagedObject(UserManagement.class, objectName); } + + /** + * Retrive {@link ServerInformation} JMX MBean. + */ + public ServerInformation getServerInformation() + { + // Get the name of the test manager + AllObjects allObject = new AllObjects(_mbsc); + allObject.querystring = "org.apache.qpid:type=ServerInformation,name=ServerInformation,*"; + + Set<ObjectName> objectNames = allObject.returnObjects(); + + TestCase.assertNotNull("Null ObjectName Set returned", objectNames); + TestCase.assertEquals("Incorrect number of objects returned", 1, objectNames.size()); + + // We have verified we have only one value in objectNames so return it + return getManagedObject(ServerInformation.class, objectNames.iterator().next()); + } + + /** + * Retrive all {@link ManagedConnection} objects. + */ + public List<ManagedConnection> getAllManagedConnections() + { + // Get the name of the test manager + AllObjects allObject = new AllObjects(_mbsc); + allObject.querystring = "org.apache.qpid:type=VirtualHost.Connection,VirtualHost=*,name=*"; + + Set<ObjectName> objectNames = allObject.returnObjects(); + + TestCase.assertNotNull("Null ObjectName Set returned", objectNames); + + return getManagedObjectList(ManagedConnection.class, objectNames); + } + + /** + * Retrive all {@link ManagedConnection} objects for a particular virtual host. + */ + public List<ManagedConnection> getManagedConnections(String vhost) + { + // Get the name of the test manager + AllObjects allObject = new AllObjects(_mbsc); + allObject.querystring = "org.apache.qpid:type=VirtualHost.Connection,VirtualHost=" + ObjectName.quote(vhost) + ",name=*"; + + Set<ObjectName> objectNames = allObject.returnObjects(); + + TestCase.assertNotNull("Null ObjectName Set returned", objectNames); + + return getManagedObjectList(ManagedConnection.class, objectNames); + } } |