Support for restarting a persistent cluster.

Option --cluster-size=N: members wait for N members before recovering store.
Stores marked as clean/dirty. Automatically recover from clean store on restart.
Stores marked with UUID to detect errors.

Not yet implemented: consistency checks, manual recovery from all dirty stores.


git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@883842 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 53 insertions, 52 deletions

diff --git a/qpid/cpp/src/tests/cluster_tests.py b/qpid/cpp/src/tests/cluster_tests.py
index ed39277f77..65c91b1d81 100755
--- a/qpid/cpp/src/tests/cluster_tests.py
+++ b/qpid/cpp/src/tests/cluster_tests.py
@@ -18,13 +18,18 @@
 # under the License.
 #
 
-import os, signal, sys, time
+import os, signal, sys, time, imp
 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."""
@@ -34,8 +39,8 @@ class ShortTests(BrokerTest):
         # Start a cluster, send some messages to member 0.
         cluster = self.cluster(2)
         s0 = cluster[0].connect().session()
-        s0.sender("q; {create:always}").send(messaging.Message("x"))
-        s0.sender("q; {create:always}").send(messaging.Message("y"))
+        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.
@@ -52,35 +57,6 @@ class ShortTests(BrokerTest):
         self.assertEqual("y", m.content)
         s2.connection.close()
 
-    def test_cluster_size(self):
-        """Verify cluster startup waits for N brokers if --cluster-size=N"""
-        class ConnectThread(Thread):
-            def __init__(self, broker):
-                Thread.__init__(self)
-                self.broker=broker
-                self.connected = False
-                self.error = None
-
-            def run(self):
-                try:
-                    self.broker.connect()
-                    self.connected = True
-                except Exception, e: self.error = RethrownException(e)
-
-        cluster = self.cluster(1, args=["--cluster-size=3"], wait_for_start=False)
-        c = ConnectThread(cluster[0])
-        c.start()
-        time.sleep(.01)
-        assert not c.connected
-        cluster.start(wait_for_start=False)
-        time.sleep(.01)
-        assert not c.connected
-        cluster.start(wait_for_start=False)
-        c.join(1)
-        assert not c.isAlive()       # Join didn't time out
-        assert c.connected
-        if c.error: raise c.error
-
 class LongTests(BrokerTest):
     """Tests that can run for a long time if -DDURATION=<minutes> is set"""
     def duration(self):
@@ -120,20 +96,22 @@ class StoreTests(BrokerTest):
     """
     Cluster tests that can only be run if there is a store available.
     """
-    args = ["--load-module",BrokerTest.store_lib]
+    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 = messaging.Message("x", durable=True)
+        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")
+        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 = self.cluster(1, self.args())
         cluster.start("restartme", expect=EXPECT_EXIT_FAIL).kill()
 
         # Send a message, retrieve from the restarted broker
@@ -141,19 +119,42 @@ class StoreTests(BrokerTest):
         m = cluster.start("restartme").get_message("q")
         self.assertEqual("x", m.content)
 
-    def test_total_shutdown(self):
-        """Test we use the correct store to recover after total shutdown"""
-        cluster = self.cluster(2, args=self.args, expect=EXPECT_EXIT_FAIL)
-        cluster[0].send_message("q", Message("a", durable=True))
-        cluster[0].kill()
-        self.assertEqual("a", cluster[1].get_message("q").content)
-        cluster[1].send_message("q", Message("b", durable=True))
-        cluster[1].kill()
-
-        # Start 1 first, we should see its store used.
-        cluster.start(name=cluster.name+"-1")
-        cluster.start(name=cluster.name+"-0")
-        self.assertEqual("b", cluster[2].get_message("q").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_for_start=False)
+        b = cluster.start("b", expect=EXPECT_EXIT_OK, wait_for_start=False)
+        c = cluster.start("c", expect=EXPECT_EXIT_FAIL, wait_for_start=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_for_start=False)
+        b = cluster.start("b", wait_for_start=False)
+        c = cluster.start("c", wait_for_start=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_for_start=False)
+        b = cluster.start("b", expect=EXPECT_EXIT_FAIL, wait_for_start=False)
+        c = cluster.start("c", expect=EXPECT_EXIT_OK, wait_for_start=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_for_start=False)
+        b = cluster.start("b", wait_for_start=False)
+        c = cluster.start("c", wait_for_start=True)
+        self.assertEqual(a.get_message("q").content, "clean")
         
-