You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/06/28 11:15:39 UTC
svn commit: r1354874 [12/14] - in /qpid/branches/java-config-and-management:
./ qpid/ qpid/bin/ qpid/cpp/ qpid/cpp/bindings/ qpid/cpp/bindings/qmf/ruby/
qpid/cpp/bindings/qpid/python/ qpid/cpp/bindings/qpid/ruby/
qpid/cpp/bindings/qpid/ruby/features/ q...
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/clustered_replication_test
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/clustered_replication_test?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/clustered_replication_test (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/clustered_replication_test Thu Jun 28 09:14:52 2012
@@ -8,9 +8,9 @@
# 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
@@ -34,11 +34,11 @@ stop_brokers() {
if [[ $PRIMARY1 ]] ; then
$QPIDD_EXEC --no-module-dir -q --port $PRIMARY1
unset PRIMARY1
- fi
+ fi
if [[ $PRIMARY2 ]] ; then
$QPIDD_EXEC --no-module-dir -q --port $PRIMARY2
unset PRIMARY2
- fi
+ fi
if [[ $DR1 ]] ; then
$QPIDD_EXEC --no-module-dir -q --port $DR1
unset DR1
@@ -50,7 +50,8 @@ stop_brokers() {
}
if test -d $PYTHON_DIR; then
- . $srcdir/ais_check
+ . cpg_check.sh
+ cpg_enabled || exit 0
#todo: these cluster names need to be unique to prevent clashes
PRIMARY_CLUSTER=PRIMARY_$(hostname)_$$
@@ -89,20 +90,20 @@ if test -d $PYTHON_DIR; then
#send more messages to primary
for i in `seq 11 20`; do echo Message$i; done | ./sender --port $PRIMARY1 --send-eos 1
- #wait for replication events to all be processed:
+ #wait for replication events to all be processed:
echo Waiting for replication to complete
echo Done | ./sender --port $PRIMARY1 --routing-key control-queue --send-eos 1
./receiver --queue control-queue --port $DR1 > /dev/null
#verify contents of test queue on dr cluster:
- echo Verifying...
+ echo Verifying...
./receiver --port $DR2 > repl.out.tmp
for i in `seq 6 20`; do echo Message$i; done | diff repl.out.tmp - || FAIL=1
if [[ $FAIL ]]; then
echo Clustered replication test failed: expectations not met!
exit 1
- else
+ else
echo Clustered replication test passed
rm -f repl*.tmp
fi
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/federated_cluster_test
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/federated_cluster_test?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/federated_cluster_test (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/federated_cluster_test Thu Jun 28 09:14:52 2012
@@ -8,9 +8,9 @@
# 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
@@ -35,7 +35,7 @@ stop_brokers() {
if [[ $BROKER_A ]] ; then
../qpidd --no-module-dir -q --port $BROKER_A
unset BROKER_A
- fi
+ fi
if [[ $NODE_1 ]] ; then
../qpidd --no-module-dir -q --port $NODE_1
unset NODE_1
@@ -92,7 +92,7 @@ run_test_pull_to_cluster_two_consumers()
wait
sort -g -k 2 fed1.out.tmp fed2.out.tmp > fed.out.tmp
diff fed.in.tmp fed.out.tmp || fail "federated link to cluster failed: expectations not met!"
-
+
rm -f fed*.tmp #cleanup
}
@@ -106,7 +106,7 @@ run_test_pull_to_cluster() {
#verify all messages are received
diff fed.in.tmp fed.out.tmp || fail "federated link to cluster failed: expectations not met!"
-
+
rm -f fed*.tmp #cleanup
}
@@ -121,22 +121,23 @@ run_test_pull_from_cluster() {
#verify all messages are received
wait
diff fed.in.tmp fed.out.tmp || fail "federated link from cluster failed: expectations not met!"
-
+
rm -f fed*.tmp #cleanup
}
if test -d ${PYTHON_DIR}; then
- . $srcdir/ais_check
+ . cpg_check.sh
+ cpg_enabled || exit 0
rm -f fed*.tmp #cleanup any files left from previous run
start_brokers
echo "brokers started"
setup
echo "setup completed"
- run_test_pull_to_cluster_two_consumers
+ run_test_pull_to_cluster_two_consumers
echo "federated link to cluster verified"
- run_test_pull_from_cluster
+ run_test_pull_from_cluster
echo "federated link from cluster verified"
if [[ $TEST_NODE_FAILURE ]] ; then
#kill first cluster node and retest
@@ -146,7 +147,7 @@ if test -d ${PYTHON_DIR}; then
echo "retesting..."
run_test_pull_to_cluster
echo "federated link to cluster verified"
- run_test_pull_from_cluster
- echo "federated link from cluster verified"
+ run_test_pull_from_cluster
+ echo "federated link from cluster verified"
fi
fi
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/federation.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/federation.py?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/federation.py (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/federation.py Thu Jun 28 09:14:52 2012
@@ -23,6 +23,7 @@ from qpid.testlib import TestBase010
from qpid.datatypes import Message
from qpid.queue import Empty
from qpid.util import URL
+import qpid.messaging
from time import sleep, time
@@ -94,6 +95,11 @@ class FederationTests(TestBase010):
break
self._brokers.append(_b)
+ # add a new-style messaging connection to each broker
+ for _b in self._brokers:
+ _b.connection = qpid.messaging.Connection(_b.url)
+ _b.connection.open()
+
def _teardown_brokers(self):
""" Un-does _setup_brokers()
"""
@@ -103,7 +109,7 @@ class FederationTests(TestBase010):
if not _b.client_session.error():
_b.client_session.close(timeout=10)
_b.client_conn.close(timeout=10)
-
+ _b.connection.close()
def test_bridge_create_and_close(self):
self.startQmf();
@@ -127,18 +133,28 @@ class FederationTests(TestBase010):
self.verify_cleanup()
def test_pull_from_exchange(self):
+ """ This test uses an alternative method to manage links and bridges
+ via the broker object.
+ """
session = self.session
-
+
self.startQmf()
qmf = self.qmf
broker = qmf.getObjects(_class="broker")[0]
- result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp")
- self.assertEqual(result.status, 0, result)
- link = qmf.getObjects(_class="link")[0]
- result = link.bridge(False, "amq.direct", "amq.fanout", "my-key", "", "", False, False, False, 0)
+ # create link
+ link_args = {"host":self.remote_host(), "port":self.remote_port(), "durable":False,
+ "authMechanism":"PLAIN", "username":"guest", "password":"guest",
+ "transport":"tcp"}
+ result = broker.create("link", "test-link-1", link_args, False)
self.assertEqual(result.status, 0, result)
+ link = qmf.getObjects(_class="link")[0]
+ # create bridge
+ bridge_args = {"link":"test-link-1", "src":"amq.direct", "dest":"amq.fanout",
+ "key":"my-key"}
+ result = broker.create("bridge", "test-bridge-1", bridge_args, False);
+ self.assertEqual(result.status, 0, result)
bridge = qmf.getObjects(_class="bridge")[0]
#setup queue to receive messages from local broker
@@ -164,9 +180,11 @@ class FederationTests(TestBase010):
self.fail("Got unexpected message in queue: " + extra.body)
except Empty: None
- result = bridge.close()
+
+ result = broker.delete("bridge", "test-bridge-1", {})
self.assertEqual(result.status, 0, result)
- result = link.close()
+
+ result = broker.delete("link", "test-link-1", {})
self.assertEqual(result.status, 0, result)
self.verify_cleanup()
@@ -376,6 +394,9 @@ class FederationTests(TestBase010):
for i in range(1, 11):
try:
msg = queue.get(timeout=5)
+ mp = msg.get("message_properties").application_headers
+ self.assertEqual(mp.__class__, dict)
+ self.assertEqual(mp['x-qpid.trace'], 'REMOTE') # check that the federation-tag override works
self.assertEqual("Message %d" % i, msg.body)
except Empty:
self.fail("Failed to find expected message containing 'Message %d'" % i)
@@ -2153,3 +2174,433 @@ class FederationTests(TestBase010):
self.verify_cleanup()
+ def test_multilink_direct(self):
+ """ Verify that two distinct links can be created between federated
+ brokers.
+ """
+ self.startQmf()
+ qmf = self.qmf
+ self._setup_brokers()
+ src_broker = self._brokers[0]
+ dst_broker = self._brokers[1]
+
+ # create a direct exchange on each broker
+ for _b in [src_broker, dst_broker]:
+ _b.client_session.exchange_declare(exchange="fedX.direct", type="direct")
+ self.assertEqual(_b.client_session.exchange_query(name="fedX.direct").type,
+ "direct", "exchange_declare failed!")
+
+ # create destination queues
+ for _q in [("HiQ", "high"), ("MedQ", "medium"), ("LoQ", "low")]:
+ dst_broker.client_session.queue_declare(queue=_q[0], auto_delete=True)
+ dst_broker.client_session.exchange_bind(queue=_q[0], exchange="fedX.direct", binding_key=_q[1])
+
+ # create two connections, one for high priority traffic
+ for _q in ["HiPri", "Traffic"]:
+ result = dst_broker.qmf_object.create("link", _q,
+ {"host":src_broker.host,
+ "port":src_broker.port},
+ False)
+ self.assertEqual(result.status, 0);
+
+ links = qmf.getObjects(_broker=dst_broker.qmf_broker, _class="link")
+ for _l in links:
+ if _l.name == "HiPri":
+ hi_link = _l
+ elif _l.name == "Traffic":
+ data_link = _l
+ else:
+ self.fail("Unexpected Link found: " + _l.name)
+
+ # now create a route for messages sent with key "high" to use the
+ # hi_link
+ result = dst_broker.qmf_object.create("bridge", "HiPriBridge",
+ {"link":hi_link.name,
+ "src":"fedX.direct",
+ "dest":"fedX.direct",
+ "key":"high"}, False)
+ self.assertEqual(result.status, 0);
+
+
+ # create routes for the "medium" and "low" links to use the normal
+ # data_link
+ for _b in [("MediumBridge", "medium"), ("LowBridge", "low")]:
+ result = dst_broker.qmf_object.create("bridge", _b[0],
+ {"link":data_link.name,
+ "src":"fedX.direct",
+ "dest":"fedX.direct",
+ "key":_b[1]}, False)
+ self.assertEqual(result.status, 0);
+
+ # now wait for the links to become operational
+ for _l in [hi_link, data_link]:
+ expire_time = time() + 30
+ while _l.state != "Operational" and time() < expire_time:
+ _l.update()
+ self.assertEqual(_l.state, "Operational", "Link failed to become operational")
+
+ # verify each link uses a different connection
+ self.assertNotEqual(hi_link.connectionRef, data_link.connectionRef,
+ "Different links using the same connection")
+
+ hi_conn = qmf.getObjects(_broker=dst_broker.qmf_broker,
+ _objectId=hi_link.connectionRef)[0]
+ data_conn = qmf.getObjects(_broker=dst_broker.qmf_broker,
+ _objectId=data_link.connectionRef)[0]
+
+
+ # send hi data, verify only goes over hi link
+
+ r_ssn = dst_broker.connection.session()
+ hi_receiver = r_ssn.receiver("HiQ");
+ med_receiver = r_ssn.receiver("MedQ");
+ low_receiver = r_ssn.receiver("LoQ");
+
+ for _c in [hi_conn, data_conn]:
+ _c.update()
+ self.assertEqual(_c.msgsToClient, 0, "Unexpected messages received")
+
+ s_ssn = src_broker.connection.session()
+ hi_sender = s_ssn.sender("fedX.direct/high")
+ med_sender = s_ssn.sender("fedX.direct/medium")
+ low_sender = s_ssn.sender("fedX.direct/low")
+
+ try:
+ hi_sender.send(qpid.messaging.Message(content="hi priority"))
+ msg = hi_receiver.fetch(timeout=10)
+ r_ssn.acknowledge()
+ self.assertEqual(msg.content, "hi priority");
+ except:
+ self.fail("Hi Pri message failure")
+
+ hi_conn.update()
+ data_conn.update()
+ self.assertEqual(hi_conn.msgsToClient, 1, "Expected 1 hi pri message")
+ self.assertEqual(data_conn.msgsToClient, 0, "Expected 0 data messages")
+
+ # send low and medium, verify it does not go over hi link
+
+ try:
+ med_sender.send(qpid.messaging.Message(content="medium priority"))
+ msg = med_receiver.fetch(timeout=10)
+ r_ssn.acknowledge()
+ self.assertEqual(msg.content, "medium priority");
+ except:
+ self.fail("Medium Pri message failure")
+
+ hi_conn.update()
+ data_conn.update()
+ self.assertEqual(hi_conn.msgsToClient, 1, "Expected 1 hi pri message")
+ self.assertEqual(data_conn.msgsToClient, 1, "Expected 1 data message")
+
+ try:
+ low_sender.send(qpid.messaging.Message(content="low priority"))
+ msg = low_receiver.fetch(timeout=10)
+ r_ssn.acknowledge()
+ self.assertEqual(msg.content, "low priority");
+ except:
+ self.fail("Low Pri message failure")
+
+ hi_conn.update()
+ data_conn.update()
+ self.assertEqual(hi_conn.msgsToClient, 1, "Expected 1 hi pri message")
+ self.assertEqual(data_conn.msgsToClient, 2, "Expected 2 data message")
+
+ # cleanup
+
+ for _b in qmf.getObjects(_broker=dst_broker.qmf_broker,_class="bridge"):
+ result = _b.close()
+ self.assertEqual(result.status, 0)
+
+ for _l in qmf.getObjects(_broker=dst_broker.qmf_broker,_class="link"):
+ result = _l.close()
+ self.assertEqual(result.status, 0)
+
+ for _q in [("HiQ", "high"), ("MedQ", "medium"), ("LoQ", "low")]:
+ dst_broker.client_session.exchange_unbind(queue=_q[0], exchange="fedX.direct", binding_key=_q[1])
+ dst_broker.client_session.queue_delete(queue=_q[0])
+
+ for _b in [src_broker, dst_broker]:
+ _b.client_session.exchange_delete(exchange="fedX.direct")
+
+ self._teardown_brokers()
+
+ self.verify_cleanup()
+
+
+ def test_multilink_shared_queue(self):
+ """ Verify that two distinct links can be created between federated
+ brokers.
+ """
+ self.startQmf()
+ qmf = self.qmf
+ self._setup_brokers()
+ src_broker = self._brokers[0]
+ dst_broker = self._brokers[1]
+
+ # create a topic exchange on the destination broker
+ dst_broker.client_session.exchange_declare(exchange="fedX.topic", type="topic")
+ self.assertEqual(dst_broker.client_session.exchange_query(name="fedX.topic").type,
+ "topic", "exchange_declare failed!")
+
+ # create a destination queue
+ dst_broker.client_session.queue_declare(queue="destQ", auto_delete=True)
+ dst_broker.client_session.exchange_bind(queue="destQ", exchange="fedX.topic", binding_key="srcQ")
+
+ # create a single source queue
+ src_broker.client_session.queue_declare(queue="srcQ", auto_delete=True)
+
+ # create two connections
+ for _q in ["Link1", "Link2"]:
+ result = dst_broker.qmf_object.create("link", _q,
+ {"host":src_broker.host,
+ "port":src_broker.port},
+ False)
+ self.assertEqual(result.status, 0);
+
+ links = qmf.getObjects(_broker=dst_broker.qmf_broker, _class="link")
+ self.assertEqual(len(links), 2)
+
+ # now create two "parallel" queue routes from the source queue to the
+ # destination exchange.
+ result = dst_broker.qmf_object.create("bridge", "Bridge1",
+ {"link":"Link1",
+ "src":"srcQ",
+ "dest":"fedX.topic",
+ "srcIsQueue": True},
+ False)
+ self.assertEqual(result.status, 0);
+ result = dst_broker.qmf_object.create("bridge", "Bridge2",
+ {"link":"Link2",
+ "src":"srcQ",
+ "dest":"fedX.topic",
+ "srcIsQueue": True},
+ False)
+ self.assertEqual(result.status, 0);
+
+
+ # now wait for the links to become operational
+ for _l in links:
+ expire_time = time() + 30
+ while _l.state != "Operational" and time() < expire_time:
+ _l.update()
+ self.assertEqual(_l.state, "Operational", "Link failed to become operational")
+
+ # verify each link uses a different connection
+ self.assertNotEqual(links[0].connectionRef, links[1].connectionRef,
+ "Different links using the same connection")
+
+ conn1 = qmf.getObjects(_broker=dst_broker.qmf_broker,
+ _objectId=links[0].connectionRef)[0]
+ conn2 = qmf.getObjects(_broker=dst_broker.qmf_broker,
+ _objectId=links[1].connectionRef)[0]
+
+ # verify messages sent to the queue are pulled by each connection
+
+ r_ssn = dst_broker.connection.session()
+ receiver = r_ssn.receiver("destQ");
+
+ for _c in [conn1, conn2]:
+ _c.update()
+ self.assertEqual(_c.msgsToClient, 0, "Unexpected messages received")
+
+ s_ssn = src_broker.connection.session()
+ sender = s_ssn.sender("srcQ")
+
+ try:
+ for x in range(5):
+ sender.send(qpid.messaging.Message(content="hello"))
+ for x in range(5):
+ msg = receiver.fetch(timeout=10)
+ self.assertEqual(msg.content, "hello");
+ r_ssn.acknowledge()
+ except:
+ self.fail("Message failure")
+
+ # expect messages to be split over each connection.
+ conn1.update()
+ conn2.update()
+ self.assertNotEqual(conn1.msgsToClient, 0, "No messages sent")
+ self.assertNotEqual(conn2.msgsToClient, 0, "No messages sent")
+ self.assertEqual(conn2.msgsToClient + conn1.msgsToClient, 5,
+ "Expected 5 messages total")
+
+ for _b in qmf.getObjects(_broker=dst_broker.qmf_broker,_class="bridge"):
+ result = _b.close()
+ self.assertEqual(result.status, 0)
+
+ for _l in qmf.getObjects(_broker=dst_broker.qmf_broker,_class="link"):
+ result = _l.close()
+ self.assertEqual(result.status, 0)
+
+ dst_broker.client_session.exchange_unbind(queue="destQ", exchange="fedX.topic", binding_key="srcQ")
+ dst_broker.client_session.exchange_delete(exchange="fedX.topic")
+
+ self._teardown_brokers()
+
+ self.verify_cleanup()
+
+
+ def test_dynamic_direct_shared_queue(self):
+ """
+ Route Topology:
+
+ +<--- B1
+ B0 <---+<--- B2
+ +<--- B3
+ """
+ session = self.session
+
+ # create the federation
+
+ self.startQmf()
+ qmf = self.qmf
+
+ self._setup_brokers()
+
+ # create direct exchange on each broker, and retrieve the corresponding
+ # management object for that exchange
+
+ exchanges=[]
+ for _b in self._brokers:
+ _b.client_session.exchange_declare(exchange="fedX.direct", type="direct")
+ self.assertEqual(_b.client_session.exchange_query(name="fedX.direct").type,
+ "direct", "exchange_declare failed!")
+ # pull the exchange out of qmf...
+ retries = 0
+ my_exchange = None
+ while my_exchange is None:
+ objs = qmf.getObjects(_broker=_b.qmf_broker, _class="exchange")
+ for ooo in objs:
+ if ooo.name == "fedX.direct":
+ my_exchange = ooo
+ break
+ if my_exchange is None:
+ retries += 1
+ self.failIfEqual(retries, 10,
+ "QMF failed to find new exchange!")
+ sleep(1)
+ exchanges.append(my_exchange)
+
+ self.assertEqual(len(exchanges), len(self._brokers), "Exchange creation failed!")
+
+ # Create 2 links per each source broker (1,2,3) to the downstream
+ # broker 0:
+ for _b in range(1,4):
+ for _l in ["dynamic", "queue"]:
+ result = self._brokers[0].qmf_object.create( "link",
+ "Link-%d-%s" % (_b, _l),
+ {"host":self._brokers[_b].host,
+ "port":self._brokers[_b].port}, False)
+ self.assertEqual(result.status, 0)
+
+ # create queue on source brokers for use by the dynamic route
+ self._brokers[_b].client_session.queue_declare(queue="fedSrcQ", exclusive=False, auto_delete=True)
+
+ for _l in range(1,4):
+ # for each dynamic link, create a dynamic bridge for the "fedX.direct"
+ # exchanges, using the fedSrcQ on each upstream source broker
+ result = self._brokers[0].qmf_object.create("bridge",
+ "Bridge-%d-dynamic" % _l,
+ {"link":"Link-%d-dynamic" % _l,
+ "src":"fedX.direct",
+ "dest":"fedX.direct",
+ "dynamic":True,
+ "queue":"fedSrcQ"}, False)
+ self.assertEqual(result.status, 0)
+
+ # create a queue route that shares the queue used by the dynamic route
+ result = self._brokers[0].qmf_object.create("bridge",
+ "Bridge-%d-queue" % _l,
+ {"link":"Link-%d-queue" % _l,
+ "src":"fedSrcQ",
+ "dest":"fedX.direct",
+ "srcIsQueue":True}, False)
+ self.assertEqual(result.status, 0)
+
+
+ # wait for the inter-broker links to become operational
+ retries = 0
+ operational = False
+ while not operational:
+ operational = True
+ for _l in qmf.getObjects(_class="link"):
+ #print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.state)))
+ if _l.state != "Operational":
+ operational = False
+ if not operational:
+ retries += 1
+ self.failIfEqual(retries, 10,
+ "inter-broker links failed to become operational.")
+ sleep(1)
+
+ # @todo - There is no way to determine when the bridge objects become
+ # active. Hopefully, this is long enough!
+ sleep(6)
+
+ # create a queue on B0, bound to "spudboy"
+ self._brokers[0].client_session.queue_declare(queue="DestQ", exclusive=True, auto_delete=True)
+ self._brokers[0].client_session.exchange_bind(queue="DestQ", exchange="fedX.direct", binding_key="spudboy")
+
+ # subscribe to messages arriving on B2's queue
+ self.subscribe(self._brokers[0].client_session, queue="DestQ", destination="f1")
+ queue = self._brokers[0].client_session.incoming("f1")
+
+ # wait until the binding key has propagated to each broker
+
+ binding_counts = [1, 1, 1, 1]
+ self.assertEqual(len(binding_counts), len(exchanges), "Update Test!")
+ for i in range(3,-1,-1):
+ retries = 0
+ exchanges[i].update()
+ while exchanges[i].bindingCount < binding_counts[i]:
+ retries += 1
+ self.failIfEqual(retries, 10,
+ "binding failed to propagate to broker %d"
+ % i)
+ sleep(3)
+ exchanges[i].update()
+
+ for _b in range(1,4):
+ # send 3 msgs from each source broker
+ for i in range(3):
+ dp = self._brokers[_b].client_session.delivery_properties(routing_key="spudboy")
+ self._brokers[_b].client_session.message_transfer(destination="fedX.direct", message=Message(dp, "Message_drp %d" % i))
+
+ # get exactly 9 (3 per broker) on B0
+ for i in range(9):
+ msg = queue.get(timeout=5)
+
+ try:
+ extra = queue.get(timeout=1)
+ self.fail("Got unexpected message in queue: " + extra.body)
+ except Empty: None
+
+ # verify that messages went across every link
+ for _l in qmf.getObjects(_broker=self._brokers[0].qmf_broker,
+ _class="link"):
+ for _c in qmf.getObjects(_broker=self._brokers[0].qmf_broker,
+ _objectId=_l.connectionRef):
+ self.assertNotEqual(_c.msgsToClient, 0, "Messages did not pass over link as expected.")
+
+ # cleanup
+
+ self._brokers[0].client_session.exchange_unbind(queue="DestQ", exchange="fedX.direct", binding_key="spudboy")
+ self._brokers[0].client_session.message_cancel(destination="f1")
+ self._brokers[0].client_session.queue_delete(queue="DestQ")
+
+ for _b in qmf.getObjects(_class="bridge"):
+ result = _b.close()
+ self.assertEqual(result.status, 0)
+
+ for _l in qmf.getObjects(_class="link"):
+ result = _l.close()
+ self.assertEqual(result.status, 0)
+
+ for _b in self._brokers:
+ _b.client_session.exchange_delete(exchange="fedX.direct")
+
+ self._teardown_brokers()
+
+ self.verify_cleanup()
+
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/ha_tests.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/ha_tests.py?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/ha_tests.py (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/ha_tests.py Thu Jun 28 09:14:52 2012
@@ -19,50 +19,121 @@
#
import os, signal, sys, time, imp, re, subprocess, glob, random, logging, shutil, math
-from qpid.messaging import Message, NotFound, ConnectionError, Connection
+import traceback
+from qpid.messaging import Message, NotFound, ConnectionError, ReceiverError, Connection, Timeout
from qpid.datatypes import uuid4
from brokertest import *
from threading import Thread, Lock, Condition
-from logging import getLogger, WARN, ERROR, DEBUG
+from logging import getLogger, WARN, ERROR, DEBUG, INFO
from qpidtoollibs import BrokerAgent
+from uuid import UUID
-log = getLogger("qpid.ha-tests")
+log = getLogger(__name__)
+
+class QmfAgent(object):
+ """Access to a QMF broker agent."""
+ def __init__(self, address):
+ self._connection = Connection.establish(
+ address, client_properties={"qpid.ha-admin":1})
+ self._agent = BrokerAgent(self._connection)
+ assert self._agent.getHaBroker(), "HA module not loaded in broker at: %s"%(address)
+
+ def __getattr__(self, name):
+ a = getattr(self._agent, name)
+ return a
class HaBroker(Broker):
- def __init__(self, test, args=[], broker_url=None, ha_cluster=True, **kwargs):
+ """Start a broker with HA enabled"""
+ def __init__(self, test, args=[], brokers_url=None, ha_cluster=True,
+ ha_replicate="all", **kwargs):
assert BrokerTest.ha_lib, "Cannot locate HA plug-in"
args = copy(args)
- args.extend(["--load-module", BrokerTest.ha_lib,
- # FIXME aconway 2012-02-13: workaround slow link failover.
- "--link-maintenace-interval=0.1",
- "--ha-cluster=%s"%ha_cluster])
- if broker_url: args.extend([ "--ha-brokers", broker_url ])
+ args += ["--load-module", BrokerTest.ha_lib,
+ "--log-enable=debug+:ha::",
+ # FIXME aconway 2012-02-13: workaround slow link failover.
+ "--link-maintenace-interval=0.1",
+ "--ha-cluster=%s"%ha_cluster]
+ if ha_replicate is not None:
+ args += [ "--ha-replicate=%s"%ha_replicate ]
+ if brokers_url: args += [ "--ha-brokers-url", brokers_url ]
Broker.__init__(self, test, args, **kwargs)
- self.commands=os.getenv("PYTHON_COMMANDS")
- assert os.path.isdir(self.commands)
+ self.qpid_ha_path=os.path.join(os.getenv("PYTHON_COMMANDS"), "qpid-ha")
+ assert os.path.exists(self.qpid_ha_path)
+ self.qpid_config_path=os.path.join(os.getenv("PYTHON_COMMANDS"), "qpid-config")
+ assert os.path.exists(self.qpid_config_path)
+ getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover.
+ self.qpid_ha_script=import_script(self.qpid_ha_path)
+ self._agent = None
- def promote(self):
- assert os.system("%s/qpid-ha promote -b %s"%(self.commands, self.host_port())) == 0
+ def __str__(self): return Broker.__str__(self)
- def set_client_url(self, url):
- assert os.system(
- "%s/qpid-ha set --public-brokers=%s -b %s"%(self.commands, url,self.host_port())) == 0
-
- def set_broker_url(self, url):
- assert os.system(
- "%s/qpid-ha set --brokers=%s -b %s"%(self.commands, url, self.host_port())) == 0
-
- def replicate(self, from_broker, queue):
- assert os.system(
- "%s/qpid-ha replicate -b %s %s %s"%(self.commands, self.host_port(), from_broker, queue)) == 0
+ def qpid_ha(self, args): self.qpid_ha_script.main(["", "-b", self.host_port()]+args)
+
+ def promote(self): self.qpid_ha(["promote"])
+ def set_client_url(self, url): self.qpid_ha(["set", "--public-url", url])
+ def set_brokers_url(self, url): self.qpid_ha(["set", "--brokers-url", url])
+ def replicate(self, from_broker, queue): self.qpid_ha(["replicate", from_broker, queue])
+ def agent(self):
+ if not self._agent: self._agent = QmfAgent(self.host_port())
+ return self._agent
+
+ def ha_status(self):
+ hb = self.agent().getHaBroker()
+ hb.update()
+ return hb.status
+
+ def wait_status(self, status):
+ def try_get_status():
+ # Ignore ConnectionError, the broker may not be up yet.
+ try: return self.ha_status() == status;
+ except ConnectionError: return False
+ assert retry(try_get_status, timeout=20), "%s status != %r"%(self, status)
+
+ # FIXME aconway 2012-05-01: do direct python call to qpid-config code.
+ def qpid_config(self, args):
+ assert subprocess.call(
+ [self.qpid_config_path, "--broker", self.host_port()]+args) == 0
def config_replicate(self, from_broker, queue):
- assert os.system(
- "%s/qpid-config --broker=%s add queue --start-replica %s %s"%(self.commands, self.host_port(), from_broker, queue)) == 0
+ self.qpid_config(["add", "queue", "--start-replica", from_broker, queue])
def config_declare(self, queue, replication):
- assert os.system(
- "%s/qpid-config --broker=%s add queue %s --replicate %s"%(self.commands, self.host_port(), queue, replication)) == 0
+ self.qpid_config(["add", "queue", queue, "--replicate", replication])
+
+ def connect_admin(self, **kwargs):
+ return Broker.connect(self, client_properties={"qpid.ha-admin":1}, **kwargs)
+
+ def wait_backup(self, address):
+ """Wait for address to become valid on a backup broker."""
+ bs = self.connect_admin().session()
+ try: wait_address(bs, address)
+ finally: bs.connection.close()
+
+ def assert_browse(self, queue, expected, **kwargs):
+ """Verify queue contents by browsing."""
+ bs = self.connect().session()
+ try:
+ wait_address(bs, queue)
+ assert_browse_retry(bs, queue, expected, **kwargs)
+ finally: bs.connection.close()
+
+ def assert_browse_backup(self, queue, expected, **kwargs):
+ """Combines wait_backup and assert_browse_retry."""
+ bs = self.connect_admin().session()
+ try:
+ wait_address(bs, queue)
+ assert_browse_retry(bs, queue, expected, **kwargs)
+ finally: bs.connection.close()
+
+ def assert_connect_fail(self):
+ try:
+ self.connect()
+ self.test.fail("Expected ConnectionError")
+ except ConnectionError: pass
+
+ def try_connect(self):
+ try: return self.connect()
+ except ConnectionError: return None
class HaCluster(object):
_cluster_count = 0
@@ -70,77 +141,82 @@ class HaCluster(object):
def __init__(self, test, n, **kwargs):
"""Start a cluster of n brokers"""
self.test = test
- self._brokers = [ HaBroker(test, name="broker%s-%s"%(HaCluster._cluster_count, i), **kwargs) for i in xrange(n)]
+ self.kwargs = kwargs
+ self._brokers = []
+ self.id = HaCluster._cluster_count
+ self.broker_id = 0
HaCluster._cluster_count += 1
+ for i in xrange(n): self.start(False)
+ self.update_urls()
self[0].promote()
+
+ def next_name(self):
+ name="cluster%s-%s"%(self.id, self.broker_id)
+ self.broker_id += 1
+ return name
+
+ def start(self, update_urls=True):
+ """Start a new broker in the cluster"""
+ b = HaBroker(self.test, name=self.next_name(), **self.kwargs)
+ self._brokers.append(b)
+ if update_urls: self.update_urls()
+ return b
+
+ def update_urls(self):
self.url = ",".join([b.host_port() for b in self])
- for b in self: b.set_broker_url(self.url)
+ for b in self: b.set_brokers_url(self.url)
def connect(self, i):
"""Connect with reconnect_urls"""
return self[i].connect(reconnect=True, reconnect_urls=self.url.split(","))
- def kill(self, i):
+ def kill(self, i, promote_next=True):
"""Kill broker i, promote broker i+1"""
- self[i].kill()
self[i].expect = EXPECT_EXIT_FAIL
- self[(i+1) % len(self)].promote()
+ self[i].kill()
+ if promote_next: self[(i+1) % len(self)].promote()
+
+ def restart(self, i):
+ """Start a broker with the same name and data directory. It will get
+ a separate log file: foo.n.log"""
+ b = self._brokers[i]
+ self._brokers[i] = HaBroker(
+ self.test, name=b.name, port=b.port(), brokers_url=self.url,
+ **self.kwargs)
- def bounce(self, i):
+ def bounce(self, i, promote_next=True):
"""Stop and restart a broker in a cluster."""
- self.kill(i)
- b = self[i]
- self._brokers[i] = HaBroker(self.test, name=b.name, port=b.port(), broker_url=self.url)
+ self.kill(i, promote_next)
+ self.restart(i)
# Behave like a list of brokers.
def __len__(self): return len(self._brokers)
def __getitem__(self,index): return self._brokers[index]
def __iter__(self): return self._brokers.__iter__()
-
-def qr_node(value="all"): return "node:{x-declare:{arguments:{'qpid.replicate':%s}}}" % value
-
-class ShortTests(BrokerTest):
- """Short HA functionality tests."""
-
- # Wait for an address to become valid.
- def wait(self, session, address):
- def check():
- try:
- session.sender(address)
- return True
- except NotFound: return False
- assert retry(check), "Timed out waiting for address %s"%(address)
-
- # Wait for address to become valid on a backup broker.
- def wait_backup(self, backup, address):
- bs = self.connect_admin(backup).session()
- self.wait(bs, address)
- bs.connection.close()
-
- # Combines wait_backup and assert_browse_retry
- def assert_browse_backup(self, backup, queue, expected, **kwargs):
- bs = self.connect_admin(backup).session()
- self.wait(bs, queue)
- self.assert_browse_retry(bs, queue, expected, **kwargs)
- bs.connection.close()
-
- def assert_missing(self, session, address):
+def wait_address(session, address):
+ """Wait for an address to become valid."""
+ def check():
try:
- session.receiver(address)
- self.fail("Should not have been replicated: %s"%(address))
- except NotFound: pass
+ session.sender(address)
+ return True
+ except NotFound: return False
+ assert retry(check), "Timed out waiting for address %s"%(address)
+
+def valid_address(session, address):
+ """Test if an address is valid"""
+ try:
+ session.receiver(address)
+ return True
+ except NotFound: return False
- def connect_admin(self, backup, **kwargs):
- """Connect to a backup broker as an admin connection"""
- return backup.connect(client_properties={"qpid.ha-admin":1}, **kwargs)
+class ReplicationTests(BrokerTest):
+ """Correctness tests for HA replication."""
def test_replication(self):
"""Test basic replication of configuration and messages before and
after backup has connected"""
- getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover.
-
def queue(name, replicate):
return "%s;{create:always,node:{x-declare:{arguments:{'qpid.replicate':%s}}}}"%(name, replicate)
@@ -170,9 +246,8 @@ class ShortTests(BrokerTest):
def verify(b, prefix, p):
"""Verify setup was replicated to backup b"""
-
# Wait for configuration to replicate.
- self.wait(b, prefix+"x");
+ wait_address(b, prefix+"x");
self.assert_browse_retry(b, prefix+"q1", ["b", "1", "4"])
self.assertEqual(p.receiver(prefix+"q1").fetch(timeout=0).content, "b")
@@ -180,7 +255,7 @@ class ShortTests(BrokerTest):
self.assert_browse_retry(b, prefix+"q1", ["1", "4"])
self.assert_browse_retry(b, prefix+"q2", []) # configuration only
- self.assert_missing(b, prefix+"q3")
+ assert not valid_address(b, prefix+"q3")
b.sender(prefix+"e1").send(Message(prefix+"e1")) # Verify binds with replicate=all
self.assert_browse_retry(b, prefix+"q1", ["1", "4", prefix+"e1"])
b.sender(prefix+"e2").send(Message(prefix+"e2")) # Verify binds with replicate=configuration
@@ -195,17 +270,17 @@ class ShortTests(BrokerTest):
# Create config, send messages before starting the backup, to test catch-up replication.
setup(p, "1", primary)
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
# Create config, send messages after starting the backup, to test steady-state replication.
setup(p, "2", primary)
# Verify the data on the backup
- b = self.connect_admin(backup).session()
+ b = backup.connect_admin().session()
verify(b, "1", p)
verify(b, "2", p)
# Test a series of messages, enqueue all then dequeue all.
s = p.sender(queue("foo","all"))
- self.wait(b, "foo")
+ wait_address(b, "foo")
msgs = [str(i) for i in range(10)]
for m in msgs: s.send(Message(m))
self.assert_browse_retry(p, "foo", msgs)
@@ -227,103 +302,91 @@ class ShortTests(BrokerTest):
self.assert_browse_retry(b, "foo", msgs[i+1:])
def test_sync(self):
- def queue(name, replicate):
- return "%s;{create:always,%s}"%(name, qr_node(replicate))
primary = HaBroker(self, name="primary")
primary.promote()
p = primary.connect().session()
- s = p.sender(queue("q","all"))
+ s = p.sender("q;{create:always}")
for m in [str(i) for i in range(0,10)]: s.send(m)
s.sync()
- backup1 = HaBroker(self, name="backup1", broker_url=primary.host_port())
+ backup1 = HaBroker(self, name="backup1", brokers_url=primary.host_port())
for m in [str(i) for i in range(10,20)]: s.send(m)
s.sync()
- backup2 = HaBroker(self, name="backup2", broker_url=primary.host_port())
+ backup2 = HaBroker(self, name="backup2", brokers_url=primary.host_port())
for m in [str(i) for i in range(20,30)]: s.send(m)
s.sync()
msgs = [str(i) for i in range(30)]
- b1 = self.connect_admin(backup1).session()
- self.wait(b1, "q");
+ b1 = backup1.connect_admin().session()
+ wait_address(b1, "q");
self.assert_browse_retry(b1, "q", msgs)
- b2 = self.connect_admin(backup2).session()
- self.wait(b2, "q");
+ b2 = backup2.connect_admin().session()
+ wait_address(b2, "q");
self.assert_browse_retry(b2, "q", msgs)
def test_send_receive(self):
"""Verify sequence numbers of messages sent by qpid-send"""
- getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover.
- primary = HaBroker(self, name="primary")
- primary.promote()
- backup1 = HaBroker(self, name="backup1", broker_url=primary.host_port())
- backup2 = HaBroker(self, name="backup2", broker_url=primary.host_port())
+ brokers = HaCluster(self, 3)
sender = self.popen(
["qpid-send",
- "--broker", primary.host_port(),
- "--address", "q;{create:always,%s}"%(qr_node("all")),
+ "--broker", brokers[0].host_port(),
+ "--address", "q;{create:always}",
"--messages=1000",
"--content-string=x"
])
receiver = self.popen(
["qpid-receive",
- "--broker", primary.host_port(),
- "--address", "q;{create:always,%s}"%(qr_node("all")),
+ "--broker", brokers[0].host_port(),
+ "--address", "q;{create:always}",
"--messages=990",
"--timeout=10"
])
- try:
- self.assertEqual(sender.wait(), 0)
- self.assertEqual(receiver.wait(), 0)
- expect = [long(i) for i in range(991, 1001)]
- sn = lambda m: m.properties["sn"]
- self.assert_browse_retry(self.connect_admin(backup1).session(), "q", expect, transform=sn)
- self.assert_browse_retry(self.connect_admin(backup2).session(), "q", expect, transform=sn)
- except:
- print self.browse(primary.connect().session(), "q", transform=sn)
- print self.browse(self.connect_admin(backup1).session(), "q", transform=sn)
- print self.browse(self.connect_admin(backup2).session(), "q", transform=sn)
- raise
+ self.assertEqual(sender.wait(), 0)
+ self.assertEqual(receiver.wait(), 0)
+ expect = [long(i) for i in range(991, 1001)]
+ sn = lambda m: m.properties["sn"]
+ brokers[1].assert_browse_backup("q", expect, transform=sn)
+ brokers[2].assert_browse_backup("q", expect, transform=sn)
def test_failover_python(self):
"""Verify that backups rejects connections and that fail-over works in python client"""
- getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover.
primary = HaBroker(self, name="primary", expect=EXPECT_EXIT_FAIL)
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
# Check that backup rejects normal connections
try:
backup.connect().session()
self.fail("Expected connection to backup to fail")
except ConnectionError: pass
# Check that admin connections are allowed to backup.
- self.connect_admin(backup).close()
+ backup.connect_admin().close()
# Test discovery: should connect to primary after reject by backup
c = backup.connect(reconnect_urls=[primary.host_port(), backup.host_port()], reconnect=True)
s = c.session()
- sender = s.sender("q;{create:always,%s}"%(qr_node()))
- self.wait_backup(backup, "q")
+ sender = s.sender("q;{create:always}")
+ backup.wait_backup("q")
sender.send("foo")
primary.kill()
assert retry(lambda: not is_running(primary.pid))
backup.promote()
- self.assert_browse_retry(s, "q", ["foo"])
+ sender.send("bar")
+ self.assert_browse_retry(s, "q", ["foo", "bar"])
c.close()
def test_failover_cpp(self):
"""Verify that failover works in the C++ client."""
primary = HaBroker(self, name="primary", expect=EXPECT_EXIT_FAIL)
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
url="%s,%s"%(primary.host_port(), backup.host_port())
- primary.connect().session().sender("q;{create:always,%s}"%(qr_node()))
- self.wait_backup(backup, "q")
+ primary.connect().session().sender("q;{create:always}")
+ backup.wait_backup("q")
sender = NumberedSender(primary, url=url, queue="q", failover_updates = False)
receiver = NumberedReceiver(primary, url=url, queue="q", failover_updates = False)
receiver.start()
sender.start()
- self.wait_backup(backup, "q")
+ backup.wait_backup("q")
assert retry(lambda: receiver.received > 10) # Wait for some messages to get thru
primary.kill()
@@ -337,122 +400,123 @@ class ShortTests(BrokerTest):
def test_backup_failover(self):
"""Verify that a backup broker fails over and recovers queue state"""
brokers = HaCluster(self, 3)
- brokers[0].connect().session().sender(
- "q;{create:always,%s}"%(qr_node())).send("a")
- for b in brokers[1:]: self.assert_browse_backup(b, "q", ["a"])
+ brokers[0].connect().session().sender("q;{create:always}").send("a")
+ for b in brokers[1:]: b.assert_browse_backup("q", ["a"], msg=b)
brokers[0].expect = EXPECT_EXIT_FAIL
brokers.kill(0)
brokers[1].connect().session().sender("q").send("b")
- self.assert_browse_backup(brokers[2], "q", ["a","b"])
+ brokers[2].assert_browse_backup("q", ["a","b"])
s = brokers[1].connect().session()
self.assertEqual("a", s.receiver("q").fetch().content)
s.acknowledge()
- self.assert_browse_backup(brokers[2], "q", ["b"])
+ brokers[2].assert_browse_backup("q", ["b"])
def test_qpid_config_replication(self):
"""Set up replication via qpid-config"""
brokers = HaCluster(self,2)
brokers[0].config_declare("q","all")
brokers[0].connect().session().sender("q").send("foo")
- self.assert_browse_backup(brokers[1], "q", ["foo"])
+ brokers[1].assert_browse_backup("q", ["foo"])
def test_standalone_queue_replica(self):
"""Test replication of individual queues outside of cluster mode"""
getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover.
- primary = HaBroker(self, name="primary", ha_cluster=False, args=["--log-enable=debug+"])
+ primary = HaBroker(self, name="primary", ha_cluster=False)
pc = primary.connect()
ps = pc.session().sender("q;{create:always}")
pr = pc.session().receiver("q;{create:always}")
- backup = HaBroker(self, name="backup", ha_cluster=False, args=["--log-enable=debug+"])
+ backup = HaBroker(self, name="backup", ha_cluster=False)
br = backup.connect().session().receiver("q;{create:always}")
# Set up replication with qpid-ha
backup.replicate(primary.host_port(), "q")
ps.send("a")
- self.assert_browse_backup(backup, "q", ["a"])
+ backup.assert_browse_backup("q", ["a"])
ps.send("b")
- self.assert_browse_backup(backup, "q", ["a", "b"])
+ backup.assert_browse_backup("q", ["a", "b"])
self.assertEqual("a", pr.fetch().content)
pr.session.acknowledge()
- self.assert_browse_backup(backup, "q", ["b"])
+ backup.assert_browse_backup("q", ["b"])
# Set up replication with qpid-config
ps2 = pc.session().sender("q2;{create:always}")
backup.config_replicate(primary.host_port(), "q2");
ps2.send("x")
- self.assert_browse_backup(backup, "q2", ["x"])
+ backup.assert_browse_backup("q2", ["x"])
def test_queue_replica_failover(self):
"""Test individual queue replication from a cluster to a standalone backup broker, verify it fails over."""
- getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover.
cluster = HaCluster(self, 2)
primary = cluster[0]
pc = cluster.connect(0)
- ps = pc.session().sender("q;{create:always,%s}"%qr_node("all"))
- pr = pc.session().receiver("q;{create:always,%s}"%qr_node("all"))
- backup = HaBroker(self, name="backup", ha_cluster=False, args=["--log-enable=debug+"])
+ ps = pc.session().sender("q;{create:always}")
+ pr = pc.session().receiver("q;{create:always}")
+ backup = HaBroker(self, name="backup", ha_cluster=False)
br = backup.connect().session().receiver("q;{create:always}")
backup.replicate(cluster.url, "q")
ps.send("a")
- self.assert_browse_backup(backup, "q", ["a"])
+ backup.assert_browse_backup("q", ["a"])
cluster.bounce(0)
- self.assert_browse_backup(backup, "q", ["a"])
+ backup.assert_browse_backup("q", ["a"])
ps.send("b")
- self.assert_browse_backup(backup, "q", ["a", "b"])
+ backup.assert_browse_backup("q", ["a", "b"])
cluster.bounce(1)
self.assertEqual("a", pr.fetch().content)
pr.session.acknowledge()
- self.assert_browse_backup(backup, "q", ["b"])
+ backup.assert_browse_backup("q", ["b"])
def test_lvq(self):
"""Verify that we replicate to an LVQ correctly"""
primary = HaBroker(self, name="primary")
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
- s = primary.connect().session().sender("lvq; {create:always, node:{x-declare:{arguments:{'qpid.last_value_queue_key':lvq-key, 'qpid.replicate':all}}}}")
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
+ s = primary.connect().session().sender("lvq; {create:always, node:{x-declare:{arguments:{'qpid.last_value_queue_key':lvq-key}}}}")
def send(key,value): s.send(Message(content=value,properties={"lvq-key":key}))
for kv in [("a","a-1"),("b","b-1"),("a","a-2"),("a","a-3"),("c","c-1"),("c","c-2")]:
send(*kv)
- self.assert_browse_backup(backup, "lvq", ["b-1", "a-3", "c-2"])
+ backup.assert_browse_backup("lvq", ["b-1", "a-3", "c-2"])
send("b","b-2")
- self.assert_browse_backup(backup, "lvq", ["a-3", "c-2", "b-2"])
+ backup.assert_browse_backup("lvq", ["a-3", "c-2", "b-2"])
send("c","c-3")
- self.assert_browse_backup(backup, "lvq", ["a-3", "b-2", "c-3"])
+ backup.assert_browse_backup("lvq", ["a-3", "b-2", "c-3"])
send("d","d-1")
- self.assert_browse_backup(backup, "lvq", ["a-3", "b-2", "c-3", "d-1"])
+ backup.assert_browse_backup("lvq", ["a-3", "b-2", "c-3", "d-1"])
def test_ring(self):
+ """Test replication with the ring queue policy"""
primary = HaBroker(self, name="primary")
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
- s = primary.connect().session().sender("q; {create:always, node:{x-declare:{arguments:{'qpid.policy_type':ring, 'qpid.max_count':5, 'qpid.replicate':all}}}}")
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
+ s = primary.connect().session().sender("q; {create:always, node:{x-declare:{arguments:{'qpid.policy_type':ring, 'qpid.max_count':5}}}}")
for i in range(10): s.send(Message(str(i)))
- self.assert_browse_backup(backup, "q", [str(i) for i in range(5,10)])
+ backup.assert_browse_backup("q", [str(i) for i in range(5,10)])
def test_reject(self):
- getLogger().setLevel(ERROR) # Hide expected WARNING log messages from failover.
+ """Test replication with the reject queue policy"""
primary = HaBroker(self, name="primary")
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
- s = primary.connect().session().sender("q; {create:always, node:{x-declare:{arguments:{'qpid.policy_type':reject, 'qpid.max_count':5, 'qpid.replicate':all}}}}")
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
+ s = primary.connect().session().sender("q; {create:always, node:{x-declare:{arguments:{'qpid.policy_type':reject, 'qpid.max_count':5}}}}")
try:
for i in range(10): s.send(Message(str(i)), sync=False)
except qpid.messaging.exceptions.TargetCapacityExceeded: pass
- self.assert_browse_backup(backup, "q", [str(i) for i in range(0,5)])
+ backup.assert_browse_backup("q", [str(i) for i in range(0,5)])
+ # Detach, don't close as there is a broken session
+ s.session.connection.detach()
def test_priority(self):
"""Verify priority queues replicate correctly"""
primary = HaBroker(self, name="primary")
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
session = primary.connect().session()
- s = session.sender("priority-queue; {create:always, node:{x-declare:{arguments:{'qpid.priorities':10, 'qpid.replicate':all}}}}")
+ s = session.sender("priority-queue; {create:always, node:{x-declare:{arguments:{'qpid.priorities':10}}}}")
priorities = [8,9,5,1,2,2,3,4,9,7,8,9,9,2]
for p in priorities: s.send(Message(priority=p))
# Can't use browse_backup as browser sees messages in delivery order not priority.
- self.wait_backup(backup, "priority-queue")
- r = self.connect_admin(backup).session().receiver("priority-queue")
+ backup.wait_backup("priority-queue")
+ r = backup.connect_admin().session().receiver("priority-queue")
received = [r.fetch().priority for i in priorities]
self.assertEqual(sorted(priorities, reverse=True), received)
@@ -460,17 +524,17 @@ class ShortTests(BrokerTest):
"""Verify priority queues replicate correctly"""
primary = HaBroker(self, name="primary")
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
session = primary.connect().session()
levels = 8
priorities = [4,5,3,7,8,8,2,8,2,8,8,16,6,6,6,6,6,6,8,3,5,8,3,5,5,3,3,8,8,3,7,3,7,7,7,8,8,8,2,3]
limits={7:0,6:4,5:3,4:2,3:2,2:2,1:2}
limit_policy = ",".join(["'qpid.fairshare':5"] + ["'qpid.fairshare-%s':%s"%(i[0],i[1]) for i in limits.iteritems()])
- s = session.sender("priority-queue; {create:always, node:{x-declare:{arguments:{'qpid.priorities':%s, %s, 'qpid.replicate':all}}}}"%(levels,limit_policy))
+ s = session.sender("priority-queue; {create:always, node:{x-declare:{arguments:{'qpid.priorities':%s, %s}}}}"%(levels,limit_policy))
messages = [Message(content=str(uuid4()), priority = p) for p in priorities]
for m in messages: s.send(m)
- self.wait_backup(backup, s.target)
- r = self.connect_admin(backup).session().receiver("priority-queue")
+ backup.wait_backup(s.target)
+ r = backup.connect_admin().session().receiver("priority-queue")
received = [r.fetch().content for i in priorities]
sort = sorted(messages, key=lambda m: priority_level(m.priority, levels), reverse=True)
fair = [m.content for m in fairshare(sort, lambda l: limits.get(l,0), levels)]
@@ -479,17 +543,144 @@ class ShortTests(BrokerTest):
def test_priority_ring(self):
primary = HaBroker(self, name="primary")
primary.promote()
- backup = HaBroker(self, name="backup", broker_url=primary.host_port())
- s = primary.connect().session().sender("q; {create:always, node:{x-declare:{arguments:{'qpid.policy_type':ring, 'qpid.max_count':5, 'qpid.priorities':10, 'qpid.replicate':all}}}}")
+ backup = HaBroker(self, name="backup", brokers_url=primary.host_port())
+ s = primary.connect().session().sender("q; {create:always, node:{x-declare:{arguments:{'qpid.policy_type':ring, 'qpid.max_count':5, 'qpid.priorities':10}}}}")
priorities = [8,9,5,1,2,2,3,4,9,7,8,9,9,2]
for p in priorities: s.send(Message(priority=p))
- # FIXME aconway 2012-02-22: there is a bug in priority ring queues that allows a low
- # priority message to displace a high one. The following commented-out assert_browse
- # is for the correct result, the uncommented one is for the actualy buggy result.
- # See https://issues.apache.org/jira/browse/QPID-3866
+
+ # FIXME aconway 2012-02-22: there is a bug in priority ring
+ # queues that allows a low priority message to displace a high
+ # one. The following commented-out assert_browse is for the
+ # correct result, the uncommented one is for the actualy buggy
+ # result. See https://issues.apache.org/jira/browse/QPID-3866
#
- # self.assert_browse_backup(backup, "q", sorted(priorities,reverse=True)[0:5], transform=lambda m: m.priority)
- self.assert_browse_backup(backup, "q", [9,9,9,9,2], transform=lambda m: m.priority)
+ # backup.assert_browse_backup("q", sorted(priorities,reverse=True)[0:5], transform=lambda m: m.priority)
+ backup.assert_browse_backup("q", [9,9,9,9,2], transform=lambda m: m.priority)
+
+ def test_backup_acquired(self):
+ """Verify that acquired messages are backed up, for all queue types."""
+ class Test:
+ def __init__(self, queue, arguments, expect):
+ self.queue = queue
+ self.address = "%s;{create:always,node:{x-declare:{arguments:{%s}}}}"%(
+ self.queue, ",".join(arguments + ["'qpid.replicate':all"]))
+ self.expect = [str(i) for i in expect]
+
+ def send(self, connection):
+ """Send messages, then acquire one but don't acknowledge"""
+ s = connection.session()
+ for m in range(10): s.sender(self.address).send(str(m))
+ s.receiver(self.address).fetch()
+
+ def wait(self, brokertest, backup):
+ backup.wait_backup(self.queue)
+
+ def verify(self, brokertest, backup):
+ backup.assert_browse_backup(self.queue, self.expect, msg=self.queue)
+
+ tests = [
+ Test("plain",[],range(10)),
+ Test("ring", ["'qpid.policy_type':ring", "'qpid.max_count':5"], range(5,10)),
+ Test("priority",["'qpid.priorities':10"], range(10)),
+ Test("fairshare", ["'qpid.priorities':10,'qpid.fairshare':5"], range(10)),
+ Test("lvq", ["'qpid.last_value_queue_key':lvq-key"], [9])
+ ]
+
+ primary = HaBroker(self, name="primary")
+ primary.promote()
+ backup1 = HaBroker(self, name="backup1", brokers_url=primary.host_port())
+ c = primary.connect()
+ for t in tests: t.send(c) # Send messages, leave one unacknowledged.
+
+ backup2 = HaBroker(self, name="backup2", brokers_url=primary.host_port())
+ # Wait for backups to catch up.
+ for t in tests:
+ t.wait(self, backup1)
+ t.wait(self, backup2)
+ # Verify acquired message was replicated
+ for t in tests: t.verify(self, backup1)
+ for t in tests: t.verify(self, backup2)
+
+ def test_replicate_default(self):
+ """Make sure we don't replicate if ha-replicate is unspecified or none"""
+ cluster1 = HaCluster(self, 2, ha_replicate=None)
+ c1 = cluster1[0].connect().session().sender("q;{create:always}")
+ cluster2 = HaCluster(self, 2, ha_replicate="none")
+ cluster2[0].connect().session().sender("q;{create:always}")
+ time.sleep(.1) # Give replication a chance.
+ try:
+ cluster1[1].connect_admin().session().receiver("q")
+ self.fail("Excpected no-such-queue exception")
+ except NotFound: pass
+ try:
+ cluster2[1].connect_admin().session().receiver("q")
+ self.fail("Excpected no-such-queue exception")
+ except NotFound: pass
+
+ def test_invalid_replication(self):
+ """Verify that we reject an attempt to declare a queue with invalid replication value."""
+ cluster = HaCluster(self, 1, ha_replicate="all")
+ try:
+ c = cluster[0].connect().session().sender("q;{create:always, node:{x-declare:{arguments:{'qpid.replicate':XXinvalidXX}}}}")
+ self.fail("Expected ConnectionError")
+ except ConnectionError: pass
+
+ def test_exclusive_queue(self):
+ """Ensure that we can back-up exclusive queues, i.e. the replicating
+ subscriptions are exempt from the exclusivity"""
+ cluster = HaCluster(self, 2)
+ def test(addr):
+ c = cluster[0].connect()
+ q = addr.split(";")[0]
+ r = c.session().receiver(addr)
+ try: c.session().receiver(addr); self.fail("Expected exclusive exception")
+ except ReceiverError: pass
+ s = c.session().sender(q).send(q)
+ cluster[1].assert_browse_backup(q, [q])
+ test("excl_sub;{create:always, link:{x-subscribe:{exclusive:True}}}");
+ test("excl_queue;{create:always, node:{x-declare:{exclusive:True}}}")
+
+ def test_auto_delete_exclusive(self):
+ """Verify that we ignore auto-delete, exclusive, non-auto-delete-timeout queues"""
+ cluster = HaCluster(self,2)
+ s = cluster[0].connect().session()
+ s.receiver("exad;{create:always,node:{x-declare:{exclusive:True,auto-delete:True}}}")
+ s.receiver("ex;{create:always,node:{x-declare:{exclusive:True}}}")
+ s.receiver("ad;{create:always,node:{x-declare:{auto-delete:True}}}")
+ s.receiver("time;{create:always,node:{x-declare:{exclusive:True,auto-delete:True,arguments:{'qpid.auto_delete_timeout':1}}}}")
+ s.receiver("q;{create:always}")
+
+ s = cluster[1].connect_admin().session()
+ cluster[1].wait_backup("q")
+ assert not valid_address(s, "exad")
+ assert valid_address(s, "ex")
+ assert valid_address(s, "ad")
+ assert valid_address(s, "time")
+
+ def test_broker_info(self):
+ """Check that broker information is correctly published via management"""
+ cluster = HaCluster(self, 3)
+
+ for broker in cluster: # Make sure HA system-id matches broker's
+ qmf = broker.agent().getHaBroker()
+ self.assertEqual(qmf.systemId, UUID(broker.agent().getBroker().systemRef))
+
+ cluster_ports = map(lambda b: b.port(), cluster)
+ cluster_ports.sort()
+ def ports(qmf):
+ qmf.update()
+ return sorted(map(lambda b: b["port"], qmf.members))
+ # Check that all brokers have the same membership as the cluster
+ for broker in cluster:
+ qmf = broker.agent().getHaBroker()
+ assert retry(lambda: cluster_ports == ports(qmf), 1), "%s != %s on %s"%(cluster_ports, ports(qmf), broker)
+ # Add a new broker, check it is updated everywhere
+ b = cluster.start()
+ cluster_ports.append(b.port())
+ cluster_ports.sort()
+ for broker in cluster:
+ qmf = broker.agent().getHaBroker()
+ assert retry(lambda: cluster_ports == ports(qmf), 1), "%s != %s"%(cluster_ports, ports(qmf))
def fairshare(msgs, limit, levels):
"""
@@ -533,49 +724,108 @@ class LongTests(BrokerTest):
if d: return float(d)*60
else: return 3 # Default is to be quick
-
- def disable_test_failover(self):
+ def test_failover_send_receive(self):
"""Test failover with continuous send-receive"""
- # FIXME aconway 2012-02-03: fails due to dropped messages,
- # known issue: sending messages to new primary before
- # backups are ready. Enable when fixed.
-
- # Start a cluster, all members will be killed during the test.
- brokers = [ HaBroker(self, name=name, expect=EXPECT_EXIT_FAIL)
- for name in ["ha0","ha1","ha2"] ]
- url = ",".join([b.host_port() for b in brokers])
- for b in brokers: b.set_broker_url(url)
- brokers[0].promote()
+ brokers = HaCluster(self, 3)
# Start sender and receiver threads
- sender = NumberedSender(brokers[0], max_depth=1000, failover_updates=False)
- receiver = NumberedReceiver(brokers[0], sender=sender, failover_updates=False)
- receiver.start()
- sender.start()
+ n = 10
+ senders = [NumberedSender(brokers[0], max_depth=1024, failover_updates=False,
+ queue="test%s"%(i)) for i in xrange(n)]
+ receivers = [NumberedReceiver(brokers[0], sender=senders[i],
+ failover_updates=False,
+ queue="test%s"%(i)) for i in xrange(n)]
+ for r in receivers: r.start()
+ for s in senders: s.start()
+
# Wait for sender & receiver to get up and running
- assert retry(lambda: receiver.received > 100)
+ assert retry(lambda: receivers[0].received > 100), "%s<=100"%receivers[0].received
# Kill and restart brokers in a cycle:
endtime = time.time() + self.duration()
i = 0
- while time.time() < endtime or i < 3: # At least 3 iterations
- sender.sender.assert_running()
- receiver.receiver.assert_running()
- port = brokers[i].port()
- brokers[i].kill()
- brokers.append(
- HaBroker(self, name="ha%d"%(i+3), broker_url=url, port=port,
- expect=EXPECT_EXIT_FAIL))
- i += 1
- brokers[i].promote()
- n = receiver.received # Verify we're still running
- def enough():
- receiver.check() # Verify no exceptions
- return receiver.received > n + 100
- assert retry(enough, timeout=5)
+ try:
+ while time.time() < endtime or i < 3: # At least 3 iterations
+ for s in senders: s.sender.assert_running()
+ for r in receivers: r.receiver.assert_running()
+ n = receivers[0].received
+ # FIXME aconway 2012-05-01: don't kill primary till it's active
+ # and backups are ready, otherwise we can lose messages. When we
+ # implement non-promotion of catchup brokers we can make this
+ # stronger: wait only for there to be at least one ready backup.
+ brokers[i%3].wait_status("active")
+ brokers[(i+1)%3].wait_status("ready")
+ brokers[(i+2)%3].wait_status("ready")
+ brokers.bounce(i%3)
+ i += 1
+ def enough(): # Verify we're still running
+ receivers[0].check() # Verify no exceptions
+ return receivers[0].received > n + 100
+ assert retry(enough), "Stalled: %s < %s+100"%(receivers[0].received, n)
+ except:
+ traceback.print_exc()
+ raise
+ finally:
+ for s in senders: s.stop()
+ for r in receivers: r.stop()
+ dead = []
+ for i in xrange(3):
+ if not brokers[i].is_running(): dead.append(i)
+ brokers.kill(i, False)
+ if dead: raise Exception("Brokers not running: %s"%dead)
+
+class RecoveryTests(BrokerTest):
+ """Tests for recovery after a failure."""
+
+ def test_queue_hold(self):
+ """Verify that the broker holds queues without sufficient backup,
+ i.e. does not complete messages sent to those queues."""
+
+ cluster = HaCluster(self, 4);
+ # Wait for the primary to be ready
+ cluster[0].wait_status("active")
+ # Create a queue before the failure.
+ s1 = cluster.connect(0).session().sender("q1;{create:always}")
+ for b in cluster: b.wait_backup("q1")
+ for i in xrange(100): s1.send(str(i))
+ # Kill primary and 2 backups
+ for i in [0,1,2]: cluster.kill(i, False)
+ cluster[3].promote() # New primary, backups will be 1 and 2
+ cluster[3].wait_status("recovering")
- sender.stop()
- receiver.stop()
- for b in brokers[i:]: b.kill()
+ def trySync(s):
+ try:
+ s.sync(timeout=.1)
+ self.fail("Expected Timeout exception")
+ except Timeout: pass
+
+ # Create a queue after the failure
+ s2 = cluster.connect(3).session().sender("q2;{create:always}")
+ # Verify that messages sent are not completed
+ for i in xrange(100,200): s1.send(str(i), sync=False); s2.send(str(i), sync=False)
+ trySync(s1)
+ self.assertEqual(s1.unsettled(), 100)
+ trySync(s2)
+ self.assertEqual(s2.unsettled(), 100)
+
+ # Verify we can receive even if sending is on hold:
+ cluster[3].assert_browse("q1", [str(i) for i in range(100)+range(100,200)])
+ # Restart backups, verify queues are released only when both backups are up
+ cluster.restart(1)
+ trySync(s1)
+ self.assertEqual(s1.unsettled(), 100)
+ trySync(s2)
+ self.assertEqual(s2.unsettled(), 100)
+ self.assertEqual(cluster[3].ha_status(), "recovering")
+ cluster.restart(2)
+
+ def settled(sender): sender.sync(); return sender.unsettled() == 0;
+ assert retry(lambda: settled(s1)), "Unsetttled=%s"%(s1.unsettled())
+ assert retry(lambda: settled(s2)), "Unsetttled=%s"%(s2.unsettled())
+ cluster[1].assert_browse_backup("q1", [str(i) for i in range(100)+range(100,200)])
+ cluster[1].assert_browse_backup("q2", [str(i) for i in range(100,200)])
+ cluster[3].wait_status("active"),
+ s1.session.connection.close()
+ s2.session.connection.close()
if __name__ == "__main__":
shutil.rmtree("brokertest.tmp", True)
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/install_env.sh.in
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/install_env.sh.in?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/install_env.sh.in (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/install_env.sh.in Thu Jun 28 09:14:52 2012
@@ -23,4 +23,4 @@ prefix=`absdir @prefix@`
export QPID_INSTALL_PREFIX=$prefix
export PATH=$prefix/bin:$prefix/sbin:$prefix/libexec/qpid/tests:$PATH
export LD_LIBRARY_PATH=$prefix/lib:$LD_LIBRARY_PATH
-export PYTHONPATH=$prefix/lib/python2.4/site-packages:$PYTHONPATH
+export PYTHONPATH=$prefix/lib/python2.6/site-packages:$PYTHONPATH
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/ipv6_test
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/ipv6_test?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/ipv6_test (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/ipv6_test Thu Jun 28 09:14:52 2012
@@ -113,7 +113,8 @@ fi
test -z $CLUSTER_LIB && exit 0 # Exit if cluster not supported.
## Test failover in a cluster using IPv6 only
-. $srcdir/ais_check # Will exit if clustering not enabled.
+. cpg_check.sh
+cpg_enabled || exit 0
pick_port() {
# We need a fixed port to set --cluster-url. Use qpidd to pick a free port.
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/logging.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/logging.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/logging.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/logging.cpp Thu Jun 28 09:14:52 2012
@@ -44,8 +44,10 @@ namespace tests {
QPID_AUTO_TEST_SUITE(loggingTestSuite)
using namespace std;
-using namespace boost;
using namespace qpid::log;
+using boost::ends_with;
+using boost::contains;
+using boost::format;
QPID_AUTO_TEST_CASE(testStatementInit) {
Statement s=QPID_LOG_STATEMENT_INIT(debug); int line=__LINE__;
@@ -256,7 +258,7 @@ QPID_AUTO_TEST_CASE(testOverhead) {
Statement statement(
Level level, const char* file="", int line=0, const char* fn=0)
{
- Statement s={0, file, line, fn, level};
+ Statement s={0, file, line, fn, level, ::qpid::log::unspecified};
return s;
}
@@ -345,11 +347,11 @@ QPID_AUTO_TEST_CASE(testLoggerStateure)
};
opts.parse(ARGC(argv), const_cast<char**>(argv));
l.configure(opts);
- QPID_LOG(critical, "foo"); int srcline=__LINE__;
+ QPID_LOG_CAT(critical, test, "foo"); int srcline=__LINE__;
ifstream log("logging.tmp");
string line;
getline(log, line);
- string expect=(format("critical %s:%d: foo")%__FILE__%srcline).str();
+ string expect=(format("[Test] critical %s:%d: foo")%__FILE__%srcline).str();
BOOST_CHECK_EQUAL(expect, line);
log.close();
unlink("logging.tmp");
@@ -377,7 +379,7 @@ QPID_AUTO_TEST_CASE(testQuoteNonPrintabl
ifstream log("logging.tmp");
string line;
getline(log, line, '\0');
- string expect="critical null\\x00tab\tspace newline\nret\r\\x80\\x99\\xFF\\x00\n";
+ string expect="[Test] critical null\\x00tab\tspace newline\nret\r\\x80\\x99\\xFF\\x00\n";
BOOST_CHECK_EQUAL(expect, line);
log.close();
unlink("logging.tmp");
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cluster-benchmark
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cluster-benchmark?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cluster-benchmark (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cluster-benchmark Thu Jun 28 09:14:52 2012
@@ -22,38 +22,45 @@
# Default options
MESSAGES="-m 10000"
-FLOW="--flow-control 100" # Flow control limit on queue depth for latency.
REPEAT="--repeat 10"
QUEUES="-q 6"
SENDERS="-s 3"
RECEIVERS="-r 3"
BROKERS= # Local broker
CLIENT_HOSTS= # No ssh, all clients are local
+# Connection options
+TCP_NODELAY=false
+RECONNECT=true
+HEARTBEAT=1
-while getopts "m:f:n:b:q:s:r:c:txyv-" opt; do
+while getopts "m:f:n:b:q:s:r:c:h:i:txyv-" opt; do
case $opt in
+ b) BROKERS="-b $OPTARG";;
+ c) CLIENT_HOSTS="-c $OPTARG";;
+ h) HEARTBEAT=$OPTARG;;
+ i) RECONNECT=$OPTARG;;
m) MESSAGES="-m $OPTARG";;
- f) FLOW="--flow-control $OPTARG";;
n) REPEAT="--repeat $OPTARG";;
- b) BROKERS="-b $OPTARG";;
q) QUEUES="-q $OPTARG";;
- s) SENDERS="-s $OPTARG";;
r) RECEIVERS="-r $OPTARG";;
- c) CLIENT_HOSTS="-c $OPTARG";;
- t) TCP_NODELAY="--connection-options {tcp-nodelay:true}";;
+ s) SENDERS="-s $OPTARG";;
+ t) TCP_NODELAY=true;;
+ v) OPTS="--verbose";;
x) SAVE_RECEIVED="--save-received";;
y) NO_DELETE="--no-delete";;
- v) OPTS="--verbose";;
-) break ;;
*) echo "Unknown option"; exit 1;;
esac
done
shift $(($OPTIND-1))
-REPLICATE="node:{x-declare:{arguments:{'qpid.replicate':all}}}"
+CONNECTION_OPTIONS="--connection-options {tcp-nodelay:$TCP_NODELAY,reconnect:$RECONNECT,heartbeat:$HEARTBEAT}"
+CREATE_OPTIONS="node:{x-declare:{arguments:{'qpid.replicate':all}}}"
+
BROKER=$(echo $BROKERS | sed s/,.*//)
run_test() { echo $*; shift; "$@"; echo; echo; echo; }
-OPTS="$OPTS $REPEAT $BROKERS --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS $SAVE_RECEIVED $TCP_NODELAY $NO_DELETE"
-OPTS="$OPTS --create-option $REPLICATE"
+OPTS="$OPTS $REPEAT $BROKERS --summarize $QUEUES $SENDERS $RECEIVERS $MESSAGES $CLIENT_HOSTS $SAVE_RECEIVED $CONNECTION_OPTIONS $NO_DELETE"
+OPTS="$OPTS --create-option $CREATE_OPTIONS"
+
run_test "Benchmark:" qpid-cpp-benchmark $OPTS "$@"
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cpp-benchmark
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cpp-benchmark?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cpp-benchmark (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-cpp-benchmark Thu Jun 28 09:14:52 2012
@@ -37,7 +37,7 @@ op.add_option("-r", "--receivers", defau
op.add_option("-m", "--messages", default=100000, type="int", metavar="N",
help="send N messages per sender (default %default)")
op.add_option("--queue-name", default="benchmark", metavar="NAME",
- help="base name for queues (default %default)")
+ help="base name for queues (default %default)")
op.add_option("--send-rate", default=0, metavar="N",
help="send rate limited to N messages/second, 0 means no limit (default %default)")
op.add_option("--receive-rate", default=0, metavar="N",
@@ -67,18 +67,16 @@ op.add_option("--sequence", dest="sequen
action="store_true", help="add a sequence number to each message")
op.add_option("--connection-options", type="str",
help="Connection options for senders & receivers")
-op.add_option("--flow-control", default=0, type="int", metavar="N",
- help="Flow control each sender to limit queue depth to 2*N. 0 means no flow control.")
op.add_option("--durable", default=False, action="store_true",
help="Use durable queues and messages")
op.add_option("--save-received", default=False, action="store_true",
help="Save received message content to files <queuename>-receiver-<n>.msg")
-op.add_option("--group-receivers", default=False, action="store_true",
- help="Run receivers for the same queue on the same host.")
op.add_option("--verbose", default=False, action="store_true",
help="Show commands executed")
op.add_option("--no-delete", default=False, action="store_true",
help="Don't delete the test queues.")
+op.add_option("--fill-drain", default=False, action="store_true",
+ help="First fill the queues, then drain them")
single_quote_re = re.compile("'")
def posix_quote(string):
@@ -150,7 +148,6 @@ def start_send(queue, opts, broker, host
"--report-header=no",
"--timestamp=%s"%(opts.timestamp and "yes" or "no"),
"--sequence=%s"%(opts.sequence and "yes" or "no"),
- "--flow-control", str(opts.flow_control),
"--durable", str(opts.durable)
]
command += opts.send_arg
@@ -169,18 +166,6 @@ def first_line(p):
raise Exception("Process exit %d: %s"%(p.returncode, error_msg(out,err)))
return out.split("\n")[0]
-def queue_exists(queue,broker):
- c = qpid.messaging.Connection(broker)
- c.open()
- try:
- s = c.session()
- try:
- s.sender(queue)
- return True
- except qpid.messaging.exceptions.NotFound:
- return False
- finally: c.close()
-
def recreate_queues(queues, brokers, no_delete, opts):
c = qpid.messaging.Connection(brokers[0])
c.open()
@@ -189,15 +174,9 @@ def recreate_queues(queues, brokers, no_
if not no_delete:
try: s.sender("%s;{delete:always}"%(q)).close()
except qpid.messaging.exceptions.NotFound: pass
- # FIXME aconway 2011-05-04: new cluster async wiring, wait for changes to propagate
- for b in brokers:
- while queue_exists(q,b): time.sleep(0.1);
address = "%s;{%s}"%(q, ",".join(opts.create_option + ["create:always"]))
if opts.verbose: print "Creating", address
s.sender(address)
- # FIXME aconway 2011-05-04: new cluster async wiring, wait for changes to propagate
- for b in brokers:
- while not queue_exists(q,b): time.sleep(0.1);
c.close()
def print_header(timestamp):
@@ -295,24 +274,32 @@ def main():
recreate_queues(queues, opts.broker, opts.no_delete, opts)
ready_receiver = ReadyReceiver(ready_queue, opts.broker[0])
- if opts.group_receivers: # Run receivers for same queue against same broker.
- receivers = []
- for q in queues:
- b = brokers.next()
- for j in xrange(opts.receivers):
- receivers.append(
- start_receive(q, j, opts, ready_queue, b, client_hosts.next()))
- else: # Don't group receivers
- receivers = [start_receive(q, j, opts, ready_queue,
- brokers.next(), client_hosts.next())
- for q in queues for j in xrange(opts.receivers)]
-
- ready_receiver.wait(filter(None, receivers)) # Wait for receivers to be ready.
- start = time.time()
- senders = [start_send(q, opts,brokers.next(), client_hosts.next())
- for q in queues for j in xrange(opts.senders)]
+ def start_receivers():
+ return [ start_receive(q, j, opts, ready_queue, brokers.next(), client_hosts.next())
+ for q in queues for j in xrange(opts.receivers) ]
+
+
+ def start_senders():
+ return [ start_send(q, opts,brokers.next(), client_hosts.next())
+ for q in queues for j in xrange(opts.senders) ]
+
if opts.report_header and i == 0: print_header(opts.timestamp)
- for p in senders + receivers: p.wait()
+
+ if opts.fill_drain:
+ # First fill the queues, then drain them
+ start = time.time()
+ senders = start_senders()
+ for p in senders: p.wait()
+ receivers = start_receivers()
+ for p in receivers: p.wait()
+ else:
+ # Run senders and receivers in parallel
+ receivers = start_receivers()
+ ready_receiver.wait(filter(None, receivers)) # Wait for receivers ready
+ start = time.time()
+ senders = start_senders()
+ for p in senders + receivers: p.wait()
+
total_sent = opts.queues * opts.senders * opts.messages
total_tp = total_sent / (time.time()-start)
send_stats=parse_senders(senders)
Modified: qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-latency-test.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-latency-test.cpp?rev=1354874&r1=1354873&r2=1354874&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-latency-test.cpp (original)
+++ qpid/branches/java-config-and-management/qpid/cpp/src/tests/qpid-latency-test.cpp Thu Jun 28 09:14:52 2012
@@ -359,19 +359,29 @@ void Sender::sendByRate()
}
uint64_t interval = TIME_SEC/opts.rate;
int64_t timeLimit = opts.timeLimit * TIME_SEC;
- uint64_t sent = 0, missedRate = 0;
+ uint64_t sent = 0;
AbsTime start = now();
+ AbsTime last = start;
while (true) {
AbsTime sentAt=now();
msg.getDeliveryProperties().setTimestamp(Duration(EPOCH, sentAt));
async(session).messageTransfer(arg::content=msg, arg::acceptMode=1);
if (opts.sync) session.sync();
++sent;
+ if (Duration(last, sentAt) > TIME_SEC*2) {
+ Duration t(start, now());
+ //check rate actually achieved thus far
+ uint actualRate = sent / (t/TIME_SEC);
+ //report inability to stay within 1% of desired rate
+ if (actualRate < opts.rate && opts.rate - actualRate > opts.rate/100) {
+ std::cerr << "WARNING: Desired send rate: " << opts.rate << ", actual send rate: " << actualRate << std::endl;
+ }
+ last = sentAt;
+ }
+
AbsTime waitTill(start, sent*interval);
Duration delay(sentAt, waitTill);
- if (delay < 0)
- ++missedRate;
- else
+ if (delay > 0)
sys::usleep(delay / TIME_USEC);
if (timeLimit != 0 && Duration(start, now()) > timeLimit) {
session.sync();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org