Implement addconnection key in graph files to specify post-deployment connections with specific types

docs/tank-options.md

@@ -31,6 +31,22 @@ addnode:
 
 ---
 
+## `addconnection`
+
+List of objects with peer names this node will connect to after network deployment is completed. Nodes are addressed by their `name` (for same-namespace peers) or by `<name>.default` for cross-namespace connections on the battlefield.
+Relies on `addconnection_init` scenario and calls RPC `addconnection` with connection `type` (default `outbound-full-relay`) and `v2` (default true) options.
+Will only work on tanks running at least Bitcoin Core v22.0
+
+```yaml
+addconnection:
+  - to: tank-0001 # use all defaults
+  - to: tank-0002
+    type: blocks-only
+    v2: false
+```
+
+---
+
 ## `image`
 
 Docker image for the Bitcoin Core container.

resources/charts/bitcoincore/templates/pod.yaml

@@ -19,6 +19,7 @@ metadata:
     {{- end }}
   annotations:
     init_peers: "{{ .Values.addnode | len }}"
+    addconnection_peers: {{ .Values.addconnection | toJson | quote }}
 spec:
   restartPolicy: "{{ .Values.restartPolicy }}"
   {{- with .Values.imagePullSecrets }}

resources/charts/bitcoincore/values.yaml

@@ -9,7 +9,7 @@ image:
   repository: bitcoindevproject/bitcoin
   pullPolicy: IfNotPresent
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "27.0"
+  tag: "31.0"
 
 imagePullSecrets: []
 nameOverride: ""
@@ -148,6 +148,7 @@ config: ""
 defaultConfig: ""
 
 addnode: []
+addconnection: []
 
 loadSnapshot:
   enabled: false

resources/scenarios/addconnection_init.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+
+import threading
+from time import sleep
+
+from commander import Commander
+
+DEFAULT_CONNECTION_TYPE = "outbound-full-relay"
+DEFAULT_TRANSPORT_PROTOCOL_TYPE = "v2"
+
+
+class AddConnectionInit(Commander):
+    def set_test_params(self):
+        self.num_nodes = None
+
+    def add_options(self, parser):
+        parser.description = "Connect tanks with specific connection types after deployment"
+        parser.usage = "warnet run /path/to/addconnection_init.py"
+
+    def run_test(self):
+        self.log.info("Waiting for RPC availability and addnode connections...")
+        self.wait_for_tanks_connected()
+
+        self.log.info("Connecting addconnection tanks...")
+
+        def addconnection(self, node, peer, conn_type, v2):
+            while True:
+                try:
+                    result = node.addconnection(peer, conn_type, v2)
+                    self.log.info(f"Connected {node.tank} to {peer}: {result}")
+                    break
+                except Exception as e:
+                    self.log.info(
+                        f"Couldn't connect {node.tank} to {peer}: {e}, retrying in 5 seconds..."
+                    )
+                    sleep(5)
+
+        conn_threads = []
+
+        for node in self.nodes:
+            for connection in node.addconnection_peers:
+                self.log.info(f"Connecting {node.tank} {connection}")
+                conn_threads.append(
+                    threading.Thread(
+                        target=addconnection,
+                        args=(
+                            self,
+                            node,
+                            connection["to"],
+                            connection.get("type", DEFAULT_CONNECTION_TYPE),
+                            connection.get("v2", DEFAULT_TRANSPORT_PROTOCOL_TYPE == "v2"),
+                        ),
+                    )
+                )
+
+        for thread in conn_threads:
+            thread.start()
+
+        all(thread.join() is None for thread in conn_threads)
+        self.log.info("All addconnection commands executed, waiting for confirmation...")
+
+        def check_addconnections(self, node):
+            expected = node.addconnection_peers
+            if not expected:
+                return
+            poll = True
+            while poll:
+                actual = node.getpeerinfo()
+                for expected_connection in expected:
+                    if any(
+                        actual_connection["addr"] == expected_connection["to"]
+                        and actual_connection["connection_type"]
+                        == (expected_connection.get("type", DEFAULT_CONNECTION_TYPE))
+                        and actual_connection["transport_protocol_type"]
+                        == (
+                            "v1"
+                            if "v2" in expected_connection and not expected_connection["v2"]
+                            else DEFAULT_TRANSPORT_PROTOCOL_TYPE
+                        )
+                        for actual_connection in actual
+                    ):
+                        self.log.info(f"Connection complete: {node.tank} {expected_connection}")
+                        poll = False
+                    else:
+                        self.log.info(
+                            f"Connection incomplete: {node.tank} {expected_connection}, retrying in 5 seconds..."
+                        )
+                        sleep(5)
+                        poll = True
+                        break
+
+        check_threads = [
+            threading.Thread(target=check_addconnections, args=(self, node)) for node in self.nodes
+        ]
+        for thread in check_threads:
+            thread.start()
+
+        all(thread.join() is None for thread in check_threads)
+        self.log.info("All addconnection connections are complete")
+
+
+def main():
+    AddConnectionInit("").main()
+
+
+if __name__ == "__main__":
+    main()

resources/scenarios/commander.py

@@ -95,7 +95,6 @@
         sleep(5)
         pod = sclient.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
         pod_ip = pod.status.pod_ip
-
     if pod.metadata.labels["mission"] == "tank":
         WARNET["tanks"].append(
             {
@@ -107,7 +106,8 @@
                 "rpc_port": int(pod.metadata.labels["RPCPort"]),
                 "rpc_user": "user",
                 "rpc_password": pod.metadata.labels["rpcpassword"],
-                "init_peers": pod.metadata.annotations["init_peers"],
+                "init_peers": int(pod.metadata.annotations["init_peers"]),
+                "addconnection_peers": json.loads(pod.metadata.annotations["addconnection_peers"]),
             }
         )
 
@@ -281,7 +281,8 @@ def setup(self):
                 coveragedir=self.options.coveragedir,
             )
             node.rpc_connected = True
-            node.init_peers = int(tank["init_peers"])
+            node.init_peers = tank["init_peers"]
+            node.addconnection_peers = tank["addconnection_peers"]
             node.p2pport = tank["p2pport"]
 
             self.nodes.append(node)

resources/scenarios/test_scenarios/connect_dag.py

@@ -1,6 +1,5 @@
 #!/usr/bin/env python3
 
-import os
 from enum import Enum, auto, unique
 
 # The base class exists inside the commander container
@@ -16,90 +15,43 @@ class ConnectionType(Enum):
 class ConnectDag(Commander):
     def set_test_params(self):
         # This is just a minimum
-        self.num_nodes = 10
+        self.num_nodes = 1
 
     def add_options(self, parser):
-        parser.description = "Connect a complete DAG from a set of unconnected nodes"
+        parser.description = "Connect some nodes"
         parser.usage = "warnet run /path/to/scenario_connect_dag.py"
 
     def run_test(self):
-        # All permutations of a directed acyclic graph with zero, one, or two inputs/outputs
-        #
-        # │ Node │ In │ Out │ Con In │ Con Out │
-        # ├──────┼────┼─────┼────────┼─────────┤
-        # │  A   │  0 │   1 │ ─      │ C       │
-        # │  B   │  0 │   2 │ ─      │ C, D    │
-        # │  C   │  2 │   2 │ A, B   │ D, E    │
-        # │  D   │  2 │   1 │ B, C   │ F       │
-        # │  E   │  2 │   0 │ C, F   │ ─       │
-        # │  F   │  1 │   2 │ D      │ E, G    │
-        # │  G   │  1 │   1 │ F      │ H       │
-        # │  H   │  1 │   0 │ G      │ ─       │
-        #
-        #           Node Graph                 Corresponding Indices
-        #  ┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈    ┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈┈
-        #          ╭──────> E                      ╭──────> 4
-        #          │        ∧                      │        ∧
-        #  A ─> C ─┤        │              0 ─> 2 ─┤        │
-        #       ∧  ╰─> D ─> F ─> G ─> H         ∧  ╰─> 3 ─> 5 ─> 6 ─> 7
-        #       │      ∧                        │      ∧
-        #  B ───┴──────╯                   1 ───┴──────╯
-
         self.connect_nodes(0, 2)
         self.connect_nodes(1, 2)
         self.connect_nodes(1, 3)
-        self.connect_nodes(2, 3)
-        self.connect_nodes(2, 4)
-        self.connect_nodes(3, 5)
-        self.connect_nodes(5, 4)
-        self.connect_nodes(5, 6)
-        self.connect_nodes(6, 7)
+
         self.sync_all()
 
-        zero_peers = self.tanks["tank-0000"].getpeerinfo()
-        one_peers = self.tanks["tank-0001"].getpeerinfo()
-        two_peers = self.tanks["tank-0002"].getpeerinfo()
-        three_peers = self.tanks["tank-0003"].getpeerinfo()
-        four_peers = self.tanks["tank-0004"].getpeerinfo()
-        five_peers = self.tanks["tank-0005"].getpeerinfo()
-        six_peers = self.tanks["tank-0006"].getpeerinfo()
-        seven_peers = self.tanks["tank-0007"].getpeerinfo()
-        eight_peers = self.tanks["tank-0008"].getpeerinfo()
-        nine_peers = self.tanks["tank-0009"].getpeerinfo()
+        tank0_peers = self.tanks["tank-0000"].getpeerinfo()
+        tank1_peers = self.tanks["tank-0001"].getpeerinfo()
+        tank2_peers = self.tanks["tank-0002"].getpeerinfo()
+        tank3_peers = self.tanks["tank-0003"].getpeerinfo()
+        tank8_peers = self.tanks["tank-0008"].getpeerinfo()
+        tank9_peers = self.tanks["tank-0009"].getpeerinfo()
 
         for node in self.nodes:
             self.log.info(f"Node {node.index}: tank={node.tank} ip={node.rpchost}")
 
-        self.assert_connection(zero_peers, 2, ConnectionType.IP)
-        self.assert_connection(one_peers, 2, ConnectionType.IP)
-        self.assert_connection(one_peers, 3, ConnectionType.IP)
-        self.assert_connection(two_peers, 0, ConnectionType.IP)
-        self.assert_connection(two_peers, 1, ConnectionType.IP)
-        self.assert_connection(two_peers, 3, ConnectionType.IP)
-        self.assert_connection(two_peers, 4, ConnectionType.IP)
-        self.assert_connection(three_peers, 1, ConnectionType.IP)
-        self.assert_connection(three_peers, 2, ConnectionType.IP)
-        self.assert_connection(three_peers, 5, ConnectionType.IP)
-        self.assert_connection(four_peers, 2, ConnectionType.IP)
-        self.assert_connection(four_peers, 5, ConnectionType.IP)
-        self.assert_connection(five_peers, 3, ConnectionType.IP)
-        self.assert_connection(five_peers, 4, ConnectionType.IP)
-        self.assert_connection(five_peers, 6, ConnectionType.IP)
-        self.assert_connection(six_peers, 5, ConnectionType.IP)
-        self.assert_connection(six_peers, 7, ConnectionType.IP)
-        self.assert_connection(seven_peers, 6, ConnectionType.IP)
-        # Check the pre-connected nodes
-        # The only connection made by DNS name would be from the initial graph edges
-        self.assert_connection(eight_peers, 9, ConnectionType.DNS)
-        self.assert_connection(nine_peers, 8, ConnectionType.IP)
+        # Check the manual connect_nodes() connections
+        self.assert_connection(tank0_peers, 2, ConnectionType.IP)
+        self.assert_connection(tank1_peers, 2, ConnectionType.IP)
+        self.assert_connection(tank1_peers, 3, ConnectionType.IP)
 
-        # TODO: This needs to cause the test to fail
-        # assert False
+        # Ensure the other end of the connection agrees
+        self.assert_connection(tank2_peers, 0, ConnectionType.IP)
+        self.assert_connection(tank2_peers, 1, ConnectionType.IP)
+        self.assert_connection(tank3_peers, 1, ConnectionType.IP)
 
-        self.log.info(
-            f"Successfully ran the connect_dag.py scenario using a temporary file: "
-            f"{os.path.basename(__file__)} "
-        )
+        # Check the pre-connected nodes
+        # The only connection made by DNS name would be from the initial graph edges
+        self.assert_connection(tank8_peers, 9, ConnectionType.DNS)
+        self.assert_connection(tank9_peers, 8, ConnectionType.IP)
 
     def assert_connection(self, connector, connectee_index, connection_type: ConnectionType):
         if connection_type == ConnectionType.DNS:

src/warnet/deploy.py

@@ -398,6 +398,7 @@ def deploy_network(directory: Path, debug: bool = False, namespace: Optional[str
         default_file = yaml.safe_load(f)
 
     needs_ln_init = False
+    addconnections = []
     supported_ln_projects = ["lnd", "cln"]
     merged = network_file["nodes"].copy()
     merged.append(default_file)
@@ -406,8 +407,8 @@ def deploy_network(directory: Path, debug: bool = False, namespace: Optional[str
             if key in node and "channels" in node[key]:
                 needs_ln_init = True
                 break
-        if needs_ln_init:
-            break
+        if "addconnection" in node:
+            addconnections.append({node["name"]: node["addconnection"]})
 
     processes = []
     for node in network_file["nodes"]:
@@ -418,6 +419,18 @@ def deploy_network(directory: Path, debug: bool = False, namespace: Optional[str
     for p in processes:
         p.join()
 
+    if addconnections:
+        name = _run(
+            scenario_file=SCENARIOS_DIR / "addconnection_init.py",
+            debug=False,
+            source_dir=SCENARIOS_DIR,
+            additional_args=("--timeout-factor=0",),
+            admin=True,
+            namespace=namespace,
+        )
+        wait_for_pod_ready(name, namespace=namespace)
+        _logs(pod_name=name, follow=True, namespace=namespace)
+
     if needs_ln_init:
         name = _run(
             scenario_file=SCENARIOS_DIR / "ln_init.py",

src/warnet/k8s.py

@@ -327,14 +327,20 @@ def snapshot_bitcoin_datadir(
         print(f"An error occurred: {str(e)}")
 
 
+# "Readiness" depends on pod:
+# A one-shot pod like a scenario that runs to completion and exits 0
+# A long-running pod that is expected to accept network traffic
 def wait_for_pod_ready(name, namespace, timeout=300):
     sclient = get_static_client()
     w = watch.Watch()
     for event in w.stream(
         sclient.list_namespaced_pod, namespace=namespace, timeout_seconds=timeout
     ):
         pod = event["object"]
-        if pod.metadata.name == name and pod.status.phase == "Running":
+        if pod.metadata.name == name and pod.status.phase in ["Running", "Succeeded"]:
+            if pod.status.phase == "Succeeded":
+                w.stop()
+                return True
             conditions = pod.status.conditions or []
             ready_condition = next((c for c in conditions if c.type == "Ready"), None)
             if ready_condition and ready_condition.status == "True":

test/dag_connection_test.py

@@ -5,6 +5,8 @@
 
 from test_base import TestBase
 
+from warnet.process import stream_command
+
 
 class DAGConnectionTest(TestBase):
     def __init__(self):
@@ -21,14 +23,16 @@ def run_test(self):
 
     def setup_network(self):
         self.log.info("Setting up network")
-        self.log.info(self.warnet(f"deploy {self.network_dir}"))
+        stream_command(f"warnet deploy {self.network_dir}")
+        self.log.info("Waiting for pods")
         self.wait_for_all_tanks_status(target="running")
+        self.log.info("Waiting for addnode connections")
         self.wait_for_all_edges()
 
     def run_connect_dag_scenario(self):
         scenario_file = self.scen_dir / "test_scenarios" / "connect_dag.py"
         self.log.info(f"Running scenario from: {scenario_file}")
-        self.warnet(f"run {scenario_file} --source_dir={self.scen_dir}")
+        stream_command(f"warnet run {scenario_file} --source_dir={self.scen_dir}")
         self.wait_for_all_scenarios()