from mkosi.log import ARG_DEBUG, die
from mkosi.partition import finalize_root, find_partitions
from mkosi.run import AsyncioThread, find_binary, fork_and_wait, run, spawn, workdir
-from mkosi.tree import copy_tree, rmtree
+from mkosi.tree import copy_tree, make_nocow, rmtree
from mkosi.user import INVOKING_USER, become_root_in_subuid_range, become_root_in_subuid_range_cmd
from mkosi.util import (
PathString,
logging.debug(f"- {k}={v}")
-def make_nocow(config: Config, path: Path) -> None:
- run(
- ["chattr", "+C", workdir(path)],
- check=False,
- stderr=subprocess.DEVNULL if not ARG_DEBUG.get() else None,
- sandbox=config.sandbox(options=["--bind", path, workdir(path)]),
- )
-
-
@contextlib.contextmanager
def start_journal_remote(config: Config, sockfd: int) -> Iterator[None]:
assert config.forward_journal
# at the same time.
d.mkdir(exist_ok=True, parents=True)
# Make sure COW is disabled so systemd-journal-remote doesn't complain on btrfs filesystems.
- make_nocow(config, d)
+ make_nocow(d, sandbox=config.sandbox)
INVOKING_USER.chown(d)
with tempfile.NamedTemporaryFile(mode="w", prefix="mkosi-journal-remote-config-") as f:
try:
def copy() -> None:
- if config.output_format in (OutputFormat.disk, OutputFormat.esp):
- attr = run(
- ["lsattr", "-l", workdir(src)],
- sandbox=config.sandbox(options=["--ro-bind", src, workdir(src)]),
- stdout=subprocess.PIPE,
- ).stdout
-
- if "No_COW" in attr:
- tmp.touch()
- make_nocow(config, tmp)
-
copy_tree(
src,
tmp,
dir=drive.directory or "/var/tmp",
prefix=f"mkosi-drive-{drive.id}",
) as file:
- make_nocow(config, Path(file.name))
+ make_nocow(Path(file.name), sandbox=config.sandbox)
file.truncate(round_up(drive.size, resource.getpagesize()))
yield Path(file.name)
shutil.copystat(tmp / d, dst / d)
+def make_nocow(path: Path, *, sandbox: SandboxProtocol = nosandbox) -> None:
+ run(
+ ["chattr", "+C", workdir(path)],
+ check=False,
+ stderr=subprocess.DEVNULL if not ARG_DEBUG.get() else None,
+ sandbox=sandbox(options=["--bind", path, workdir(path)]),
+ )
+
+
def copy_tree(
src: Path,
dst: Path,
attrs += ",xattr"
def copy() -> None:
+ if src.is_file():
+ attr = run(
+ ["lsattr", "-l", workdir(src)],
+ sandbox=sandbox(options=["--ro-bind", src, workdir(src)]),
+ stdout=subprocess.PIPE,
+ ).stdout
+
+ if "No_COW" in attr:
+ fdst = dst / src.name if dst.is_dir() else dst
+ fdst.touch()
+ make_nocow(fdst, sandbox=sandbox)
+
cmdline: list[PathString] = [
"cp",
"--recursive",