cli: add 'network logs' command

src/warnet/cli/cluster.py

@@ -5,6 +5,8 @@
 
 import click
 
+from .util import run_command
+
 MANIFEST_PATH = files("manifests")
 RPC_PATH = files("images").joinpath("rpc")
 
@@ -26,46 +28,6 @@ def cluster():
     pass
 
 
-def run_command(command, stream_output=False, env=None):
-    # Merge the current environment with the provided env
-    full_env = os.environ.copy()
-    if env:
-        # Convert all env values to strings (only a safeguard)
-        env = {k: str(v) for k, v in env.items()}
-        full_env.update(env)
-
-    if stream_output:
-        process = subprocess.Popen(
-            ["/bin/bash", "-c", command],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            text=True,
-            bufsize=1,
-            universal_newlines=True,
-            env=full_env,
-        )
-
-        for line in iter(process.stdout.readline, ""):
-            print(line, end="")
-
-        process.stdout.close()
-        return_code = process.wait()
-
-        if return_code != 0:
-            print(f"Command failed with return code {return_code}")
-            return False
-        return True
-    else:
-        result = subprocess.run(
-            command, shell=True, capture_output=True, text=True, executable="/bin/bash"
-        )
-        if result.returncode != 0:
-            print(f"Error: {result.stderr}")
-            return False
-        print(result.stdout)
-        return True
-
-
 @cluster.command()
 @click.option("--clean", is_flag=True, help="Remove configuration files")
 def setup_minikube(clean):

src/warnet/cli/network.py

@@ -9,6 +9,7 @@
 from rich.table import Table
 
 from .rpc import rpc_call  # noqa: I001
+from .util import run_command
 
 
 DEFAULT_GRAPH_FILE = files("graphs").joinpath("default.graphml")
@@ -158,3 +159,16 @@ def export(network: str, activity: str, exclude: str):
     print(
         rpc_call("network_export", {"network": network, "activity": activity, "exclude": exclude})
     )
+
+
+@network.command()
+@click.option("--follow", "-f", is_flag=True, help="Follow logs")
+def logs(follow: bool):
+    """Get Kubernetes logs from the RPC server"""
+    command = "kubectl logs rpc-0"
+    stream_output = False
+    if follow:
+        command += " --follow"
+        stream_output = True
+
+    run_command(command, stream_output=stream_output)

src/warnet/cli/util.py

@@ -0,0 +1,42 @@
+import os
+import subprocess
+
+
+def run_command(command, stream_output=False, env=None):
+    # Merge the current environment with the provided env
+    full_env = os.environ.copy()
+    if env:
+        # Convert all env values to strings (only a safeguard)
+        env = {k: str(v) for k, v in env.items()}
+        full_env.update(env)
+
+    if stream_output:
+        process = subprocess.Popen(
+            ["/bin/bash", "-c", command],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            bufsize=1,
+            universal_newlines=True,
+            env=full_env,
+        )
+
+        for line in iter(process.stdout.readline, ""):
+            print(line, end="")
+
+        process.stdout.close()
+        return_code = process.wait()
+
+        if return_code != 0:
+            print(f"Command failed with return code {return_code}")
+            return False
+        return True
+    else:
+        result = subprocess.run(
+            command, shell=True, capture_output=True, text=True, executable="/bin/bash"
+        )
+        if result.returncode != 0:
+            print(f"Error: {result.stderr}")
+            return False
+        print(result.stdout)
+        return True