use docker api to inject failures

Author: kelvich

tests2/docker-entrypoint.sh

@@ -66,7 +66,7 @@ if [ "$1" = 'postgres' ]; then
 			multimaster.ignore_tables_without_pk = 1
 			multimaster.node_id = $NODE_ID
 			multimaster.conn_strings = '$CONNSTRS'
-			multimaster.heartbeat_recv_timeout = 1000
+			multimaster.heartbeat_recv_timeout = 1100
 			multimaster.heartbeat_send_timeout = 250
 			multimaster.twopc_min_timeout = 200000
 		EOF

tests2/lib/bank_client.py

@@ -195,12 +195,16 @@ def bgrun(self):
         self.evloop_process = multiprocessing.Process(target=self.run, args=())
         self.evloop_process.start()
 
-    def get_status(self):
+    # XXX: rename to start/get aggregation
+    # XXX: introduce periodic report from client
+    def get_status(self, print=True):
         # print('test: sending status request')
         self.parent_pipe.send('status')
         # print('test: awaitng status response')
         resp = self.parent_pipe.recv()
         # print('test: got status response')
+        if print:
+            MtmClient.print_aggregates(resp)
         return resp
 
     def stop(self):

tests2/requirements.txt

@@ -1,4 +1,4 @@
 psycopg2
 aiopg
 aioprocessing
-git+git://github.com/douglas/toxiproxy-python.git
+docker

tests2/test_recovery.py

@@ -1,11 +1,18 @@
+#
+# Based on Aphyr's test for CockroachDB.
+#
+
 import unittest
 import time
 import subprocess
 import datetime
+import docker
 
-from toxiproxy import Toxiproxy
 from lib.bank_client import MtmClient
+from lib.failure_injector import *
 
+TEST_DURATION = 10
+TEST_RECOVERY_TIME = 10
 
 class RecoveryTest(unittest.TestCase):
     @classmethod
@@ -23,69 +30,71 @@ def setUpClass(self):
         ], n_accounts=1000)
         self.client.bgrun()
 
-        # self.toxiproxy = Toxiproxy() #Toxiproxy(server_host="toxi")
-
-
     @classmethod
     def tearDownClass(self):
         print('tearDown')
         self.client.stop()
+        # XXX: check nodes data identity here
+
         # subprocess.check_call(['docker-compose','down'])
 
-    # def test_normal_operations(self):
-    #     print('### normalOpsTest ###')
+    def test_normal_operations(self):
+        print('### normal_operations ###')
+
+        # nofailure here
+        # clean aggregates
+        self.client.get_status()
+
+        time.sleep(TEST_DURATION)
+        aggs = self.client.get_status()
+
+        for agg in aggs:
+            self.assertTrue( aggs[agg]['finish']['commit'] > 0 )
+
+        # nofailure ends here
+        self.client.get_status()
+
+        time.sleep(TEST_RECOVERY_TIME)
+        aggs = self.client.get_status()
+
+        for agg in aggs:
+            self.assertTrue( aggs[agg]['finish']['commit'] > 0 )
 
-    #     for i in range(3):
-    #         time.sleep(3)
-    #         aggs = self.client.get_status()
-    #         MtmClient.print_aggregates(aggs)
-    #         print(aggs)
-    #         for agg in aggs:
-    #             self.assertTrue( aggs[agg]['finish']['commit'] > 0 )
 
     def test_node_partition(self):
         print('### nodePartitionTest ###')
 
-        # print('### split node3 ###')
-        # for proxy in ['rep31', 'rep32', 'rep23', 'rep13', 'arb31', 'arb32', 'arb23', 'arb13']:
-        #     self.toxiproxy.get_proxy(proxy).disable()
+        failure = SingleNodePartition('node3')
 
-        # clear tx history
-        self.client.get_status()
+        # split one node
+        failure.start()
+        # clean aggregates
+        self.client.get_status(print=False)
 
-        for i in range(1000):
-            print(i, datetime.datetime.now())
-            time.sleep(3)
-            aggs = self.client.get_status()
-            MtmClient.print_aggregates(aggs)
-            # self.assertTrue( aggs['transfer_0']['finish']['commit'] > 0 )
-            # self.assertTrue( aggs['transfer_1']['finish']['commit'] > 0 )
-            # self.assertTrue( aggs['transfer_2']['finish']['commit'] == 0 )
-            self.assertTrue( aggs['sumtotal_0']['isolation']  == 0)
-            self.assertTrue( aggs['sumtotal_1']['isolation']  == 0)
-            self.assertTrue( aggs['sumtotal_2']['isolation']  == 0)
-
-        # print('### join node3 ###')
-        # for proxy in ['rep31', 'rep32', 'rep23', 'rep13', 'arb31', 'arb32', 'arb23', 'arb13']:
-        #     self.toxiproxy.get_proxy(proxy).enable()
+        time.sleep(TEST_DURATION)
+        aggs = self.client.get_status()
+
+        self.assertTrue( aggs['transfer_0']['finish']['commit'] > 0 )
+        self.assertTrue( aggs['transfer_1']['finish']['commit'] > 0 )
+        self.assertTrue( 'commit' not in aggs['transfer_2']['finish'] )
+        self.assertTrue( aggs['sumtotal_0']['isolation'] == 0)
+        self.assertTrue( aggs['sumtotal_1']['isolation'] == 0)
+        self.assertTrue( aggs['sumtotal_2']['isolation'] == 0)
 
+        # join splitted node
+        failure.stop()
         # clear tx history
-        self.client.get_status()
+        self.client.get_status(print=False)
 
-        for i in range(5):
-            print(i, datetime.datetime.now())
-            time.sleep(3)
-            aggs = self.client.get_status()
-            MtmClient.print_aggregates(aggs)
-            self.assertTrue( aggs['sumtotal_0']['isolation']  == 0)
-            self.assertTrue( aggs['sumtotal_1']['isolation']  == 0)
-            self.assertTrue( aggs['sumtotal_2']['isolation']  == 0)
+        time.sleep(TEST_RECOVERY_TIME)
+        aggs = self.client.get_status()
 
-        # check that during last aggregation all nodes were working
         self.assertTrue( aggs['transfer_0']['finish']['commit'] > 0 )
         self.assertTrue( aggs['transfer_1']['finish']['commit'] > 0 )
         self.assertTrue( aggs['transfer_2']['finish']['commit'] > 0 )
-
+        self.assertTrue( aggs['sumtotal_0']['isolation'] == 0)
+        self.assertTrue( aggs['sumtotal_1']['isolation'] == 0)
+        self.assertTrue( aggs['sumtotal_2']['isolation'] == 0)
 
 
         subprocess.check_call(['blockade','join'])