nested aggregates

Author: kelvich

contrib/mmts/tests2/lib/bank_client.py

@@ -111,20 +111,24 @@ def status(self):
         while self.running:
             msg = yield from self.child_pipe.coro_recv()
             if msg == 'status':
-                # print('evloop: got status request')
                 serialized_aggs = {}
-                for name, aggregate in self.aggregates.items():
-                    serialized_aggs[name] = aggregate.as_dict() 
-                    aggregate.clear_values()
+
+                for conn_id, conn_aggs in self.aggregates.items():
+                    serialized_aggs[conn_id] = {}
+                    for aggname, agg in conn_aggs.items():
+                        serialized_aggs[conn_id][aggname] = agg.as_dict()
+                        agg.clear_values()
+
                 self.child_pipe.send(serialized_aggs)
-                # print('evloop: sent status response')
             else:
                 print('evloop: unknown message')
 
     @asyncio.coroutine
     def exec_tx(self, tx_block, aggname_prefix, conn_i):
         aggname = "%s_%i" % (aggname_prefix, conn_i)
-        agg = self.aggregates[aggname] = MtmTxAggregate(aggname)
+        if conn_i not in self.aggregates:
+            self.aggregates[conn_i] = {}
+        agg = self.aggregates[conn_i][aggname_prefix] = MtmTxAggregate(aggname)
         pool = yield from aiopg.create_pool(self.dsns[conn_i])
         conn = yield from pool.acquire()
         cur = yield from conn.cursor()
@@ -167,8 +171,8 @@ def total_tx(self, conn, cur, agg):
         total = yield from cur.fetchone()
         if total[0] != 0:
             agg.isolation += 1
-            # print(self.oops)
-            # print('Isolation error, total = ', total[0])
+            print(self.oops)
+            print('Isolation error, total = ', total[0])
             # yield from cur.execute('select * from mtm.get_nodes_state()')
             # nodes_state = yield from cur.fetchall()
             # for i, col in enumerate(self.nodes_state_fields):
@@ -177,7 +181,6 @@ def total_tx(self, conn, cur, agg):
             #          print("%19s" % nodes_state[j][i], end="\t")
             #     print("\n")
 
-
     def run(self):
         # asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
         self.loop = asyncio.get_event_loop()
@@ -196,13 +199,9 @@ def bgrun(self):
         self.evloop_process = multiprocessing.Process(target=self.run, args=())
         self.evloop_process.start()
 
-    # XXX: introduce periodic report from client?
     def get_aggregates(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
@@ -216,7 +215,7 @@ def stop(self):
         self.evloop_process.terminate()
 
     @classmethod
-    def print_aggregates(cls, serialized_agg):
+    def print_aggregates(cls, aggs):
             columns = ['running_latency', 'max_latency', 'isolation', 'finish']
 
             # print table header
@@ -225,23 +224,18 @@ def print_aggregates(cls, serialized_agg):
                 print(col, end="\t")
             print("\n", end="")
 
-            serialized_agg
-
-            for aggname in sorted(serialized_agg.keys()):
-                agg = serialized_agg[aggname]
-                print("%s\t" % aggname, end="")
-                for col in columns:
-                    if col in agg:
+            for conn_id in aggs.keys():
+                for aggname in aggs[conn_id].keys():
+                    agg = aggs[conn_id][aggname]
+                    print("Node %d: %s\t" % (conn_id + 1, aggname), end="")
+                    for col in columns:
                         if isinstance(agg[col], float):
                             print("%.2f\t" % (agg[col],), end="\t")
                         else:
                             print(agg[col], end="\t")
-                    else:
-                        print("-\t", end="")
-                print("")
+                    print("")
             print("")
 
-
 if __name__ == "__main__":
     c = MtmClient(['dbname=postgres user=postgres host=127.0.0.1',
         'dbname=postgres user=postgres host=127.0.0.1 port=5433',

contrib/mmts/tests2/lib/failure_injector.py

@@ -2,12 +2,14 @@
 
 class FailureInjector(object):
 
-    def __init__(self):
-        self.docker_api = docker.Client()
+    # def __init__(self):
+    #     self.docker_api = docker.Client()
 
     def container_exec(self, node, command):
+        self.docker_api = docker.Client()
         exec_id = self.docker_api.exec_create(node, command, user='root')
         output = self.docker_api.exec_start(exec_id)
+        self.docker_api.close()
         # print(command, ' -> ', output)
 
 
@@ -39,7 +41,7 @@ def start(self):
         self.container_exec(self.nodeA, "iptables -A OUTPUT -s {} -j DROP".format(self.nodeB) )
 
     def stop(self):
-        self.container_exec(self.nodeA, "iptables -D INPUT  -s {} -j DROP".format(self.nodeB))
+        self.container_exec(self.nodeA, "iptables -D INPUT -s {} -j DROP".format(self.nodeB))
         self.container_exec(self.nodeA, "iptables -D OUTPUT -s {} -j DROP".format(self.nodeB))
 
 

contrib/mmts/tests2/test_recovery.py

@@ -71,19 +71,19 @@ def test_node_partition(self):
         time.sleep(TEST_RECOVERY_TIME)
         aggs = self.client.get_aggregates()
 
-        self.assertTrue( 'commit' in aggs_failure['transfer_0']['finish'] )
-        self.assertTrue( 'commit' in aggs_failure['transfer_1']['finish'] )
-        self.assertTrue( 'commit' not in aggs_failure['transfer_2']['finish'] )
-        self.assertTrue( aggs_failure['sumtotal_0']['isolation'] == 0)
-        self.assertTrue( aggs_failure['sumtotal_1']['isolation'] == 0)
-        self.assertTrue( aggs_failure['sumtotal_2']['isolation'] == 0)
-
-        self.assertTrue( 'commit' in aggs['transfer_0']['finish'] )
-        self.assertTrue( 'commit' in aggs['transfer_1']['finish'] )
-        self.assertTrue( 'commit' 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)
+        self.assertTrue( 'commit' in aggs_failure[0]['transfer']['finish'] )
+        self.assertTrue( 'commit' in aggs_failure[1]['transfer']['finish'] )
+        self.assertTrue( 'commit' not in aggs_failure[2]['transfer']['finish'] )
+        self.assertTrue( aggs_failure[0]['sumtotal']['isolation'] == 0)
+        self.assertTrue( aggs_failure[1]['sumtotal']['isolation'] == 0)
+        self.assertTrue( aggs_failure[2]['sumtotal']['isolation'] == 0)
+
+        self.assertTrue( 'commit' in aggs[0]['transfer']['finish'] )
+        self.assertTrue( 'commit' in aggs[1]['transfer']['finish'] )
+        self.assertTrue( 'commit' in aggs[2]['transfer']['finish'] )
+        self.assertTrue( aggs[0]['sumtotal']['isolation'] == 0)
+        self.assertTrue( aggs[1]['sumtotal']['isolation'] == 0)
+        self.assertTrue( aggs[2]['sumtotal']['isolation'] == 0)
 
 
     def test_edge_partition(self):