Chaos Monkey now restarts nodes

Two new modes in chaos monkey:
- `node_restart` ocasionally restarts one of the nodes that are initially running (but always just one)
- `node_set` maintains an extra set of nodes (which are initially shut down), and prediodically restarts them arbitrarily, ocasionally also wiping out their data folders.

Author: SkidanovAlex

Cargo.lock

@@ -1,7 +1,9 @@
+import multiprocessing
 import threading
 import subprocess
 import json
 import os
+import signal
 import atexit
 import shutil
 import requests
@@ -60,22 +62,22 @@ def _get_command_line(self, near_root, node_dir, boot_key, boot_node_addr):
     def wait_for_rpc(self, timeout=1):
         retry.retry(lambda: self.get_status(), timeout=timeout)
 
-    def json_rpc(self, method, params):
+    def json_rpc(self, method, params, timeout=2):
         j = {
             'method': method,
             'params': params,
             'id': 'dontcare',
             'jsonrpc': '2.0'
         }
-        r = requests.post("http://%s:%s" % self.rpc_addr(), json=j, timeout=1)
+        r = requests.post("http://%s:%s" % self.rpc_addr(), json=j, timeout=timeout)
         r.raise_for_status()
         return json.loads(r.content)
 
     def send_tx(self, signed_tx):
         return self.json_rpc('broadcast_tx_async', [base64.b64encode(signed_tx).decode('utf8')])
 
     def get_status(self):
-        r = requests.get("http://%s:%s/status" % self.rpc_addr(), timeout=1)
+        r = requests.get("http://%s:%s/status" % self.rpc_addr(), timeout=2)
         r.raise_for_status()
         return json.loads(r.content)
 
@@ -106,7 +108,7 @@ def __init__(self, port, rpc_port, near_root, node_dir):
         self.node_key = Key.from_json_file(os.path.join(node_dir, "node_key.json"))
         self.signer_key = Key.from_json_file(os.path.join(node_dir, "validator_key.json"))
 
-        self.handle = None
+        self.pid = multiprocessing.Value('i', 0)
 
         atexit.register(atexit_cleanup, self)
 
@@ -124,14 +126,17 @@ def start(self, boot_key, boot_node_addr):
         self.stderr = open(os.path.join(self.node_dir, 'stderr'), 'a')
         cmd = self._get_command_line(
             self.near_root, self.node_dir, boot_key, boot_node_addr)
-        self.handle = subprocess.Popen(
-            cmd, stdout=self.stdout, stderr=self.stderr, env=env)
-        self.wait_for_rpc()
+        self.pid.value = subprocess.Popen(
+            cmd, stdout=self.stdout, stderr=self.stderr, env=env).pid
+        self.wait_for_rpc(5)
 
     def kill(self):
-        if self.handle is not None:
-            self.handle.kill()
-            self.handle = None
+        if self.pid.value != 0:
+            os.kill(self.pid.value, signal.SIGKILL)
+            self.pid.value = 0
+
+    def reset_data(self):
+        shutil.rmtree(os.path.join(self.node_dir, "data"))
 
     def cleanup(self):
         self.kill()

pytest/lib/cluster.py

@@ -25,8 +25,12 @@ def stop_network(pid):
         f.write(str(pid))
 
 def resume_network(pid):
-    with open('/sys/fs/cgroup/net_cls/tasks', 'w') as f:
-        f.write(str(pid))
+    try:
+        with open('/sys/fs/cgroup/net_cls/tasks', 'w') as f:
+            f.write(str(pid))
+    except ProcessLookupError:
+        # the process was killed in the meantime
+        pass
 
 if __name__ == "__main__":
     import time

pytest/lib/network.py

@@ -38,7 +38,7 @@ def monkey_network_hammering(stopped, error, nodes, nonces):
     s = [False for x in nodes]
     while stopped.value == 0:
         node_idx = random.randint(0, len(nodes) - 2)
-        pid = nodes[node_idx].handle.pid
+        pid = nodes[node_idx].pid.value
         if s[node_idx]:
             print("Resuming network for process %s" % pid)
             resume_network(pid)
@@ -50,7 +50,7 @@ def monkey_network_hammering(stopped, error, nodes, nonces):
         time.sleep(0.5)
     for i, x in enumerate(s):
         if x:
-            pid = nodes[i].handle.pid
+            pid = nodes[i].pid.value
             print("Resuming network for process %s" % pid)
             resume_network(pid)