"""Sprint 13.29: per-LXC mount points enumeration. The Mount Points tab in the LXC modal calls ``GET /api/lxc//mount-points`` which delegates here. We parse the container config (``/etc/pve/lxc/.conf``) for ``mpX:`` entries — the rootfs is intentionally excluded (the user asked for *user-added* mounts, not the container's own disk). Each ``mpX:`` is classified into one of three types based on the source syntax: * ``pve_volume`` — ``storage_id:vol-id`` (block device assigned from a PVE storage; appears as a separate volume, not a path) * ``pve_storage_bind`` — absolute path under ``/mnt/pve/`` that resolves to a registered PVE storage (typical NFS/CIFS share bound into the container) * ``host_bind`` — any other absolute path on the host For each entry we resolve the source-side capacity (so the value is available even when the LXC is stopped) and, when the LXC is running, enrich with runtime fields read from ``/proc//mounts``: the filesystem actually mounted on the target, mount options, and a stale-detection stat with timeout. Ad-hoc mounts done inside the container (NFS/CIFS mounted from inside the CT, not via ``mpX:``) are listed alongside the configured ones with a ``ad_hoc`` type so the user sees the complete picture. """ from __future__ import annotations import os import re import shlex import subprocess from pathlib import Path from typing import Any, Optional _LXC_CONF_DIR = Path("/etc/pve/lxc") _PCT = "/usr/sbin/pct" _PVESH = "/usr/sbin/pvesh" _PVESM = "/usr/sbin/pvesm" _MP_LINE_RE = re.compile(r"^(?Pmp\d+):\s*(?P.+)$") _REMOTE_FS_RE = re.compile(r"^(nfs|cifs|smb)", re.IGNORECASE) # Hard timeouts so a stuck `pct exec` or `pvesm status` never freezes # the request. Same defaults as mount_monitor. _EXEC_TIMEOUT = int(os.environ.get("PROXMENUX_LXC_EXEC_TIMEOUT", "3")) _STAT_TIMEOUT = int(os.environ.get("PROXMENUX_MOUNT_STAT_TIMEOUT", "2")) # --------------------------------------------------------------------------- # Config parsing # --------------------------------------------------------------------------- def _parse_mp_line(rest: str) -> dict[str, Any]: """Parse the value side of an ``mpX:`` line. Format: ``,mp=[,opt1=val1,opt2,...]`` The first comma-separated token is the source — either an absolute path (host bind) or ``storage_id:vol-id`` (PVE volume). Subsequent tokens are key=value pairs; ``mp=`` carries the target path inside the CT, the rest are mount options (acl, backup, ro, replicate, quota, shared, size, etc). """ parts = rest.strip().split(",") if not parts: return {} source = parts[0].strip() out: dict[str, Any] = {"source": source} options: list[str] = [] for token in parts[1:]: token = token.strip() if not token: continue if "=" in token: k, v = token.split("=", 1) k = k.strip() v = v.strip() if k == "mp": out["target"] = v else: # Numeric-looking values pass through as strings. Frontend # treats them as opaque badges. out.setdefault("config_options", {})[k] = v else: options.append(token) if options: out.setdefault("config_flags", []).extend(options) return out def _read_lxc_config(vmid: str) -> list[dict[str, Any]]: """Return the parsed mpX entries from /etc/pve/lxc/.conf. Skips comment lines and the rootfs entry (per Sprint 13.29 scope). Stops at the first snapshot section header (``[snapshot_name]``) because mp lines below that point are config history, not active. """ conf = _LXC_CONF_DIR / f"{vmid}.conf" out: list[dict[str, Any]] = [] try: text = conf.read_text(encoding="utf-8", errors="replace") except OSError: return out for raw in text.splitlines(): line = raw.strip() if line.startswith("["): # Snapshot section — stop reading active config. break if not line or line.startswith("#"): continue m = _MP_LINE_RE.match(line) if not m: continue parsed = _parse_mp_line(m.group("rest")) parsed["mp_index"] = m.group("key") # mp0, mp1, ... out.append(parsed) return out def _mount_target_key(target: str) -> str: """Return a stable comparison key for a CT-side mount target. Proxmox config accepts a trailing slash in ``mp=/path/`` while ``/proc//mounts`` reports the realised target as ``/path``. They name the same mount point, so comparisons must not treat the spelling difference as a runtime divergence. Keep the root path intact: stripping its only slash would turn it into an empty key. """ if target == "/": return target return target.rstrip("/") # --------------------------------------------------------------------------- # Type classification + source resolution # --------------------------------------------------------------------------- def _list_pve_storages() -> dict[str, dict[str, Any]]: """Map storage_id → ``{type, content, total_kib, used_kib, avail_kib}`` from ``pvesm status``. One subprocess call covers every classifier decision below.""" out: dict[str, dict[str, Any]] = {} try: proc = subprocess.run( [_PVESM, "status"], capture_output=True, text=True, timeout=_EXEC_TIMEOUT, ) if proc.returncode != 0: return out # Header: Name Type Status Total(KiB) Used Available % for line in proc.stdout.strip().splitlines()[1:]: parts = line.split() if len(parts) < 6: continue try: out[parts[0]] = { "type": parts[1], "status": parts[2], "total_kib": int(parts[3]), "used_kib": int(parts[4]), "avail_kib": int(parts[5]), } except ValueError: continue except (subprocess.TimeoutExpired, OSError): pass return out def _classify(source: str, pve_storages: dict[str, dict[str, Any]]) -> dict[str, Any]: """Decide whether ``source`` is a PVE volume, a PVE-storage bind, or a plain host-directory bind. Returns the classification dict that ends up on the response.""" # `:` syntax → PVE volume (block device). if ":" in source and not source.startswith("/"): sid = source.split(":", 1)[0] st = pve_storages.get(sid, {}) return { "type": "pve_volume", "origin_storage": sid, "origin_storage_type": st.get("type", ""), "origin_label": source, } if source.startswith("/mnt/pve/"): rest = source[len("/mnt/pve/"):] sid = rest.split("/", 1)[0] if "/" in rest else rest if sid in pve_storages: st = pve_storages[sid] return { "type": "pve_storage_bind", "origin_storage": sid, "origin_storage_type": st.get("type", ""), "origin_label": source, } # Anything else absolute is a plain host bind. Origin label is the # path itself; capacity comes from `df` of that path. return { "type": "host_bind", "origin_storage": "", "origin_storage_type": "", "origin_label": source, } # --------------------------------------------------------------------------- # Capacity lookup # --------------------------------------------------------------------------- def _df_path(path: str) -> dict[str, Optional[int]]: """``df`` against a host path with timeout. Same pattern as mount_monitor — used here for ``host_bind`` origins.""" empty = {"total_bytes": None, "used_bytes": None, "available_bytes": None} try: proc = subprocess.run( ["df", "-B1", "--output=size,used,avail", path], capture_output=True, text=True, timeout=_STAT_TIMEOUT, ) if proc.returncode != 0: return empty lines = [ln for ln in proc.stdout.strip().splitlines() if ln.strip()] if len(lines) < 2: return empty parts = lines[-1].split() if len(parts) < 3: return empty try: return { "total_bytes": int(parts[0]), "used_bytes": int(parts[1]), "available_bytes": int(parts[2]), } except ValueError: return empty except (subprocess.TimeoutExpired, OSError): return empty _SIZE_UNIT_TO_BYTES = { "": 1, "B": 1, "K": 1024, "KB": 1024, "KIB": 1024, "M": 1024 ** 2, "MB": 1024 ** 2, "MIB": 1024 ** 2, "G": 1024 ** 3, "GB": 1024 ** 3, "GIB": 1024 ** 3, "T": 1024 ** 4, "TB": 1024 ** 4, "TIB": 1024 ** 4, } def _parse_pve_size(value: str) -> Optional[int]: """Convert PVE-style sizes (``150G``, ``32M``, ``2T``) to bytes. PVE stores volume sizes in lxc.conf as ``size=`` where unit is a single letter from {K,M,G,T} (powers of 1024). Returns None for empty/unparseable input — callers fall through to pvesm-based totals. """ if value is None: return None s = str(value).strip().upper() if not s: return None m = re.match(r"^(\d+(?:\.\d+)?)\s*([KMGT]?I?B?)$", s) if not m: return None try: magnitude = float(m.group(1)) except ValueError: return None unit = m.group(2) or "" multiplier = _SIZE_UNIT_TO_BYTES.get(unit) if multiplier is None: return None return int(magnitude * multiplier) def _df_via_host_pid(host_pid: str, ct_target: str) -> dict[str, Optional[int]]: """``df`` the CT-internal path via ``/proc//root`` so we get the filesystem as the container sees it, including ZFS dataset quotas. Used for ``pve_volume`` mounts whose ``pvesm status`` numbers reflect the whole storage pool instead of the per-subvol quota — without this the UI showed 851 GB total for a 150 GB ZFS subvol because pvesm reports the rpool's free space. Note: this path does NOT measure NFS/CIFS mounts that were set up from INSIDE the CT (`mount -t nfs` / `/etc/fstab` inside the container). Those live in the CT's own mount namespace and aren't visible to the host's `df` even through `/proc//root`. Use `_df_via_pct_exec` for ad-hoc mounts. """ empty = {"total_bytes": None, "used_bytes": None, "available_bytes": None} if not host_pid or not ct_target: return empty full = f"/proc/{host_pid}/root{ct_target}" try: proc = subprocess.run( ["df", "-B1", "--output=size,used,avail", full], capture_output=True, text=True, timeout=_STAT_TIMEOUT, ) except subprocess.TimeoutExpired: # A filesystem that does not answer df will not answer statfs # either; asking again would only double the wait. return empty except OSError: return empty if proc.returncode == 0: lines = [ln for ln in proc.stdout.strip().splitlines() if ln.strip()] parts = lines[-1].split() if len(lines) >= 2 else [] if len(parts) >= 3: try: return { "total_bytes": int(parts[0]), "used_bytes": int(parts[1]), "available_bytes": int(parts[2]), } except ValueError: pass return _statfs_via_host(full) def _statfs_via_host(full: str) -> dict[str, Optional[int]]: """Capacity of a path read from its filesystem rather than the mount table. df names a path by finding the mount that holds it in the host's own table, and a volume of a native OCI container is not in that table: df answers "no file systems processed" and the tab showed no usage for any of its volumes. statfs asks the filesystem of the path directly and returns the same three figures df prints. It is only reached when df completes without an answer, so every mount df can measure keeps the value it had. """ empty = {"total_bytes": None, "used_bytes": None, "available_bytes": None} try: proc = subprocess.run( ["stat", "-f", "-c", "%S %b %f %a", full], capture_output=True, text=True, timeout=_STAT_TIMEOUT, ) if proc.returncode != 0: return empty block, total, free, available = (int(value) for value in proc.stdout.split()) except (subprocess.TimeoutExpired, OSError, ValueError): return empty if block <= 0 or total <= 0: return empty return { "total_bytes": total * block, "used_bytes": (total - free) * block, "available_bytes": available * block, } def _df_via_pct_exec(vmid: str, ct_target: str, timeout: int = 6) -> dict[str, Optional[int]]: """``df`` a path from INSIDE the CT via ``pct exec``. Needed for ad-hoc NFS/CIFS mounts that live in the CT's own mount namespace and aren't visible from the host (so `_df_via_host_pid` returns empty for them). Heavier than the host-side df (full `pct exec` round-trip ~1-3s), so we only use it for ad-hoc mounts. The 6s timeout is generous enough for NFS over slow links but won't drag the request past the proxy timeout. """ empty = {"total_bytes": None, "used_bytes": None, "available_bytes": None} if not vmid or not ct_target: return empty try: proc = subprocess.run( [_PCT, "exec", vmid, "--", "df", "-B1", "--output=size,used,avail", ct_target], capture_output=True, text=True, timeout=timeout, ) if proc.returncode != 0: return empty lines = [ln for ln in proc.stdout.strip().splitlines() if ln.strip()] if len(lines) < 2: return empty parts = lines[-1].split() if len(parts) < 3: return empty return { "total_bytes": int(parts[0]), "used_bytes": int(parts[1]), "available_bytes": int(parts[2]), } except (subprocess.TimeoutExpired, OSError, ValueError): return empty def _capacity_for(source: str, classification: dict[str, Any], pve_storages: dict[str, dict[str, Any]], config_options: Optional[dict[str, Any]] = None, host_pid: str = "", target: str = "") -> dict[str, Optional[int]]: """Return total/used/available bytes for the *source* of a mount. ``pve_volume`` quota handling (Sprint 14.x — Ignacio Seijo 10/05): A ``mp6: local-zfs:subvol-310-disk-1,size=150G,...`` line carved out a 150 GB subvol from a 1 TB pool. The previous code read ``pvesm status local-zfs`` and reported 851 GB total / 19% used — reflecting the whole pool, not the subvol. We now prefer, in order: 1) ``df`` of ``/proc//root/`` when the CT is up — gives the correct view-from-inside numbers including the quota. 2) ``size=`` from lxc.conf as the total; usage is unknown when the CT isn't running, so the UI shows total only. 3) Fallback to ``pvesm status`` (pool numbers) when the entry has no declared size — that's the legacy behaviour for sizeless block volumes (lvm raw, rbd). ``pve_storage_bind`` mounts (NFS, CIFS at ``/mnt/pve/...``) keep the pvesm-based numbers because the storage IS the source of truth for those. ``host_bind`` falls back to ``df`` of the host path. None values mean the lookup didn't succeed and the UI will render n/a. """ ctype = classification.get("type") config_options = config_options or {} declared_size_bytes = _parse_pve_size(config_options.get("size")) if ctype == "pve_volume": # 1) Live numbers from inside the CT (respects quota). if host_pid and target: live = _df_via_host_pid(host_pid, target) if live.get("total_bytes") is not None: return live # 2) CT down (or df failed): expose declared quota as total. if declared_size_bytes is not None: return { "total_bytes": declared_size_bytes, "used_bytes": None, "available_bytes": None, } # 3) No quota declared: legacy pool-level numbers. sid = classification.get("origin_storage", "") st = pve_storages.get(sid) if not st: return {"total_bytes": None, "used_bytes": None, "available_bytes": None} return { "total_bytes": st["total_kib"] * 1024 if st.get("total_kib") is not None else None, "used_bytes": st["used_kib"] * 1024 if st.get("used_kib") is not None else None, "available_bytes": st["avail_kib"] * 1024 if st.get("avail_kib") is not None else None, } if ctype == "pve_storage_bind": sid = classification.get("origin_storage", "") st = pve_storages.get(sid) if not st: return {"total_bytes": None, "used_bytes": None, "available_bytes": None} # pvesm reports KiB; multiply by 1024 to keep the contract with # the host-side mount monitor (which returns bytes from `df`). return { "total_bytes": st["total_kib"] * 1024 if st.get("total_kib") is not None else None, "used_bytes": st["used_kib"] * 1024 if st.get("used_kib") is not None else None, "available_bytes": st["avail_kib"] * 1024 if st.get("avail_kib") is not None else None, } if ctype == "host_bind": return _df_path(source) return {"total_bytes": None, "used_bytes": None, "available_bytes": None} # --------------------------------------------------------------------------- # Runtime state (LXC running) # --------------------------------------------------------------------------- def _ct_status(vmid: str) -> tuple[bool, str]: """Return (running, init_pid). pid is empty string when stopped.""" try: proc = subprocess.run( [_PCT, "status", vmid, "--verbose"], capture_output=True, text=True, timeout=_EXEC_TIMEOUT, ) if proc.returncode != 0: return False, "" running = False pid = "" for line in proc.stdout.splitlines(): low = line.strip().lower() if low.startswith("status:"): running = "running" in low elif low.startswith("pid:"): pid = line.split(":", 1)[1].strip() return running, pid except (subprocess.TimeoutExpired, OSError): return False, "" def _read_ct_proc_mounts(host_pid: str) -> list[dict[str, Any]]: """Read /proc//mounts from the host side — works because the kernel exposes every namespace's mount table under that path. We don't need a second pct exec. """ out: list[dict[str, Any]] = [] if not host_pid: return out try: with open(f"/proc/{host_pid}/mounts", "r", encoding="utf-8", errors="replace") as f: for line in f: parts = line.strip().split() if len(parts) < 4: continue source, target, fstype, options = parts[0], parts[1], parts[2], parts[3] out.append({ "rt_source": source, "rt_target": target, "rt_fstype": fstype, "rt_options": options, "rt_readonly": "ro" in set(options.split(",")), }) except OSError: pass return out def _host_source_state(source: str) -> dict[str, Any]: """Inspect a host-side bind source to detect 'zombie' binds. Reported by Ignacio Seijo (11/05): when the host unmounted ``/mnt/nas1_con_backup`` the CT kept reporting it as ``mounted`` because the bind into the CT's mount namespace was still live — the kernel doesn't propagate the host-side umount to the child namespace. The CT's view becomes a frozen snapshot of whatever was under the path at bind time (usually an empty dir). Returns ``{exists, is_mountpoint, error}``. ``exists=False`` means the source path is gone entirely (e.g. a USB drive that was physically removed). ``is_mountpoint=False`` while ``exists=True`` is the zombie-bind case the UI flags. Only meaningful for absolute host paths. Storage-id sources (``local-zfs:subvol-...``) return ``{None, None, None}`` since there is no host path to inspect. """ empty = {"exists": None, "is_mountpoint": None, "error": None} if not source or not source.startswith("/"): return empty try: st_exists = os.path.exists(source) except OSError as e: return {"exists": None, "is_mountpoint": None, "error": str(e)} if not st_exists: return {"exists": False, "is_mountpoint": False, "error": "path missing"} try: proc = subprocess.run( ["mountpoint", "-q", source], capture_output=True, text=True, timeout=_STAT_TIMEOUT, ) is_mp = (proc.returncode == 0) return {"exists": True, "is_mountpoint": is_mp, "error": None} except (subprocess.TimeoutExpired, OSError) as e: return {"exists": True, "is_mountpoint": None, "error": str(e)} def _stat_via_host(host_pid: str, ct_target: str, timeout: int = _STAT_TIMEOUT) -> dict[str, Any]: """Stat the container-internal target through /proc//root — detects stale NFS without another pct exec round-trip.""" if not host_pid: return {"reachable": False, "error": "CT pid unknown"} full = f"/proc/{host_pid}/root{ct_target}" try: result = subprocess.run( ["stat", "-c", "%i", full], capture_output=True, text=True, timeout=timeout, ) if result.returncode == 0: return {"reachable": True, "error": None} err = (result.stderr or result.stdout).strip() or "stat returned non-zero" return {"reachable": False, "error": err} except subprocess.TimeoutExpired: return {"reachable": False, "error": f"stat timed out after {timeout}s"} except OSError as e: return {"reachable": False, "error": str(e)} # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def get_lxc_mount_points_static(vmid: str) -> dict[str, Any]: """Static half of the mount-points payload — safe to cache indefinitely because it only reads config and classifies against PVE's storage inventory. Returns: - ``ok`` (bool) - ``vmid`` (str) - ``mount_points`` — list of configured mp0/mp1/... entries with source / target / type / origin classification / host source existence flags. No `df`, no `stat`, no ad-hoc detection. The runtime enrichment (capacity, health, ad-hoc mounts, runtime_mounted flag) lives in `get_lxc_mount_points_runtime` and is fetched fresh on every modal open by the client. That split lets the backend cache this half indefinitely (with event invalidation on start/stop) while still giving the user real- time capacity when they actually look.""" if not re.match(r"^\d+$", vmid): return {"ok": False, "error": "invalid vmid"} config_entries = _read_lxc_config(vmid) pve_storages = _list_pve_storages() out: list[dict[str, Any]] = [] for entry in config_entries: source = entry.get("source", "") target = entry.get("target", "") cls = _classify(source, pve_storages) host_src = _host_source_state(source) out.append({ "mp_index": entry.get("mp_index", ""), "source": source, "target": target, "type": cls["type"], "origin_storage": cls.get("origin_storage", ""), "origin_storage_type": cls.get("origin_storage_type", ""), "origin_label": cls.get("origin_label", source), "config_options": entry.get("config_options", {}), "config_flags": entry.get("config_flags", []), "host_source_exists": host_src["exists"], "host_source_is_mountpoint": host_src["is_mountpoint"], }) # Cheap hint so the client can render the Mount Points tab # immediately for CTs that ONLY have ad-hoc NFS/CIFS mounts done # from inside the container (nothing in .conf, so `out` is # empty). Without this hint the tab appears only after the # runtime endpoint returns 200-500 ms later, pushing the other # tabs sideways. Reading /proc//mounts is a pure file read # (~1 ms, no subprocess), filter by remote fs family so only # storage counts — plain bind mounts of /dev/* passthrough # devices don't inflate the count. # # IMPORTANT: exclude runtime targets that match a declared mp. # When a host mp source is itself a remote share (e.g. mp0 binds # /mnt/pve/Piblic which is a CIFS mount on the host), the same # mount surfaces in /proc//mounts with an `nfs`/`cifs` # fstype from the CT's perspective. Without the filter the hint # double-counted it, so the badge showed mp+1 when the tab really # only had `mp` cards to render. ad_hoc_hint_count = 0 running, host_pid = _ct_status(vmid) if running and host_pid: try: config_targets = { _mount_target_key(entry.get("target", "")) for entry in config_entries if entry.get("target") } for rt in _read_ct_proc_mounts(host_pid): if not _REMOTE_FS_RE.match(rt.get("rt_fstype", "")): continue if _mount_target_key(rt.get("rt_target", "")) in config_targets: continue ad_hoc_hint_count += 1 except Exception: pass return { "ok": True, "vmid": vmid, "mount_points": out, "ad_hoc_hint_count": ad_hoc_hint_count, } def get_lxc_mount_points_runtime(vmid: str) -> dict[str, Any]: """Runtime half — always fresh, no cache. Fetched by the client every time the Mount Points tab is opened so the operator sees live capacity + reachability, plus any ad-hoc NFS/CIFS mounts the container itself has made since the last static snapshot. Returns: - ``ok`` (bool) - ``vmid`` (str) - ``running`` (bool) - ``runtime`` — dict keyed by target, containing runtime state + capacity per configured mount point - ``ad_hoc`` — list of NFS/CIFS/SMB mounts done inside the CT that aren't backed by an mp config line The client merges `runtime[target]` onto the matching card from the static payload; ad-hoc mounts render as their own cards under a "Mounted inside container" divider. If the CT is down or the client had no static payload for a target, the tab still renders whatever runtime info is available (never blanks).""" if not re.match(r"^\d+$", vmid): return {"ok": False, "error": "invalid vmid"} config_entries = _read_lxc_config(vmid) pve_storages = _list_pve_storages() running, host_pid = _ct_status(vmid) rt_mounts = _read_ct_proc_mounts(host_pid) if running else [] # Same parallelisation as the pre-split path: `df`/`stat` per # mount point are I/O-bound. Serialised, a CT with 5+ binds # tripped Caddy's 3s reverse-proxy timeout. from concurrent.futures import ThreadPoolExecutor rt_by_target: dict[str, dict[str, Any]] = { _mount_target_key(m["rt_target"]): m for m in rt_mounts } runtime_by_target: dict[str, dict[str, Any]] = {} matched_targets: set[str] = set() def _gather_one(entry): src = entry.get("source", "") tgt = entry.get("target", "") classification = _classify(src, pve_storages) capacity = _capacity_for( src, classification, pve_storages, config_options=entry.get("config_options", {}), host_pid=host_pid if running else "", target=tgt, ) target_key = _mount_target_key(tgt) live_target = bool(running and tgt and target_key in rt_by_target) health = _stat_via_host(host_pid, tgt) if live_target else None return entry, capacity, target_key, live_target, health if config_entries: max_workers = max(2, min(8, len(config_entries))) with ThreadPoolExecutor(max_workers=max_workers) as pool: gathered = list(pool.map(_gather_one, config_entries)) else: gathered = [] for entry, cap, target_key, live_target, health in gathered: target = entry.get("target", "") rt_item: dict[str, Any] = {**cap} if live_target: rt = rt_by_target[target_key] rt_item.update({ "runtime_mounted": True, "runtime_source": rt["rt_source"], "runtime_fstype": rt["rt_fstype"], "runtime_options": rt["rt_options"], "runtime_readonly": rt["rt_readonly"], "runtime_reachable": health["reachable"], "runtime_error": health["error"], }) matched_targets.add(target_key) elif running: rt_item["runtime_mounted"] = False rt_item["runtime_error"] = "configured but not mounted" else: rt_item["runtime_mounted"] = None # CT down runtime_by_target[target] = rt_item # Ad-hoc remote mounts inside the running CT — same logic and # parallelisation as before. ad_hoc: list[dict[str, Any]] = [] if running: ad_hoc_candidates = [ rt for rt in rt_mounts if _mount_target_key(rt["rt_target"]) not in matched_targets and _REMOTE_FS_RE.match(rt["rt_fstype"]) ] if ad_hoc_candidates: max_workers = max(2, min(8, len(ad_hoc_candidates))) with ThreadPoolExecutor(max_workers=max_workers) as pool: def _gather_adhoc(rt): h = _stat_via_host(host_pid, rt["rt_target"]) if h.get("reachable"): cap = _df_via_pct_exec(vmid, rt["rt_target"]) else: cap = {"total_bytes": None, "used_bytes": None, "available_bytes": None} return rt, h, cap results = list(pool.map(_gather_adhoc, ad_hoc_candidates)) for rt, health, cap in results: ad_hoc.append({ "mp_index": "", "source": rt["rt_source"], "target": rt["rt_target"], "type": "ad_hoc", "origin_storage": "", "origin_storage_type": "", "origin_label": rt["rt_source"], "config_options": {}, "config_flags": [], "total_bytes": cap["total_bytes"], "used_bytes": cap["used_bytes"], "available_bytes": cap["available_bytes"], "runtime_mounted": True, "runtime_source": rt["rt_source"], "runtime_fstype": rt["rt_fstype"], "runtime_options": rt["rt_options"], "runtime_readonly": rt["rt_readonly"], "runtime_reachable": health["reachable"], "runtime_error": health["error"], }) return { "ok": True, "vmid": vmid, "running": running, "runtime": runtime_by_target, "ad_hoc": ad_hoc, } def get_lxc_mount_points(vmid: str) -> dict[str, Any]: """Legacy combined entry point — kept for backwards compatibility with any caller that still wants the pre-split shape. New code should hit the static/runtime pair separately. Merges the two halves so the returned dict matches what the single-endpoint route used to return before the split.""" if not re.match(r"^\d+$", vmid): return {"ok": False, "error": "invalid vmid"} config_entries = _read_lxc_config(vmid) pve_storages = _list_pve_storages() running, host_pid = _ct_status(vmid) rt_mounts = _read_ct_proc_mounts(host_pid) if running else [] # Index runtime mounts by their CT-side target path so we can # match a config entry to its current realised state in O(1). rt_by_target: dict[str, dict[str, Any]] = { _mount_target_key(m["rt_target"]): m for m in rt_mounts } out: list[dict[str, Any]] = [] matched_targets: set[str] = set() # Pre-compute per-entry subprocess work in parallel so a CT with # many mountpoints doesn't pay N×(_STAT_TIMEOUT + _STAT_TIMEOUT) # serialised cost. The previous serial path tripped Caddy's 3s # reverse-proxy timeout (Ignacio Seijo 11/05: "/api/lxc/210/ # mount-points → 502 (3.00s)") on hosts with 5+ binds. ThreadPool # is the right primitive — these are all I/O-bound `df`/`stat` # calls hitting independent paths. from concurrent.futures import ThreadPoolExecutor def _gather_one(entry): src = entry.get("source", "") tgt = entry.get("target", "") classification = _classify(src, pve_storages) capacity = _capacity_for( src, classification, pve_storages, config_options=entry.get("config_options", {}), host_pid=host_pid if running else "", target=tgt, ) host_src = _host_source_state(src) target_key = _mount_target_key(tgt) live_target = bool(running and tgt and target_key in rt_by_target) health = _stat_via_host(host_pid, tgt) if live_target else None return entry, classification, capacity, host_src, target_key, live_target, health max_workers = max(2, min(8, len(config_entries) or 1)) with ThreadPoolExecutor(max_workers=max_workers) as pool: gathered = list(pool.map(_gather_one, config_entries)) for entry, cls, cap, host_src, target_key, live_target, health in gathered: source = entry.get("source", "") target = entry.get("target", "") item: dict[str, Any] = { "mp_index": entry.get("mp_index", ""), "source": source, "target": target, "type": cls["type"], "origin_storage": cls.get("origin_storage", ""), "origin_storage_type": cls.get("origin_storage_type", ""), "origin_label": cls.get("origin_label", source), "config_options": entry.get("config_options", {}), "config_flags": entry.get("config_flags", []), "host_source_exists": host_src["exists"], "host_source_is_mountpoint": host_src["is_mountpoint"], **cap, } # Runtime enrichment when CT is up. if live_target: rt = rt_by_target[target_key] item.update({ "runtime_mounted": True, "runtime_source": rt["rt_source"], "runtime_fstype": rt["rt_fstype"], "runtime_options": rt["rt_options"], "runtime_readonly": rt["rt_readonly"], "runtime_reachable": health["reachable"], "runtime_error": health["error"], }) matched_targets.add(target_key) elif running: # CT is running but the configured mount isn't in # /proc//mounts — divergence. Could be a startup # error, missing source, ACL problem, etc. item["runtime_mounted"] = False item["runtime_error"] = "configured but not mounted" else: item["runtime_mounted"] = None # CT down — no runtime info out.append(item) # Ad-hoc remote mounts inside the running CT (NFS/CIFS/SMB) that # don't correspond to any mpX config entry — these are mounts the # user did from inside the CT (e.g. `mount -t nfs ...`) and the # original Sprint 13.24 issue revolves around catching them. ad_hoc: list[dict[str, Any]] = [] if running: ad_hoc_candidates = [ rt for rt in rt_mounts if _mount_target_key(rt["rt_target"]) not in matched_targets and _REMOTE_FS_RE.match(rt["rt_fstype"]) ] # Same parallelisation as the configured-mp loop: stat'ing # stale NFS exports serially can dominate the request and # push it past the proxy timeout. Capacity (`df`) is fetched # in the SAME pool so the UI can render the usage bar for # ad-hoc NFS/CIFS mounts too — null capacity was a regression # spotted on CT 103 /mnt/Media. Skip df when stat already # showed the mount as unreachable, otherwise the df subprocess # blocks on the same broken export. if ad_hoc_candidates: with ThreadPoolExecutor(max_workers=max_workers) as pool: def _gather_adhoc(rt): h = _stat_via_host(host_pid, rt["rt_target"]) if h.get("reachable"): # NFS/CIFS mounts done inside the CT live in the # container's own mount namespace and aren't # visible to `df` from the host even via # /proc//root — use `pct exec df` instead. cap = _df_via_pct_exec(vmid, rt["rt_target"]) else: cap = {"total_bytes": None, "used_bytes": None, "available_bytes": None} return rt, h, cap results = list(pool.map(_gather_adhoc, ad_hoc_candidates)) for rt, health, cap in results: ad_hoc.append({ "mp_index": "", "source": rt["rt_source"], "target": rt["rt_target"], "type": "ad_hoc", "origin_storage": "", "origin_storage_type": "", "origin_label": rt["rt_source"], "config_options": {}, "config_flags": [], "total_bytes": cap["total_bytes"], "used_bytes": cap["used_bytes"], "available_bytes": cap["available_bytes"], "runtime_mounted": True, "runtime_source": rt["rt_source"], "runtime_fstype": rt["rt_fstype"], "runtime_options": rt["rt_options"], "runtime_readonly": rt["rt_readonly"], "runtime_reachable": health["reachable"], "runtime_error": health["error"], }) return { "ok": True, "vmid": vmid, "running": running, "mount_points": out, "ad_hoc": ad_hoc, }