Add styling to standard out for improved readability

Tune sleep / pauses down
Set simln container restartPolicy: Never

Author: macgyver13

resources/charts/bitcoincore/charts/cln/values.yaml

@@ -101,9 +101,9 @@ tolerations: []
 affinity: {}
 
 baseConfig: |
-  log-level=debug
-  # developer
-  # dev-fast-gossip
+  log-level=info
+  developer
+  dev-fast-gossip
   bitcoin-rpcuser=user
   # bitcoind.rpcpass are set in configmap.yaml
 

resources/plugins/simln/charts/simln/values.yaml

@@ -4,6 +4,8 @@ image:
   tag: "0.2.3"
   pullPolicy: IfNotPresent
 
+restartPolicy: Never
+
 workingVolume:
   name: working-volume
   mountPath: /working

resources/plugins/simln/plugin.py

@@ -200,15 +200,17 @@ def transfer_cln_certs(name):
 def copyfile(pod_name, src_container, source_path, dst_name, dst_container, dst_path):
     namespace=get_default_namespace()
     file_data = read_file_from_container(pod_name, source_path, src_container, namespace)
-    if not write_file_to_container(
+    if write_file_to_container(
         dst_name, 
         dst_container, 
         dst_path, 
         file_data,
         namespace=namespace,
         quiet=True,
     ):
-        print(f"Failed to copy {source_path} from {pod_name} to {dst_name}:{dst_path}")
+        log.info(f"Copied {source_path} to {dst_path}")
+    else:
+        log.error(f"Failed to copy {source_path} from {pod_name} to {dst_name}:{dst_path}")
 
 
 def _sh(pod, method: str, params: tuple[str, ...]) -> str:

resources/scenarios/commander.py

@@ -82,6 +82,27 @@ def auth_proxy_request(self, method, path, postdata):
 AuthServiceProxy.oldrequest = AuthServiceProxy._request
 AuthServiceProxy._request = auth_proxy_request
 
+# Create a custom formatter
+class ColorFormatter(logging.Formatter):
+    """Custom formatter to add color based on log level."""
+    # Define ANSI color codes
+    RED = '\033[91m'
+    YELLOW = '\033[93m'
+    GREEN = '\033[92m'
+    RESET = '\033[0m'
+
+    FORMATS = {
+        logging.DEBUG: f"{RESET}%(name)-8s - Thread-%(thread)d - %(message)s{RESET}",
+        logging.INFO: f"{RESET}%(name)-8s - %(message)s{RESET}",
+        logging.WARNING: f"{YELLOW}%(name)-8s - %(message)s{RESET}",
+        logging.ERROR: f"{RED}%(name)-8s - %(message)s{RESET}",
+        logging.CRITICAL: f"{RED}##%(name)-8s - %(message)s##{RESET}"
+    }
+
+    def format(self, record):
+        log_fmt = self.FORMATS.get(record.levelno)
+        formatter = logging.Formatter(log_fmt)
+        return formatter.format(record)
 
 class Commander(BitcoinTestFramework):
     # required by subclasses of BitcoinTestFramework
@@ -155,8 +176,7 @@ def setup(self):
         # Scenarios log directly to stdout which gets picked up by the
         # subprocess manager in the server, and reprinted to the global log.
         ch = logging.StreamHandler(sys.stdout)
-        formatter = logging.Formatter(fmt="%(name)-8s %(message)s")
-        ch.setFormatter(formatter)
+        ch.setFormatter(ColorFormatter())
         self.log.addHandler(ch)
 
         # Keep a separate index of tanks by pod name

resources/scenarios/ln_framework/ln.py

@@ -103,6 +103,28 @@ def to_lnd_chanpolicy(self, capacity):
             "min_htlc_msat_specified": True,
         }
 
+# Create a custom formatter
+class ColorFormatter(logging.Formatter):
+    """Custom formatter to add color based on log level."""
+    # Define ANSI color codes
+    RED = '\033[91m'
+    YELLOW = '\033[93m'
+    GREEN = '\033[92m'
+    RESET = '\033[0m'
+
+    FORMATS = {
+        logging.DEBUG: f"{RESET}%(asctime)s - (name)-8s - Thread-%(thread)d - %(message)s{RESET}",
+        logging.INFO: f"{RESET}%(asctime)s - (name)-8s - %(message)s{RESET}",
+        logging.WARNING: f"{YELLOW}%(asctime)s - (name)-8s - %(message)s{RESET}",
+        logging.ERROR: f"{RED}%(asctime)s - (name)-8s - %(message)s{RESET}",
+        logging.CRITICAL: f"{RED}##%(asctime)s - (name)-8s - %(message)s##{RESET}"
+    }
+
+    def format(self, record):
+        log_fmt = self.FORMATS.get(record.levelno)
+        formatter = logging.Formatter(log_fmt)
+        return formatter.format(record)
+    
 class LNNode(ABC):
     @abstractmethod
     def __init__(self, pod_name):
@@ -111,8 +133,7 @@ def __init__(self, pod_name):
         # Configure logger if it has no handlers
         if not self.log.handlers:
             handler = logging.StreamHandler()
-            formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
-            handler.setFormatter(formatter)
+            handler.setFormatter(ColorFormatter())
             self.log.addHandler(handler)
             self.log.setLevel(logging.INFO)
 
@@ -165,7 +186,7 @@ def rpc(self, method: str, params: list[str] = [], namespace: Optional[str] = "d
                     continue
                 return response
             except Exception as e:
-                self.log.info(f"CLN rpc error: {e}")
+                self.log.error(f"CLN rpc error: {e}, wait and retry...")
                 sleep(2)
         return None
 
@@ -181,7 +202,7 @@ def newaddress(self, max_tries=2):
             if "bech32" in res:
                 return True, res["bech32"]
             else:
-                self.log.info(
+                self.log.warning(
                     f"Couldn't get wallet address from {self.name}:\n  {res}\n  wait and retry..."
                 )
             sleep(2)
@@ -207,23 +228,21 @@ def walletbalance(self, max_tries=2):
 
     def connect(self, target_uri, max_tries=5):
         attempt=0
-        self.log.info(f"CLN connect {self.name} to {target_uri}")
         while attempt < max_tries:
             attempt+=1
             response = self.rpc("connect", [target_uri])
             if response:
                 res = json.loads(response)
                 if "id" in res:
-                    self.log.debug(f"finished connect response: {response}")
                     return {}
                 elif "code" in res and res["code"] == 402:
-                    self.log.info(f"failed connect response: {response}")
+                    self.log.warning(f"failed connect 402: {response}, wait and retry...")
                     sleep(5)
                 else:
                     return res
             else:
-                self.log.debug(f"connect response: {response}")
-                sleep(5)
+                self.log.debug(f"connect response: {response}, wait and retry...")
+                sleep(2)
         return ""
 
     def channel(self, pk, capacity, push_amt, fee_rate, max_tries=5):
@@ -240,13 +259,13 @@ def channel(self, pk, capacity, push_amt, fee_rate, max_tries=5):
             if response:
                 res = json.loads(response)
                 if "txid" in res:
-                    self.log.debug(f"open channel succeeded: {res}")
                     return {"txid": res["txid"], "outpoint": f'{res["txid"]}:{res["outnum"]}'}
                 else:
-                    self.log.info(f"unable to open channel: {res}")
+                    self.log.warning(f"unable to open channel: {res}, wait and retry...")
+                    sleep(1)
             else:
-                self.log.debug(f"channel response: {response}")
-                sleep(5)
+                self.log.debug(f"channel response: {response}, wait and retry...")
+                sleep(2)
         return ""
 
     def graph(self, max_tries=2):
@@ -257,34 +276,25 @@ def graph(self, max_tries=2):
             if response:
                 res = json.loads(response)
                 if "channels" in res:
-                    return {"edges": res["channels"]}
+                    # Map to desired output
+                    filtered_channels = [ch for ch in res['channels'] if ch['direction'] == 1]
+                    # Sort by short_channel_id - block -> index -> output
+                    sorted_channels = sorted(filtered_channels, key=lambda x: x['short_channel_id'])
+                    # Add capacity by dividing amount_msat by 1000
+                    for channel in sorted_channels:
+                        channel['capacity'] = channel['amount_msat'] // 1000
+                    return {'edges': sorted_channels}
                 else:
-                    self.log.info(f"unable to open channel: {res}")
+                    self.log.warning(f"unable to open channel: {res}, wait and retry...")
+                    sleep(1)
             else:
-                self.log.debug(f"channel response: {response}")
-                sleep(5)
+                self.log.debug(f"channel response: {response}, wait and retry...")
+                sleep(2)
         return ""
 
     def update(self, txid_hex: str, policy: dict, capacity: int, max_tries=2):
-        self.log.info("Channel Policy Updates not supported by CLN yet!")
+        self.log.warning("Channel Policy Updates not supported by CLN yet!")
         return
-        # ln_policy = Policy.from_dict(policy).to_lnd_chanpolicy(capacity)
-        # data = {"chan_point": {"funding_txid_str": txid_hex, "output_index": 0}, **ln_policy}
-        # attempt=0
-        # while attempt < max_tries:
-        #     attempt+=1
-        #     response = self.rpc("setchannel")
-        #     if response:
-        #         res = json.loads(response)
-        #         if "channels" in res:
-        #             print(f"graph succeeded: {res}")
-        #             return {"edges": res["channels"]}
-        #         else:
-        #             print(f"unable to open channel: {res}")
-        #     else:
-        #         print(f"channel response: {response}")
-        #         sleep(5)
-        # return ""
 
 class LND(LNNode):
     def __init__(self, pod_name):
@@ -350,7 +360,7 @@ def newaddress(self, max_tries=10):
             if "address" in res:
                 return True, res["address"]
             else:
-                self.log.info(
+                self.log.warning(
                     f"Couldn't get wallet address from {self.name}:\n  {res}\n  wait and retry..."
                 )
             sleep(1)
@@ -388,9 +398,8 @@ def channel(self, pk, capacity, push_amt, fee_rate):
             if "result" in res:
                 res["txid"] = self.b64_to_hex(res["result"]["chan_pending"]["txid"], reverse=True)
                 res["outpoint"] = f'{res["txid"]}:{res["result"]["chan_pending"]["output_index"]}'
-                self.log.info(f"LND channel RESPONSE: {res}")
         except Exception as e:
-            self.log.info(f"Error opening LND channel: {e}")
+            self.log.error(f"Error opening LND channel: {e}")
         return res
 
     def update(self, txid_hex: str, policy: dict, capacity: int):

resources/scenarios/ln_init.py

@@ -6,7 +6,7 @@
 from commander import Commander
 from ln_framework.ln import Policy
 
-THREAD_JOIN_TIMEOUT=15
+THREAD_JOIN_TIMEOUT=20
 
 class LNInit(Commander):
     def set_test_params(self):
@@ -145,7 +145,7 @@ def get_ln_uri(self, ln):
         for ch in self.channels:
             node_names = self.node_names(ln_nodes)
             if not ch["source"] in node_names or not ch["target"] in node_names:
-                self.log.info(f"LN Channel {ch} not available, removing")
+                self.log.error(f"LN Channel {ch} not available, removing")
                 self.channels.remove(ch)
                 continue
             src = self.lns[ch["source"]]
@@ -184,7 +184,7 @@ def connect_ln(self, pair):
             threading.Thread(target=connect_ln, args=(self, pair)) for pair in connections
         ]
         for thread in p2p_threads:
-            sleep(5)
+            sleep(1)
             thread.start()
 
         all(thread.join(timeout=THREAD_JOIN_TIMEOUT) is None for thread in p2p_threads)
@@ -258,7 +258,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(5)
+                sleep(2)
                 t.start()
                 ch_threads.append(t)
 
@@ -275,13 +275,9 @@ def open_channel(self, ch, fee_rate):
             block_txs = block["tx"]
             block_height = block["height"]
             for ch in channels:
-                if "txid" not in ch:
-                    print(f"{ch} does not have a txid")
-                    continue
-                assert ch["id"]["block"] == block_height, f"Actual block:{block_height}\n{ch}"
-                assert (
-                    block_txs[ch["id"]["index"]] == ch["txid"]
-                ), f"Actual txid:{block_txs[ch["id"]["index"]]}\n{ch}"
+                assert ch["txid"] != "N/A", f"Channel:{ch} did not receive txid"
+                assert ch["id"]["block"] == block_height, f"Actual block:{block_height}"
+                assert ch["txid"] in block_txs, f"Block:{block_height} does not contain {ch["txid"]}"
             self.log.info("👍")
 
         gen(5)
@@ -299,10 +295,10 @@ def ln_all_chs(self, ln):
             self.log.info(f"LN {ln.name} has graph with all {expected} channels")
 
         ch_ann_threads = [
-            threading.Thread(target=ln_all_chs, args=(self, ln)) for ln in self.lns.values()
+            threading.Thread(target=ln_all_chs, args=(self, ln)) for ln in ln_nodes
         ]
         for thread in ch_ann_threads:
-            sleep(5)
+            sleep(1)
             thread.start()
 
         all(thread.join(timeout=THREAD_JOIN_TIMEOUT) is None for thread in ch_ann_threads)
@@ -359,16 +355,18 @@ def policy_equal(pol1, pol2, capacity):
             return pol1.to_lnd_chanpolicy(capacity) == pol2.to_lnd_chanpolicy(capacity)
 
         def matching_graph(self, expected, ln):
-            while True:
+            done = False
+            while not done:
                 actual = ln.graph()["edges"]
-                assert len(expected) == len(actual)
-                done = True
+                self.log.debug(f"LN {ln.name} channel graph edges: {actual}")
+                assert (len(expected) == len(actual),f"Expected edges {expected}, actual edges {actual}")
+                if len(actual) > 0: done = True
                 for i, actual_ch in enumerate(actual):
                     expected_ch = expected[i]
                     capacity = expected_ch["capacity"]
                     # We assert this because it isn't updated as part of policy.
                     # If this fails we have a bigger issue
-                    assert int(actual_ch["capacity"]) == capacity
+                    assert int(actual_ch["capacity"]) == capacity, f"LN {ln.name} graph capacity mismatch:\n actual: {actual_ch}\n expected: {expected_ch}"
 
                     # Policies were not defined in network.yaml
                     if "source_policy" not in expected_ch or "target_policy" not in expected_ch:
@@ -389,17 +387,15 @@ def matching_graph(self, expected, ln):
                     ):
                         continue
                     done = False
-                    break
                 if done:
                     self.log.info(f"LN {ln.name} graph channel policies all match expected source")
-                    break
                 else:
-                    sleep(1)
+                    sleep(5)
 
         expected = sorted(self.channels, key=lambda ch: (ch["id"]["block"], ch["id"]["index"]))
         policy_threads = [
             threading.Thread(target=matching_graph, args=(self, expected, ln))
-            for ln in self.lns.values()
+            for ln in ln_nodes
         ]
         for thread in policy_threads:
             thread.start()