test_referee test version without hardcoded sleep times and fsync=on

Author: kelvich

tests2/docker-entrypoint.sh

@@ -51,7 +51,7 @@ if [ "$1" = 'postgres' ]; then
 			listen_addresses='*' 
 			max_prepared_transactions = 100
 			synchronous_commit = on
-			fsync = off
+			fsync = on
 			wal_level = logical
 			max_worker_processes = 30
 			max_replication_slots = 10
@@ -62,11 +62,11 @@ if [ "$1" = 'postgres' ]; then
 			# log_statement = all
 
 			multimaster.max_nodes = 3
-			# multimaster.volkswagen_mode = 1
-			# multimaster.ignore_tables_without_pk = 1
 			multimaster.heartbeat_recv_timeout = 1100
 			multimaster.heartbeat_send_timeout = 250
-			multimaster.max_recovery_lag = 1000000000
+			multimaster.max_recovery_lag = 1GB
+			multimaster.min_recovery_lag = 10kB
+			multimaster.preserve_commit_order = off
 		EOF
 
 		if [ -n "$NODE_ID" ]; then

tests2/lib/bank_client.py

@@ -159,14 +159,15 @@ def no_prepared_tx(self):
     def status(self):
         while self.running:
             msg = yield from self.child_pipe.coro_recv()
-            if msg == 'status':
+            if msg == 'status' or msg == 'status_noclean':
                 serialized_aggs = []
 
                 for conn_id, conn_aggs in self.aggregates.items():
                     serialized_aggs.append({})
                     for aggname, agg in conn_aggs.items():
                         serialized_aggs[conn_id][aggname] = agg.as_dict()
-                        agg.clear_values()
+                        if msg == 'status':
+                            agg.clear_values()
 
                 yield from self.child_pipe.coro_send(serialized_aggs)
             else:
@@ -284,9 +285,14 @@ def bgrun(self):
         self.evloop_process = multiprocessing.Process(target=self.run, args=())
         self.evloop_process.start()
 
-    def get_aggregates(self, _print=True):
-        self.parent_pipe.send('status')
+    def get_aggregates(self, _print=True, clean=True):
+        if clean:
+            self.parent_pipe.send('status')
+        else:
+            self.parent_pipe.send('status_noclean')
+
         resp = self.parent_pipe.recv()
+
         if _print:
             MtmClient.print_aggregates(resp)
         return resp

tests2/lib/test_helper.py

@@ -5,6 +5,7 @@
 
 TEST_WARMING_TIME = 5
 TEST_DURATION = 10
+TEST_MAX_RECOVERY_TIME = 300
 TEST_RECOVERY_TIME = 30
 TEST_SETUP_TIME = 20
 TEST_STOP_DELAY = 5
@@ -34,7 +35,20 @@ def assertNoCommits(self, aggs):
         if commits:
             raise AssertionError('There are commits during aggregation interval')
 
-    def performFailure(self, failure, wait=0):
+    def awaitCommit(self, node_id):
+        total_sleep = 0
+
+        while total_sleep <= TEST_MAX_RECOVERY_TIME:
+            aggs = self.client.get_aggregates(clean=False, _print=False)
+            # print('=== ',aggs[node_id]['transfer']['finish'])
+            if ('commit' in aggs[node_id]['transfer']['finish'] and
+                    aggs[node_id]['transfer']['finish']['commit'] > 10):
+                break
+            time.sleep(5)
+            total_sleep += 5
+
+
+    def performFailure(self, failure, wait=0, node_wait_for_commit=-1):
 
         time.sleep(TEST_WARMING_TIME)
 
@@ -57,9 +71,13 @@ def performFailure(self, failure, wait=0):
         print('Eliminate failure at ',datetime.datetime.utcnow())
 
         self.client.clean_aggregates()
-        time.sleep(TEST_RECOVERY_TIME)
-        aggs = self.client.get_aggregates()
 
+        if node_wait_for_commit >= 0:
+            self.awaitCommit(node_wait_for_commit)
+        else:
+            time.sleep(TEST_RECOVERY_TIME)
+
+        aggs = self.client.get_aggregates()
         return (aggs_failure, aggs)
 
     def nodeExecute(dsn, statements):

tests2/test_referee.py

@@ -25,9 +25,6 @@ def setUpClass(cls):
             '--build',
             '-d'])
 
-        # XXX: add normal wait here
-        time.sleep(TEST_SETUP_TIME)
-
         cls.client = MtmClient([
             "dbname=regression user=postgres host=127.0.0.1 port=15432",
             "dbname=regression user=postgres host=127.0.0.1 port=15433"
@@ -42,8 +39,6 @@ def tearDownClass(cls):
         print('tearDown')
         cls.client.stop()
 
-        time.sleep(TEST_STOP_DELAY)
-
         if not cls.client.is_data_identic():
             raise AssertionError('Different data on nodes')
 
@@ -55,7 +50,7 @@ def tearDownClass(cls):
 
     def setUp(self):
         warnings.simplefilter("ignore", ResourceWarning)
-        time.sleep(20)
+
         print('Start new test at ',datetime.datetime.utcnow())
 
     def tearDown(self):
@@ -64,7 +59,7 @@ def tearDown(self):
     def test_node_crash(self):
         print('### test_node_crash ###')
 
-        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node2'))
+        aggs_failure, aggs = self.performFailure(CrashRecoverNode('node2'), node_wait_for_commit=1)
 
         self.assertCommits(aggs_failure[:1])
         self.assertNoCommits(aggs_failure[1:])
@@ -77,7 +72,7 @@ def test_node_crash(self):
     def test_partition_referee(self):
         print('### test_partition_referee ###')
 
-        aggs_failure, aggs = self.performFailure(SingleNodePartition('node2'))
+        aggs_failure, aggs = self.performFailure(SingleNodePartition('node2'), node_wait_for_commit=1)
 
         self.assertCommits(aggs_failure[:1])
         self.assertNoCommits(aggs_failure[1:])
@@ -89,7 +84,7 @@ def test_partition_referee(self):
     def test_double_failure_referee(self):
         print('### test_double_failure_referee ###')
 
-        aggs_failure, aggs = self.performFailure(SingleNodePartition('node2'))
+        aggs_failure, aggs = self.performFailure(SingleNodePartition('node2'), node_wait_for_commit=1)
 
         self.assertCommits(aggs_failure[:1])
         self.assertNoCommits(aggs_failure[1:])
@@ -98,7 +93,7 @@ def test_double_failure_referee(self):
         self.assertCommits(aggs)
         self.assertIsolation(aggs)
 
-        aggs_failure, aggs = self.performFailure(SingleNodePartition('node1'))
+        aggs_failure, aggs = self.performFailure(SingleNodePartition('node1'), node_wait_for_commit=0)
 
         self.assertNoCommits(aggs_failure[:1])
         self.assertCommits(aggs_failure[1:])
@@ -121,7 +116,7 @@ def test_winner_restart(self):
         self.assertCommits(aggs[1:])
         self.assertIsolation(aggs)
 
-        aggs_failure, aggs = self.performFailure(RestartNode('node2'))
+        aggs_failure, aggs = self.performFailure(RestartNode('node2'), node_wait_for_commit=1)
 
         self.assertNoCommits(aggs_failure)
         self.assertIsolation(aggs_failure)
@@ -133,7 +128,7 @@ def test_winner_restart(self):
         # need to start node1 to perform consequent tests
         docker_api = docker.from_env()
         docker_api.containers.get('node1').start()
-        time.sleep(35)
+        self.awaitCommit(0)
 
 
 if __name__ == '__main__':