#!/usr/bin/env python # 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. # import os, signal, sys, time, imp, re from qpid import datatypes, messaging from qpid.brokertest import * from qpid.harness import Skipped from qpid.messaging import Message from threading import Thread from logging import getLogger log = getLogger("qpid.cluster_tests") # Import scripts as modules qpid_cluster=import_script(checkenv("QPID_CLUSTER_EXEC")) class ShortTests(BrokerTest): """Short cluster functionality tests.""" def test_message_replication(self): """Test basic cluster message replication.""" # Start a cluster, send some messages to member 0. cluster = self.cluster(2) s0 = cluster[0].connect().session() s0.sender("q; {create:always}").send(Message("x")) s0.sender("q; {create:always}").send(Message("y")) s0.connection.close() # Verify messages available on member 1. s1 = cluster[1].connect().session() m = s1.receiver("q", capacity=1).fetch(timeout=1) s1.acknowledge() self.assertEqual("x", m.content) s1.connection.close() # Start member 2 and verify messages available. s2 = cluster.start().connect().session() m = s2.receiver("q", capacity=1).fetch(timeout=1) s2.acknowledge() self.assertEqual("y", m.content) s2.connection.close() def test_store_direct_update_match(self): """Verify that brokers stores an identical message whether they receive it direct from clients or during an update, no header or other differences""" cluster = self.cluster(0, args=["--load-module", self.test_store_lib]) cluster.start(args=["--test-store-dump", "direct.dump"]) # Try messages with various headers cluster[0].send_message("q", Message(durable=True, content="foobar", subject="subject", reply_to="reply_to", properties={"n":10})) # Try messages of different sizes for size in range(0,10000,100): cluster[0].send_message("q", Message(content="x"*size, durable=True)) # Try sending via named exchange c = cluster[0].connect_old() s = c.session(str(qpid.datatypes.uuid4())) s.exchange_bind(exchange="amq.direct", binding_key="foo", queue="q") props = s.delivery_properties(routing_key="foo", delivery_mode=2) s.message_transfer( destination="amq.direct", message=qpid.datatypes.Message(props, "content")) # Now update a new member and compare their dumps. cluster.start(args=["--test-store-dump", "updatee.dump"]) assert file("direct.dump").read() == file("updatee.dump").read() os.remove("direct.dump") os.remove("updatee.dump") class LongTests(BrokerTest): """Tests that can run for a long time if -DDURATION= is set""" def duration(self): d = self.config.defines.get("DURATION") if d: return float(d)*60 else: return 3 # Default is to be quick def test_failover(self): """Test fail-over during continuous send-receive with errors""" # Original cluster will all be killed so expect exit with failure cluster = self.cluster(3, expect=EXPECT_EXIT_FAIL) for b in cluster: ErrorGenerator(b) # Start sender and receiver threads cluster[0].declare_queue("test-queue") sender = NumberedSender(cluster[1], 1000) # Max queue depth receiver = NumberedReceiver(cluster[2], sender) receiver.start() sender.start() # Kill original brokers, start new ones for the duration. endtime = time.time() + self.duration() i = 0 while time.time() < endtime: cluster[i].kill() i += 1 b = cluster.start(expect=EXPECT_EXIT_FAIL) ErrorGenerator(b) time.sleep(1) sender.stop() receiver.stop(sender.sent) for i in range(i, len(cluster)): cluster[i].kill() class StoreTests(BrokerTest): """ Cluster tests that can only be run if there is a store available. """ def args(self): assert BrokerTest.store_lib return ["--load-module", BrokerTest.store_lib] def test_store_loaded(self): """Ensure we are indeed loading a working store""" broker = self.broker(self.args(), name="recoverme", expect=EXPECT_EXIT_FAIL) m = Message("x", durable=True) broker.send_message("q", m) broker.kill() broker = self.broker(self.args(), name="recoverme") self.assertEqual("x", broker.get_message("q").content) def test_kill_restart(self): """Verify we can kill/resetart a broker with store in a cluster""" cluster = self.cluster(1, self.args()) cluster.start("restartme", expect=EXPECT_EXIT_FAIL).kill() # Send a message, retrieve from the restarted broker cluster[0].send_message("q", "x") m = cluster.start("restartme").get_message("q") self.assertEqual("x", m.content) def test_persistent_restart(self): """Verify persistent cluster shutdown/restart scenarios""" cluster = self.cluster(0, args=self.args() + ["--cluster-size=3"]) a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False) b = cluster.start("b", expect=EXPECT_EXIT_OK, wait=False) c = cluster.start("c", expect=EXPECT_EXIT_FAIL, wait=True) a.send_message("q", Message("1", durable=True)) # Kill & restart one member. c.kill() self.assertEqual(a.get_message("q").content, "1") a.send_message("q", Message("2", durable=True)) c = cluster.start("c", expect=EXPECT_EXIT_OK) self.assertEqual(c.get_message("q").content, "2") # Shut down the entire cluster cleanly and bring it back up a.send_message("q", Message("3", durable=True)) qpid_cluster.main(["qpid-cluster", "-kf", a.host_port()]) a = cluster.start("a", wait=False) b = cluster.start("b", wait=False) c = cluster.start("c", wait=True) self.assertEqual(a.get_message("q").content, "3") def test_persistent_partial_failure(self): # Kill 2 members, shut down the last cleanly then restart # Ensure we use the clean database cluster = self.cluster(0, args=self.args() + ["--cluster-size=3"]) a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False) b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=False) c = cluster.start("c", expect=EXPECT_EXIT_OK, wait=True) a.send_message("q", Message("4", durable=True)) a.kill() b.kill() self.assertEqual(c.get_message("q").content, "4") c.send_message("q", Message("clean", durable=True)) qpid_cluster.main(["qpid-cluster", "-kf", c.host_port()]) a = cluster.start("a", wait=False) b = cluster.start("b", wait=False) c = cluster.start("c", wait=True) self.assertEqual(a.get_message("q").content, "clean") def test_wrong_cluster_id(self): # Start a cluster1 broker, then try to restart in cluster2 cluster1 = self.cluster(0, args=self.args()) a = cluster1.start("a", expect=EXPECT_EXIT_OK) a.terminate() cluster2 = self.cluster(1, args=self.args()) try: a = cluster2.start("a", expect=EXPECT_EXIT_FAIL) self.fail("Expected exception") except: pass def test_wrong_shutdown_id(self): # Start 2 members and shut down. cluster = self.cluster(0, args=self.args()+["--cluster-size=2"]) a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False) b = cluster.start("b", expect=EXPECT_EXIT_OK, wait=False) self.assertEqual(0, qpid_cluster.main(["qpid_cluster", "-kf", a.host_port()])) self.assertEqual(a.wait(), 0) self.assertEqual(b.wait(), 0) # Restart with a different member and shut down. a = cluster.start("a", expect=EXPECT_EXIT_OK, wait=False) c = cluster.start("c", expect=EXPECT_EXIT_OK, wait=False) self.assertEqual(0, qpid_cluster.main(["qpid_cluster", "-kf", a.host_port()])) self.assertEqual(a.wait(), 0) self.assertEqual(c.wait(), 0) # Mix members from both shutdown events, they should fail a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False) b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=False) def test_total_failure(self): # Verify we abort with sutiable error message if no clean stores. cluster = self.cluster(0, args=self.args()+["--cluster-size=2"]) a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False) b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=True) a.kill() b.kill() a = cluster.start("a", expect=EXPECT_EXIT_FAIL, wait=False) b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait=False) assert a.wait() != 0 assert b.wait() != 0 msg = re.compile("critical.*no clean store") assert msg.search(file(a.log).read()) assert msg.search(file(b.log).read())