diff --git a/docs/tank-options.md b/docs/tank-options.md index 110e56db7..4ae664356 100644 --- a/docs/tank-options.md +++ b/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 `.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. diff --git a/resources/charts/bitcoincore/templates/pod.yaml b/resources/charts/bitcoincore/templates/pod.yaml index 2577468f9..559aae746 100644 --- a/resources/charts/bitcoincore/templates/pod.yaml +++ b/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 }} diff --git a/resources/charts/bitcoincore/values.yaml b/resources/charts/bitcoincore/values.yaml index 5b8f486b9..0e3e5f960 100644 --- a/resources/charts/bitcoincore/values.yaml +++ b/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 diff --git a/resources/scenarios/addconnection_init.py b/resources/scenarios/addconnection_init.py new file mode 100644 index 000000000..36096266a --- /dev/null +++ b/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() diff --git a/resources/scenarios/commander.py b/resources/scenarios/commander.py index edc9b132c..d7efa137f 100644 --- a/resources/scenarios/commander.py +++ b/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) diff --git a/resources/scenarios/test_scenarios/connect_dag.py b/resources/scenarios/test_scenarios/connect_dag.py index c45caef6f..ab1a01c81 100644 --- a/resources/scenarios/test_scenarios/connect_dag.py +++ b/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: diff --git a/src/warnet/deploy.py b/src/warnet/deploy.py index 9b9f3329f..1ba54d420 100644 --- a/src/warnet/deploy.py +++ b/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", diff --git a/src/warnet/k8s.py b/src/warnet/k8s.py index 6516c2c4a..ac879ea8b 100644 --- a/src/warnet/k8s.py +++ b/src/warnet/k8s.py @@ -327,6 +327,9 @@ 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() @@ -334,7 +337,10 @@ def wait_for_pod_ready(name, namespace, timeout=300): 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": diff --git a/test/dag_connection_test.py b/test/dag_connection_test.py index dee38356a..7d3104341 100755 --- a/test/dag_connection_test.py +++ b/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() diff --git a/test/data/ten_semi_unconnected/network.yaml b/test/data/ten_semi_unconnected/network.yaml index f058ac70f..a6c576c79 100644 --- a/test/data/ten_semi_unconnected/network.yaml +++ b/test/data/ten_semi_unconnected/network.yaml @@ -1,31 +1,23 @@ nodes: - name: tank-0000 - config: | - debug=rpc - debug=validation - name: tank-0001 - config: | - debug=net - debug=validation - name: tank-0002 - config: | - debug=validation - name: tank-0003 - config: | - debug=validation + addconnection: + - to: tank-0004 + type: block-relay-only - name: tank-0004 + addconnection: + - to: tank-0006 + type: addr-fetch - name: tank-0005 - config: | - debug=validation + addconnection: + - to: tank-0006 + - to: tank-0007 + v2: false - name: tank-0006 - name: tank-0007 - config: | - debug=validation - name: tank-0008 addnode: - tank-0009 - config: | - debug=validation - name: tank-0009 - config: | - debug=validation diff --git a/test/data/ten_semi_unconnected/node-defaults.yaml b/test/data/ten_semi_unconnected/node-defaults.yaml index 7e021cad1..e6fec61f4 100644 --- a/test/data/ten_semi_unconnected/node-defaults.yaml +++ b/test/data/ten_semi_unconnected/node-defaults.yaml @@ -1,4 +1,3 @@ image: repository: bitcoindevproject/bitcoin pullPolicy: IfNotPresent - tag: "27.0"