Add ln_mixed node test

Update documentation to address CLN support
Make ln_framework/ln.py work in commander or local device
Add CLN support to warnet/ln.py
Add short sleep between thread starts to improve stability

Author: macgyver13

.github/workflows/test.yml

@@ -41,6 +41,7 @@ jobs:
           - graph_test.py
           - logging_test.py
           - ln_basic_test.py
+          - ln_mixed_test.py
           - ln_test.py
           - rpc_test.py
           - services_test.py

docs/plugins.md

@@ -34,7 +34,7 @@ plugins:
   postDeploy:  # Plugins will run after all the `deploy` code has run.
     simln:
       entrypoint: "../plugins/simln"
-      activity: '[{"source": "tank-0003-lnd", "destination": "tank-0005-lnd", "interval_secs": 1, "amount_msat": 2000}]'
+      activity: '[{"source": "tank-0003-lnd", "destination": "tank-0005-cln", "interval_secs": 1, "amount_msat": 2000}]'
     hello:
       entrypoint: "../plugins/hello"
       podName: "hello-post-deploy"

docs/scenarios.md

@@ -72,3 +72,9 @@ Total Tanks: 6 | Active Scenarios: 0
 ## Running a custom scenario
 
 You can write your own scenario file and run it in the same way.
+
+## Scenarios with lightning nodes
+
+When defining network.yaml all lnd nodes should be indexed in the same block before any cln nodes otherwise node responsiveness causes the expected index to get out of order with actual regardless of how the channels are opened
+Review `test/data/ln_mixed/network.yaml` for an example
+

resources/plugins/simln/README.md

@@ -13,19 +13,29 @@ SimLN also requires access details for each node; however, the SimLN plugin will
 
 ```` JSON
 {
-  "id": <node_id>,
+  "id": <node_id>-lnd,
   "address": https://<ip:port or domain:port>,
   "macaroon": <path_to_selected_macaroon>,
   "cert": <path_to_tls_cert>
 }
 ````
+SimLN also supports Core Lightning (CLN).  CLN nodes connection details are transfered from the CLN node to SimLN node during launch-activity processing.
+```` JSON
+{
+  "id": <node_id>-cln,
+  "address": https://<domain:port>,
+  "ca_cert": /working/<node_id>-cln-ca.pem,
+  "client_cert": /working/<node_id>-cln-client.pem,
+  "client_key": /working/<node_id>-cln-client-key.pem
+}
+````
 
-Since SimLN already has access to those LND connection details, it means you can focus on the "activity" definitions.
+Since SimLN already has access to those LND and CLN connection details, it means you can focus on the "activity" definitions.
 
 ### Launch activity definitions from the command line
 The SimLN plugin takes "activity" definitions like so:
 
-`./simln/plugin.py launch-activity '[{\"source\": \"tank-0003-ln\", \"destination\": \"tank-0005-ln\", \"interval_secs\": 1, \"amount_msat\": 2000}]'"''`
+`./simln/plugin.py launch-activity '[{\"source\": \"tank-0003-lnd\", \"destination\": \"tank-0005-lnd\", \"interval_secs\": 1, \"amount_msat\": 2000}]'"''`
 
 ### Launch activity definitions from within `network.yaml`
 When you initialize a new Warnet network, Warnet will create a new `network.yaml` file.  If your `network.yaml` file includes lightning nodes, then you can use SimLN to produce activity between those nodes like this:

resources/plugins/simln/plugin.py

@@ -168,7 +168,7 @@ def _generate_activity_json(activity: Optional[list[dict]]) -> str:
         ln_name = i.metadata.name
         port = 10009
         node = {"id": ln_name}
-        if "cln" in ln_name:
+        if "-cln" in ln_name:
             port = 9736
             node["ca_cert"] = f"/working/{ln_name}-ca.pem"
             node["client_cert"] = f"/working/{ln_name}-client.pem"

resources/scenarios/ln_framework/ln.py

@@ -1,6 +1,7 @@
 import base64
 import http.client
 import json
+import os
 import ssl
 from abc import ABC, abstractmethod
 from time import sleep
@@ -19,7 +20,10 @@
 
 # execute kubernetes command
 def run_command(name, command: list[str], namespace: Optional[str] = "default") -> str:
-    config.load_incluster_config()
+    if os.getenv("KUBERNETES_SERVICE_HOST") and os.getenv("KUBERNETES_SERVICE_PORT"):
+        config.load_incluster_config()
+    else:
+        config.load_kube_config()
     sclient = client.CoreV1Api()
     resp = stream(
         sclient.connect_get_namespaced_pod_exec,
@@ -209,7 +213,7 @@ def uri(self):
             return None
         return f"{res['id']}@{res['address'][0]['address']}:{res['address'][0]['port']}"
 
-    def walletbalance(self, max_tries=2):
+    def walletbalance(self, max_tries=2) -> int:
         attempt = 0
         while attempt < max_tries:
             attempt += 1
@@ -221,6 +225,18 @@ def walletbalance(self, max_tries=2):
             return int(sum(o["amount_msat"] for o in res["outputs"]) / 1000)
         return 0
 
+    def channelbalance(self, max_tries=2) -> int:
+        attempt = 0
+        while attempt < max_tries:
+            attempt += 1
+            response = self.rpc("listfunds")
+            if not response:
+                sleep(2)
+                continue
+            res = json.loads(response)
+            return int(sum(o["our_amount_msat"] for o in res["channels"]) / 1000)
+        return 0
+
     def connect(self, target_uri, max_tries=5) -> dict:
         attempt = 0
         while attempt < max_tries:
@@ -263,6 +279,23 @@ def channel(self, pk, capacity, push_amt, fee_rate, max_tries=5) -> dict:
                 sleep(2)
         return None
 
+    def createinvoice(self, sats, label, description="new invoice") -> str:
+        response = self.rpc("invoice", [sats * 1000, label, description])
+        if response:
+            res = json.loads(response)
+            return res["bolt11"]
+        return None
+
+    def payinvoice(self, payment_request) -> str:
+        response = self.rpc("pay", [payment_request])
+        if response:
+            res = json.loads(response)
+            if "code" in res:
+                return res["message"]
+            else:
+                return res["payment_hash"]
+        return None
+
     def graph(self, max_tries=2) -> dict:
         attempt = 0
         while attempt < max_tries:
@@ -304,7 +337,13 @@ def __init__(self, pod_name, logger):
         }
         self.impl = "lnd"
 
+    def reset_connection(self):
+        self.conn = http.client.HTTPSConnection(
+            host=self.name, port=8080, timeout=5, context=INSECURE_CONTEXT
+        )
+
     def get(self, uri):
+        attempt = 0
         while True:
             try:
                 self.conn.request(
@@ -313,7 +352,12 @@ def get(self, uri):
                     headers=self.headers,
                 )
                 return self.conn.getresponse().read().decode("utf8")
-            except Exception:
+            except Exception as e:
+                self.reset_connection()
+                attempt += 1
+                if attempt > 5:
+                    self.log.error(f"Error LND POST, Abort: {e}")
+                    return None
                 sleep(1)
 
     def post(self, uri, data):
@@ -323,7 +367,6 @@ def post(self, uri, data):
         post_header["Content-Type"] = "application/json"
         attempt = 0
         while True:
-            attempt += 1
             try:
                 self.conn.request(
                     method="POST",
@@ -344,7 +387,12 @@ def post(self, uri, data):
                     except Exception:
                         break
                 return stream
-            except Exception:
+            except Exception as e:
+                self.reset_connection()
+                attempt += 1
+                if attempt > 5:
+                    self.log.error(f"Error LND POST, Abort: {e}")
+                    return None
                 sleep(1)
 
     def newaddress(self, max_tries=10):
@@ -362,10 +410,14 @@ def newaddress(self, max_tries=10):
             sleep(1)
         return False, ""
 
-    def walletbalance(self):
+    def walletbalance(self) -> int:
         res = self.get("/v1/balance/blockchain")
         return int(json.loads(res)["confirmed_balance"])
 
+    def channelbalance(self) -> int:
+        res = self.get("/v1/balance/channels")
+        return int(json.loads(res)["balance"])
+
     def uri(self):
         res = self.get("/v1/getinfo")
         info = json.loads(res)
@@ -420,6 +472,28 @@ def update(self, txid_hex: str, policy: dict, capacity: int):
         )
         return json.loads(res)
 
+    def createinvoice(self, sats, label, description="new invoice") -> str:
+        b64_desc = base64.b64encode(description.encode("utf-8"))
+        response = self.post(
+            "/v1/invoices", data={"value": sats, "memo": label, "description_hash": b64_desc}
+        )
+        if response:
+            res = json.loads(response)
+            return res["payment_request"]
+        return None
+
+    def payinvoice(self, payment_request) -> str:
+        response = self.post(
+            "/v1/channels/transaction-stream", data={"payment_request": payment_request}
+        )
+        if response:
+            res = json.loads(response)
+            if "payment_error" in res:
+                return res["payment_error"]
+            else:
+                return res["payment_hash"]
+        return None
+
     def graph(self):
         res = self.get("/v1/graph")
         return json.loads(res)

resources/scenarios/ln_init.py

@@ -185,6 +185,7 @@ def connect_ln(self, pair):
             threading.Thread(target=connect_ln, args=(self, pair)) for pair in connections
         ]
         for thread in p2p_threads:
+            sleep(0.25)
             thread.start()
 
         all(thread.join(timeout=THREAD_JOIN_TIMEOUT) is None for thread in p2p_threads)
@@ -258,6 +259,7 @@ def open_channel(self, ch, fee_rate):
                 assert index == ch["id"]["index"], "Channel ID indexes are not consecutive"
                 assert fee_rate >= 1, "Too many TXs in block, out of fee range"
                 t = threading.Thread(target=open_channel, args=(self, ch, fee_rate))
+                sleep(0.25)
                 t.start()
                 ch_threads.append(t)
 
@@ -305,6 +307,7 @@ def ln_all_chs(self, ln):
 
         ch_ann_threads = [threading.Thread(target=ln_all_chs, args=(self, ln)) for ln in ln_nodes]
         for thread in ch_ann_threads:
+            sleep(0.25)
             thread.start()
 
         all(thread.join(timeout=THREAD_JOIN_TIMEOUT * 2) is None for thread in ch_ann_threads)
@@ -335,6 +338,7 @@ def update_policy(self, ln, txid_hex, policy, capacity):
                         ch["capacity"],
                     ),
                 )
+                sleep(0.25)
                 ts.start()
                 update_threads.append(ts)
             if "target_policy" in ch:
@@ -348,6 +352,7 @@ def update_policy(self, ln, txid_hex, policy, capacity):
                         ch["capacity"],
                     ),
                 )
+                sleep(0.25)
                 tt.start()
                 update_threads.append(tt)
         count = len(update_threads)
@@ -408,6 +413,7 @@ def matching_graph(self, expected, ln):
             threading.Thread(target=matching_graph, args=(self, expected, ln)) for ln in ln_nodes
         ]
         for thread in policy_threads:
+            sleep(0.25)
             thread.start()
 
         all(thread.join(timeout=THREAD_JOIN_TIMEOUT) is None for thread in policy_threads)

src/warnet/ln.py

@@ -31,7 +31,10 @@ def _rpc(pod_name: str, method: str, params: str = "", namespace: Optional[str]
     pod = get_pod(pod_name)
     namespace = get_default_namespace_or(namespace)
     chain = pod.metadata.labels["chain"]
-    cmd = f"kubectl -n {namespace} exec {pod_name} -- lncli --network {chain} {method} {' '.join(map(str, params))}"
+    ln_client = "lncli"
+    if "-cln" in pod_name:
+        ln_client = "lightning-cli"
+    cmd = f"kubectl -n {namespace} exec {pod_name} -- {ln_client} --network {chain} {method} {' '.join(map(str, params))}"
     return run_command(cmd)
 
 
@@ -48,7 +51,10 @@ def pubkey(
 
 def _pubkey(pod: str):
     info = _rpc(pod, "getinfo")
-    return json.loads(info)["identity_pubkey"]
+    pubkey_key = "identity_pubkey"
+    if "-cln" in pod:
+        pubkey_key = "id"
+    return json.loads(info)[pubkey_key]
 
 
 @ln.command()
@@ -64,8 +70,11 @@ def host(
 
 def _host(pod):
     info = _rpc(pod, "getinfo")
-    uris = json.loads(info)["uris"]
-    if uris and len(uris) >= 0:
-        return uris[0].split("@")[1]
+    if "-cln" in pod:
+        return json.loads(info)["alias"]
     else:
-        return ""
+        uris = json.loads(info)["uris"]
+        if uris and len(uris) >= 0:
+            return uris[0].split("@")[1]
+        else:
+            return ""