Merge pull request #288 from pinheadmz/circuitbreaker

Author: willcl-ark

src/backends/backend_interface.py

@@ -6,6 +6,7 @@
 class ServiceType(Enum):
     BITCOIN = 1
     LIGHTNING = 2
+    CIRCUITBREAKER = 3
 
 
 class BackendInterface(ABC):

src/backends/compose/compose_backend.py

@@ -39,6 +39,8 @@
 GRAFANA_PROVISIONING = "grafana-provisioning"
 CONTAINER_PREFIX_BITCOIND = "tank-bitcoin"
 CONTAINER_PREFIX_LN = "tank-ln"
+CONTAINER_PREFIX_CIRCUITBREAKER = "tank-ln-cb"
+LND_MOUNT_PATH = '/root/.lnd'
 
 logger = logging.getLogger("docker-interface")
 logging.getLogger("docker.utils.config").setLevel(logging.WARNING)
@@ -91,15 +93,15 @@ def up(self, *_):
             )
 
     def down(self, warnet):
-        command = ["docker", "compose", "down"]
+        command = ["docker", "compose", "down", "-v"]
         try:
             with subprocess.Popen(
                 command,
                 cwd=str(self.config_dir),
                 stdout=subprocess.PIPE,
                 stderr=subprocess.STDOUT,
             ) as process:
-                logger.debug(f"Running docker compose down with PID {process.pid}")
+                logger.debug(f"Running 'docker compose down -v' with PID {process.pid}")
                 if process.stdout:
                     for line in process.stdout:
                         logger.info(line.decode().rstrip())
@@ -114,6 +116,8 @@ def get_container_name(self, tank_index: int, service: ServiceType) -> str:
                 return f"{self.network_name}-{CONTAINER_PREFIX_BITCOIND}-{tank_index:06}"
             case ServiceType.LIGHTNING:
                 return f"{self.network_name}-{CONTAINER_PREFIX_LN}-{tank_index:06}"
+            case ServiceType.CIRCUITBREAKER:
+                return f"{self.network_name}-{CONTAINER_PREFIX_CIRCUITBREAKER}-{tank_index:06}"
             case _:
                 raise Exception("Unsupported service type")
 
@@ -301,7 +305,7 @@ def _write_docker_compose(self, warnet):
 
         # Pass services object to each tank so they can add whatever they need.
         for tank in warnet.tanks:
-            self.add_services(tank, compose["services"])
+            self.add_services(tank, compose)
 
         # Initialize services and add them to the compose
         services = [
@@ -362,7 +366,8 @@ def copy_configs(self, tank):
         shutil.copyfile(TEMPLATES / ENTRYPOINT_NAME, tank.config_dir / ENTRYPOINT_NAME)
         set_execute_permission(tank.config_dir / ENTRYPOINT_NAME)
 
-    def add_services(self, tank: Tank, services):
+    def add_services(self, tank: Tank, compose):
+        services = compose["services"]
         assert tank.index is not None
         container_name = self.get_container_name(tank.index, ServiceType.BITCOIN)
         services[container_name] = {}
@@ -411,7 +416,7 @@ def add_services(self, tank: Tank, services):
             services[container_name]["labels"].update({"collect_logs": True})
 
         if tank.lnnode is not None:
-            self.add_lnd_service(tank, services)
+            self.add_lnd_service(tank, compose)
 
         # Add the prometheus data exporter in a neighboring container
         if tank.exporter:
@@ -427,8 +432,10 @@ def add_services(self, tank: Tank, services):
                 "networks": [tank.network_name],
             }
 
-    def add_lnd_service(self, tank, services):
+    def add_lnd_service(self, tank, compose):
+        services = compose["services"]
         ln_container_name = self.get_container_name(tank.index, ServiceType.LIGHTNING)
+        ln_cb_container_name = self.get_container_name(tank.index, ServiceType.CIRCUITBREAKER)
         bitcoin_container_name = self.get_container_name(tank.index, ServiceType.BITCOIN)
         # These args are appended to the Dockerfile `ENTRYPOINT ["lnd"]`
         args = [
@@ -447,6 +454,7 @@ def add_lnd_service(self, tank, services):
             f"--externalip={tank.lnnode.ipv4}",
             f"--rpclisten=0.0.0.0:{tank.lnnode.rpc_port}",
             f"--alias={tank.index}",
+            f"--tlsextradomain={ln_container_name}",
         ]
         services[ln_container_name] = {
             "container_name": ln_container_name,
@@ -477,6 +485,23 @@ def add_lnd_service(self, tank, services):
         )
         if tank.collect_logs:
             services[ln_container_name]["labels"].update({"collect_logs": True})
+        if tank.lnnode.cb is not None:
+            services[ln_container_name].update({
+                "volumes": [f"{ln_container_name}-data:{LND_MOUNT_PATH}"]
+            })
+            services[ln_cb_container_name] = {
+                "container_name": ln_cb_container_name,
+                "image": tank.lnnode.cb,
+                "volumes": [f"{ln_container_name}-data:{LND_MOUNT_PATH}"],
+                "command": "--network=regtest " +
+                           f"--rpcserver={ln_container_name}:{tank.lnnode.rpc_port} " +
+                           f" --tlscertpath={LND_MOUNT_PATH}/tls.cert " +
+                           f" --macaroonpath={LND_MOUNT_PATH}/data/chain/bitcoin/regtest/admin.macaroon",
+                "networks": [tank.network_name],
+                "restart": "on-failure",
+            }
+            compose["volumes"].update({f"{ln_container_name}-data": None})
+
 
     def get_ipv4_address(self, container: Container) -> str:
         """

src/backends/kubernetes/kubernetes_backend.py

@@ -23,8 +23,10 @@
 POD_PREFIX = "tank"
 BITCOIN_CONTAINER_NAME = "bitcoin"
 LN_CONTAINER_NAME = "ln"
+LN_CB_CONTAINER_NAME = "ln-cb"
 MAIN_NAMESPACE = "warnet"
 PROMETHEUS_METRICS_PORT = 9332
+LND_MOUNT_PATH = '/root/.lnd'
 
 
 logger = logging.getLogger("KubernetesBackend")
@@ -107,7 +109,7 @@ def get_file(self, tank_index: int, service: ServiceType, file_path: str):
         return decoded_bytes
 
     def get_pod_name(self, tank_index: int, type: ServiceType) -> str:
-        if type == ServiceType.LIGHTNING:
+        if type == ServiceType.LIGHTNING or type == ServiceType.CIRCUITBREAKER:
             return f"{self.network_name}-{POD_PREFIX}-ln-{tank_index:06d}"
         return f"{self.network_name}-{POD_PREFIX}-{tank_index:06d}"
 
@@ -173,18 +175,19 @@ def get_status(self, tank_index: int, service: ServiceType) -> RunningStatus:
 
     def exec_run(self, tank_index: int, service: ServiceType, cmd: str):
         pod_name = self.get_pod_name(tank_index, service)
-        if service == ServiceType.BITCOIN:
-            exec_cmd = ["/bin/bash", "-c", f"{cmd}"]
-        elif service == ServiceType.LIGHTNING:
-            exec_cmd = ["/bin/sh", "-c", f"{cmd}"]
+        exec_cmd = ["/bin/sh", "-c", f"{cmd}"]
         self.log.debug(f"Running {exec_cmd=:} on {tank_index=:}")
+        if service == ServiceType.BITCOIN:
+            container = BITCOIN_CONTAINER_NAME
+        if service == ServiceType.LIGHTNING:
+            container = LN_CONTAINER_NAME
+        if service == ServiceType.CIRCUITBREAKER:
+            container = LN_CB_CONTAINER_NAME
         result = stream(
             self.client.connect_get_namespaced_pod_exec,
             pod_name,
             self.namespace,
-            container=BITCOIN_CONTAINER_NAME
-            if service == ServiceType.BITCOIN
-            else LN_CONTAINER_NAME,
+            container=container,
             command=exec_cmd,
             stderr=True,
             stdin=False,
@@ -432,7 +435,7 @@ def remove_prometheus_service_monitors(self, tanks):
                 if e.status != 404:
                     raise e
 
-    def create_lnd_container(self, tank, bitcoind_service_name) -> client.V1Container:
+    def create_lnd_container(self, tank, bitcoind_service_name, volume_mounts) -> client.V1Container:
         # These args are appended to the Dockerfile `ENTRYPOINT ["lnd"]`
         bitcoind_rpc_host = f"{bitcoind_service_name}.{self.namespace}"
         lightning_dns = f"lightning-{tank.index}.{self.namespace}"
@@ -475,12 +478,33 @@ def create_lnd_container(self, tank, bitcoind_service_name) -> client.V1Containe
                 privileged=True,
                 capabilities=client.V1Capabilities(add=["NET_ADMIN", "NET_RAW"]),
             ),
+            volume_mounts=volume_mounts,
         )
         self.log.debug(f"Created lightning container for tank {tank.index}")
         return lightning_container
 
+    def create_circuitbreaker_container(self, tank, volume_mounts) -> client.V1Container:
+        self.log.debug(f"Creating circuitbreaker container for tank {tank.index}")
+        cb_container = client.V1Container(
+            name=LN_CB_CONTAINER_NAME,
+            image=tank.lnnode.cb,
+            args=[
+                "--network=regtest",
+                f"--rpcserver=localhost:{tank.lnnode.rpc_port}",
+                f"--tlscertpath={LND_MOUNT_PATH}/tls.cert",
+                f"--macaroonpath={LND_MOUNT_PATH}/data/chain/bitcoin/regtest/admin.macaroon"
+            ],
+            security_context=client.V1SecurityContext(
+                privileged=True,
+                capabilities=client.V1Capabilities(add=["NET_ADMIN", "NET_RAW"]),
+            ),
+            volume_mounts=volume_mounts,
+        )
+        self.log.debug(f"Created circuitbreaker container for tank {tank.index}")
+        return cb_container
+
     def create_pod_object(
-        self, tank: Tank, container: client.V1Container, name: str
+        self, tank: Tank, containers: list[client.V1Container], volumes: list[client.V1Volume], name: str
     ) -> client.V1Pod:
         # Create and return a Pod object
         # TODO: pass a custom namespace , e.g. different warnet sims can be deployed into diff namespaces
@@ -500,7 +524,8 @@ def create_pod_object(
                 # Might need some more thinking on the pod restart policy, setting to Never for now
                 # This means if a node has a problem it dies
                 restart_policy="OnFailure",
-                containers=[container],
+                containers=containers,
+                volumes=volumes,
             ),
         )
 
@@ -588,7 +613,7 @@ def deploy_pods(self, warnet):
             # Create and deploy bitcoind pod and service
             bitcoind_container = self.create_bitcoind_container(tank)
             bitcoind_pod = self.create_pod_object(
-                tank, bitcoind_container, self.get_pod_name(tank.index, ServiceType.BITCOIN)
+                tank, [bitcoind_container], [], self.get_pod_name(tank.index, ServiceType.BITCOIN)
             )
 
             if tank.exporter and self.check_logging_crds_installed():
@@ -609,11 +634,30 @@ def deploy_pods(self, warnet):
 
             # Create and deploy LND pod
             if tank.lnnode:
-                lnd_container = self.create_lnd_container(tank, bitcoind_service.metadata.name)
+                conts = []
+                vols = []
+                volume_mounts = []
+                if tank.lnnode.cb:
+                    # Create a shared volume between containers in the pod
+                    volume_name = f"ln-cb-data-{tank.index}"
+                    vols.append(client.V1Volume(
+                        name=volume_name,
+                        empty_dir=client.V1EmptyDirVolumeSource()
+                    ))
+                    volume_mounts.append(client.V1VolumeMount(
+                        name=volume_name,
+                        mount_path=LND_MOUNT_PATH,
+                    ))
+                    # Add circuit breaker container
+                    conts.append(self.create_circuitbreaker_container(tank, volume_mounts))
+                # Add lnd container
+                conts.append(self.create_lnd_container(tank, bitcoind_service.metadata.name, volume_mounts))
+                # Put it all together in a pod
                 lnd_pod = self.create_pod_object(
-                    tank, lnd_container, self.get_pod_name(tank.index, ServiceType.LIGHTNING)
+                    tank, conts, vols, self.get_pod_name(tank.index, ServiceType.LIGHTNING)
                 )
                 self.client.create_namespaced_pod(namespace=self.namespace, body=lnd_pod)
+                # Create service for the pod
                 lightning_service = self.create_lightning_service(tank)
                 try:
                     self.client.delete_namespaced_service(

src/cli/network.py

@@ -122,9 +122,12 @@ def status(network: str):
     assert isinstance(result, list), "Result is not a list"  # Make mypy happy
     for tank in result:
         lightning_status = ""
+        circuitbreaker_status = ""
         if "lightning_status" in tank:
             lightning_status = f"\tLightning: {tank['lightning_status']}"
-        print(f"Tank: {tank['tank_index']} \tBitcoin: {tank['bitcoin_status']}{lightning_status}")
+        if "circuitbreaker_status" in tank:
+            circuitbreaker_status = f"\tCircuit Breaker: {tank['circuitbreaker_status']}"
+        print(f"Tank: {tank['tank_index']} \tBitcoin: {tank['bitcoin_status']}{lightning_status}{circuitbreaker_status}")
 
 
 @network.command()

src/templates/node_schema.json

@@ -12,7 +12,8 @@
     "collect_logs": {"type": "boolean", "default": false},
     "build_args": {"type": "string"},
     "ln": {"type": "string"},
-    "ln-image": {"type": "string"}
+    "ln-image": {"type": "string"},
+    "ln-cb-image": {"type": "string"}
   },
   "additionalProperties": false,
   "oneOf": [

src/warnet/lnnode.py

@@ -7,7 +7,7 @@
 
 
 class LNNode:
-    def __init__(self, warnet, tank, impl, image, backend: BackendInterface):
+    def __init__(self, warnet, tank, impl, image, backend: BackendInterface, cb=None):
         self.warnet = warnet
         self.tank = tank
         assert impl == "lnd"
@@ -16,6 +16,7 @@ def __init__(self, warnet, tank, impl, image, backend: BackendInterface):
         if image:
             self.image = image
         self.backend = backend
+        self.cb = cb
         self.ipv4 = generate_ipv4_addr(self.warnet.subnet)
         self.rpc_port = 10009
 
@@ -26,6 +27,12 @@ def __str__(self):
     def status(self) -> RunningStatus:
         return self.warnet.container_interface.get_status(self.tank.index, ServiceType.LIGHTNING)
 
+    @property
+    def cb_status(self) -> RunningStatus:
+        if not self.cb:
+            return None
+        return self.warnet.container_interface.get_status(self.tank.index, ServiceType.CIRCUITBREAKER)
+
     @exponential_backoff(max_retries=20, max_delay=300)
     @handle_json
     def lncli(self, cmd) -> dict:

src/warnet/server.py

@@ -21,6 +21,7 @@
 import jsonschema
 import networkx as nx
 import scenarios
+from backends import ServiceType
 from flask import Flask, jsonify, request
 from flask_jsonrpc.app import JSONRPC
 from flask_jsonrpc.exceptions import ServerError
@@ -180,6 +181,7 @@ def setup_rpc(self):
         self.jsonrpc.register(self.graph_validate)
         # Debug
         self.jsonrpc.register(self.generate_deployment)
+        self.jsonrpc.register(self.exec_run)
         # Server
         self.jsonrpc.register(self.server_stop)
         # Logs
@@ -563,6 +565,8 @@ def network_status(self, network: str = "warnet") -> list[dict]:
                 status = {"tank_index": tank.index, "bitcoin_status": tank.status.name.lower()}
                 if tank.lnnode is not None:
                     status["lightning_status"] = tank.lnnode.status.name.lower()
+                    if tank.lnnode.cb is not None:
+                        status["circuitbreaker_status"] = tank.lnnode.cb_status.name.lower()
                 stats.append(status)
             return stats
         except Exception as e:
@@ -612,6 +616,14 @@ def logs_grep(self, pattern: str, network: str = "warnet") -> str:
             self.logger.error(msg)
             raise ServerError(message=msg) from e
 
+    def exec_run(self, index: int, service_type: int, cmd: str, network: str = "warnet") -> str:
+        """
+        Execute an arbitrary command in an arbitrary container,
+        identified by tank index and ServiceType
+        """
+        wn = Warnet.from_network(network, self.backend)
+        return wn.container_interface.exec_run(index, ServiceType(service_type), cmd)
+
 
 def run_server():
     parser = argparse.ArgumentParser(description="Run the server")

src/warnet/tank.py

@@ -75,7 +75,8 @@ def parse_graph_node(self, node):
         if "ln" in node:
             impl = node["ln"]
             image = node.get("ln-image", None)
-            self.lnnode = LNNode(self.warnet, self, impl, image, self.warnet.container_interface)
+            cb_image = node.get("ln-cb-image", None)
+            self.lnnode = LNNode(self.warnet, self, impl, image, self.warnet.container_interface, cb_image)
 
         self.config_dir = self.warnet.config_dir / str(self.suffix)
         self.config_dir.mkdir(parents=True, exist_ok=True)

test/data/ln.graphml

@@ -4,6 +4,7 @@
   <key attr.name="tc_netem"       attr.type="string" for="node" id="tc_netem"/>
   <key attr.name="ln"             attr.type="string" for="node" id="ln"/>
   <key attr.name="ln-image"       attr.type="string" for="node" id="ln-image"/>
+  <key attr.name="ln-cb-image"    attr.type="string" for="node" id="ln-cb-image"/>
   <key attr.name="channel"        attr.type="string" for="edge" id="channel"/>
   <key attr.name="collect_logs"   attr.type="boolean" for="node" id="collect_logs"/>
   <key attr.name="image"          attr.type="string" for="node" id="image"/>
@@ -19,12 +20,14 @@
         <data key="bitcoin_config">-uacomment=w1</data>
         <data key="ln">lnd</data>
         <data key="ln-image">lightninglabs/lnd:v0.15.5-beta</data>
+        <data key="ln-cb-image">pinheadmz/circuitbreaker:278737d</data>
         <data key="collect_logs">true</data>
     </node>
     <node id="2">
         <data key="version">26.0</data>
         <data key="bitcoin_config">-uacomment=w2</data>
         <data key="ln">lnd</data>
+        <data key="ln-cb-image">pinheadmz/circuitbreaker:278737d</data>
     </node>
     <node id="3">
         <data key="version">26.0</data>