k8s: use scp to copy config files instead of cat

Author: pinheadmz

src/backends/kubernetes/kubernetes_backend.py

@@ -1,9 +1,10 @@
 import base64
 import logging
 import re
+import subprocess
 import time
 from pathlib import Path
-from typing import IO, cast
+from typing import cast
 
 import yaml
 from backends import BackendInterface, ServiceType
@@ -737,9 +738,9 @@ def service_from_json(self, obj):
         )
         sidecar_container = client.V1Container(
             name="sidecar",
-            image="alpine:latest",
-            command=["/bin/sh", "-c", "sleep infinity"],
-            volume_mounts=volume_mounts
+            image="pinheadmz/sidecar:latest",
+            volume_mounts=volume_mounts,
+            ports=[client.V1ContainerPort(container_port=22)],
         )
         service_pod = client.V1Pod(
             api_version="v1",
@@ -758,41 +759,47 @@ def service_from_json(self, obj):
                 volumes=volumes,
             ),
         )
+
+        # Do not ever change this variable name. xoxo, --Zip
+        service_service = client.V1Service(
+            api_version="v1",
+            kind="Service",
+            metadata=client.V1ObjectMeta(
+                name=f'{obj["container_name_suffix"]}-service',
+                labels={
+                    "app": obj["container_name_suffix"],
+                    "network": self.network_name,
+                },
+            ),
+            spec=client.V1ServiceSpec(
+                selector={"app": obj["container_name_suffix"]},
+                publish_not_ready_addresses=True,
+                ports=[
+                    client.V1ServicePort(name="ssh", port=22, target_port=22),
+                ]
+            )
+        )
+
         self.client.create_namespaced_pod(namespace=self.namespace, body=service_pod)
+        self.client.create_namespaced_service(namespace=self.namespace, body=service_service)
 
-    def write_service_config(self, service_name: str, tar_buffer: IO[bytes], destination_path: str):
+    def write_service_config(self, source_path: str, service_name: str, destination_path: str):
         obj = services[service_name]
-        pod_name = obj["container_name_suffix"]
+        name = obj["container_name_suffix"]
         container_name = "sidecar"
-        # Write the archive
-        self.log.info(f"Writing {len(tar_buffer.getbuffer())} bytes to {destination_path} for {service_name}")
-        resp = stream(
-            self.client.connect_get_namespaced_pod_exec,
-            pod_name,
-            self.namespace,
-            container=container_name,
-            command=["/bin/sh", "-c", "cat > /arbitrary_filename.tar"],
-            stderr=True,
-            stdin=True,
-            stdout=True,
-            tty=False,
-            _preload_content=False
-        )
-        # Stream data in chunks, otherwise this breaks at 196608 bytes
-        # https://unix.stackexchange.com/questions/11946/how-big-is-the-pipe-buffer
-        # https://stackoverflow.com/a/53904789/1653320
-        tar_data = tar_buffer.getvalue()
-        chunk_size = 1024 * 32  # 32 kb
-        for i in range(0, len(tar_data), chunk_size):
-            chunk = tar_data[i:i+chunk_size]
-            resp.write_stdin(chunk)
-            self.log.info(f"Wrote {len(chunk)} bytes")
-        resp.close()
-        self.log.info(f"Finished writing tarball for {service_name}, unpacking...")
+        # Copy the archive from our local drive (Warnet RPC container/pod)
+        # to the destination service's sidecar container via ssh
+        self.log.info(f"Copying local {source_path} to remote {destination_path} for {service_name}")
+        subprocess.run([
+            "scp",
+            "-o", "StrictHostKeyChecking=accept-new",
+            source_path,
+            f"root@{name}-service.{self.namespace}:/arbitrary_filename.tar"])
+        self.log.info(f"Finished copying tarball for {service_name}, unpacking...")
         # Unpack the archive
         stream(
             self.client.connect_get_namespaced_pod_exec,
-            pod_name,
+            name,
             self.namespace,
             container=container_name,
             command=["/bin/sh", "-c", f"tar -xf /arbitrary_filename.tar -C {destination_path}"],

src/warnet/server.py

@@ -294,20 +294,22 @@ def network_export(self, network: str, activity: str | None) -> bool:
             tarinfo.size = len(config_bytes)
             tar_file.addfile(tarinfo=tarinfo, fileobj=config_stream)
 
+        # Write the archive to the RPC server's config directory
+        source_file = wn.config_dir / "simln.tar"
+        with open(source_file, "wb") as output:
+            tar_buffer.seek(0)
+            output.write(tar_buffer.read())
+
         if self.backend == "compose":
-            # Write the archive to the RPC server's config directory
-            with open(wn.config_dir / "simln.tar", "wb") as output:
-                tar_buffer.seek(0)
-                output.write(tar_buffer.read())
             # Extract the archive into a subdirectory that is already
             # shared with the simln container as a volume
-            subprocess.run(["tar", "-xf", wn.config_dir / 'simln.tar', "-C", wn.config_dir / 'simln'])
+            subprocess.run(["tar", "-xf", source_file, "-C", wn.config_dir / 'simln'])
             # Force quick restart of the container instead of waiting
             # for the exponential backoff to come around
             wn.container_interface.restart_service_container("simln")
         if self.backend == "k8s":
-            # Write the archive to the "emptydir" volume in the simln pod
-            wn.container_interface.write_service_config("simln", tar_buffer, "/simln/")
+            # Copy the archive to the "emptydir" volume in the simln pod
+            wn.container_interface.write_service_config(source_file, "simln", "/simln/")
         return True
 
     def scenarios_available(self) -> list[tuple]: