refactor nix_csi: split monolithic service.py into focused modules

Extract functionality from service.py (620 lines) into dedicated modules:
- constants.py: configuration and path constants
- store.py: Nix store path regex and extraction utilities
- hardlinks.py: filesystem operations for hardlink trees
- builders.py: Kubernetes builder pod discovery
- cache.py: cache connectivity and copy operations
- nix.py: Nix build/eval operations
- volume.py: volume preparation and mounting

service.py now focuses on CSI orchestration (341 lines).

Removed unused kubernetes.py (get_builder_ips was never called).
Absorbed copytocache.py into cache.py.

Author: Lillecarl

python/nix_csi/builders.py

@@ -0,0 +1,42 @@
+import kr8s
+import logging
+
+from .constants import BUILDERS_ENABLED, BUILDERS_SERVICE, NAMESPACE
+
+logger = logging.getLogger("nix-csi")
+
+
+async def get_builder_uris() -> list[str]:
+    """Query k8s API for builder pods, return list of SSH URIs for --builders flag."""
+    if not BUILDERS_ENABLED:
+        return []
+
+    try:
+        # Use kr8s to query pods with label selector
+        # kr8s.asyncio.get() returns an async generator, iterate with async for
+        uris = []
+        async for pod in kr8s.asyncio.get(
+            "pods", namespace=NAMESPACE, label_selector="app.kubernetes.io/name=builder"
+        ):
+            if pod.status.phase == "Running":
+                pod_name = pod.metadata.name
+                uri = f"ssh://nix@{pod_name}.{BUILDERS_SERVICE}.{NAMESPACE}.svc.cluster.local"
+                uris.append(uri)
+
+        logger.debug(f"Discovered {len(uris)} builder pods: {uris}")
+        return uris
+    except Exception as e:
+        logger.warning(f"Failed to discover builder pods: {e}")
+        return []
+
+
+def build_builder_args(uris: list[str]) -> list[str]:
+    """Build nix command arguments for using builder pods."""
+    if not uris:
+        return []
+
+    args = ["--max-jobs", "0"]
+    for uri in uris:
+        args.extend(["--builders", uri])
+    args.append("--builders-use-substitutes")
+    return args

python/nix_csi/cache.py

@@ -0,0 +1,94 @@
+import asyncio
+import logging
+from asyncio import Semaphore, sleep
+from collections import defaultdict
+from pathlib import Path
+
+from grpclib import GRPCError
+
+from .constants import CACHE_ENABLED
+from .subprocessing import run_captured, try_console
+
+logger = logging.getLogger("nix-csi")
+
+# Locks that prevent the same derivation to be uploaded in parallel
+copy_lock: defaultdict[Path, Semaphore] = defaultdict(Semaphore)
+
+
+async def check_cache_connectivity() -> bool:
+    """Check if the cache is reachable via SSH."""
+    if not CACHE_ENABLED:
+        return False
+
+    try:
+        logger.debug("Trying to connect to cache")
+        await asyncio.wait_for(
+            try_console("ssh", "nix@nix-cache", "--", "true"),
+            timeout=2.0,
+        )
+        logger.debug("Cache connectivity check succeeded")
+        return True
+    except (GRPCError, OSError, asyncio.TimeoutError):
+        logger.warning("Cache connectivity check failed")
+        return False
+
+
+def get_substituter_args() -> list[str]:
+    """Get nix command arguments for using the cache as a substituter."""
+    return [
+        "--extra-substituters",
+        "ssh-ng://nix@nix-cache?trusted=1&priority=20",
+    ]
+
+
+async def copy_to_cache(package_path: Path) -> None:
+    """
+    Copy a package and its closure to the cache.
+
+    TODO: Rewrite this entire copy process to support user-supplied copy scripts.
+    This will allow end-users to copy to arbitrary destinations (S3, GCS, custom caches, etc.)
+    rather than hard-coding ssh-ng://nix@nix-cache.
+
+    TODO: Building should be moved to separate builder pods rather than happening
+    within the CSI daemonset. The daemonset should only handle mounting pre-built paths.
+    This will improve separation of concerns and allow dedicated builder infrastructure.
+    """
+    # Only run one copy per path per time
+    async with copy_lock[package_path]:
+        paths = [str(package_path)]
+        # Try to get derivation paths recursively. This may fail if we only have
+        # store paths without .drv files (e.g., fetched from substituters), which is normal.
+        path_info_drv = await run_captured(
+            "nix",
+            "path-info",
+            "--recursive",
+            "--derivation",
+            package_path,
+        )
+        if path_info_drv.returncode == 0:
+            paths += path_info_drv.stdout.splitlines()
+        else:
+            logger.debug(
+                f"No derivation paths found for {package_path} (normal if fetched from substituters)"
+            )
+
+        # Filter out .drv files and deduplicate (path-info runs return overlapping results)
+        paths = {p for p in paths if not p.endswith(".drv")}
+
+        if len(paths) > 0:
+            for attempt in range(6):
+                if attempt > 0:
+                    exp_backoff = min(5 * (2 ** (attempt - 1)), 60)
+                    logger.warning(
+                        f"Retry {attempt}/6 copying to cache after {exp_backoff}s: {package_path}"
+                    )
+                    await sleep(exp_backoff)
+
+                nix_copy = await run_captured(
+                    "nix", "copy", "--to", "ssh-ng://nix@nix-cache", *paths
+                )
+                if nix_copy.returncode == 0:
+                    logger.debug(f"Successfully copied to cache: {package_path}")
+                    break
+            else:
+                logger.error(f"Failed to copy to cache after 6 attempts: {package_path}")

python/nix_csi/constants.py

@@ -0,0 +1,35 @@
+import os
+from asyncio import Semaphore
+from importlib import metadata
+from pathlib import Path
+
+CSI_PLUGIN_NAME = "nix.csi.store"
+CSI_VENDOR_VERSION = metadata.version("nix-csi")
+
+# Exit code from mount command when target is already mounted
+MOUNT_ALREADY_MOUNTED = 32
+
+# Paths we base everything on.
+# Remember that these are CSI pod paths not node paths.
+NIX_ROOT = Path("/")
+CSI_ROOT = NIX_ROOT / "nix/var/nix-csi"
+CSI_VOLUMES = CSI_ROOT / "volumes"
+CSI_GCROOTS = NIX_ROOT / "nix/var/nix/gcroots/nix-csi"
+
+# Configurable via kubenix option: rsyncConcurrency (default: 1)
+# Set via RSYNC_CONCURRENCY environment variable
+RSYNC_CONCURRENCY = Semaphore(max(int(os.environ.get("RSYNC_CONCURRENCY", "1")), 1))
+
+# Configurable via kubenix option: nodeBuildTimeout (default: 300)
+# Set via NIX_BUILD_TIMEOUT environment variable
+NIX_BUILD_TIMEOUT = float(os.environ.get("NIX_BUILD_TIMEOUT", "300"))
+
+# Builder configuration
+# Set via environment variables from kubenix when builders are enabled
+BUILDERS_ENABLED = os.environ.get("BUILDERS_ENABLED", "false").lower() == "true"
+
+NAMESPACE = os.environ.get("KUBE_NAMESPACE", "nix-csi")
+BUILDERS_SERVICE = "nix-csi-builders"
+
+# Simple string check is fine - value controlled by easykubenix (always "true" or "false")
+CACHE_ENABLED = os.environ.get("CACHE_ENABLED", "false") == "true"

python/nix_csi/copytocache.py

@@ -0,0 +1,58 @@
+import os
+from pathlib import Path
+
+
+def hardlink_tree(src: Path, dst: Path) -> None:
+    """Hardlink a single directory tree, preserving symlinks."""
+    dst.mkdir(parents=True, exist_ok=True)
+
+    for entry in os.scandir(src):
+        dst_path = dst / entry.name
+
+        if entry.is_symlink():
+            os.symlink(os.readlink(entry.path), dst_path)
+        elif entry.is_dir(follow_symlinks=False):
+            hardlink_tree(Path(entry.path), dst_path)
+        elif entry.is_file(follow_symlinks=False):
+            dst_path.hardlink_to(entry.path)
+
+
+def hardlink_closure(store_paths: list[Path], dst: Path) -> None:
+    """
+    Hardlink multiple store paths into dst.
+
+    store_paths: [/nix/store/abc-foo, /nix/store/def-bar, ...]
+    dst: volume_root/nix/store
+    result: dst/abc-foo/..., dst/def-bar/...
+    """
+    dst.mkdir(parents=True, exist_ok=True)
+
+    for store_path in store_paths:
+        target = dst / store_path.name
+        if target.exists():
+            continue  # already copied (deduplication across volumes)
+        hardlink_tree(store_path, target)
+
+
+def deref_hardlink_tree(src: Path, dst: Path) -> None:
+    """
+    Recursively copy src to dst, dereferencing symlinks and
+    hardlinking files for space efficiency.
+
+    All symlink targets must exist on the same filesystem as dst.
+    """
+    src = Path(src)
+    dst = Path(dst)
+    dst.mkdir(parents=True, exist_ok=True)
+
+    for entry in os.scandir(src):
+        dst_path = dst / entry.name
+        resolved = Path(entry.path).resolve()
+
+        if not resolved.exists():
+            raise FileNotFoundError(f"Broken symlink: {entry.path} -> {resolved}")
+
+        if resolved.is_dir():
+            deref_hardlink_tree(resolved, dst_path)
+        elif resolved.is_file():
+            dst_path.hardlink_to(resolved)