feat: add modify docker config
This commit is contained in:
Executable
+281
@@ -0,0 +1,281 @@
|
||||
#!/usr/bin/env python3
|
||||
# 按类别增量更新 /etc/docker/daemon.json（仅补缺失项，不覆盖已有配置）
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
DAEMON_JSON = Path("/etc/docker/daemon.json")
|
||||
PATCH_DIR_NAME = Path("docker") / "daemon.d"
|
||||
|
||||
CATEGORIES: dict[str, str] = {
|
||||
"base": "日志轮转等基础项（log-driver / log-opts）",
|
||||
"mirror": "镜像加速（registry-mirrors）",
|
||||
}
|
||||
|
||||
|
||||
def dotfile_root() -> Path:
|
||||
return Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def usage_text() -> str:
|
||||
cats = "\n".join(f" {name:<8} {desc}" for name, desc in CATEGORIES.items())
|
||||
return f"""用法: docker-daemon-config [选项] [类别 ...]
|
||||
|
||||
将 ~/.dotfile/docker/daemon.d/ 下对应片段合并进 /etc/docker/daemon.json。
|
||||
已有 key 保留不变；仅插入缺失的 key（嵌套对象递归补全）。
|
||||
|
||||
类别:
|
||||
{cats}
|
||||
all 应用上述全部类别
|
||||
|
||||
示例:
|
||||
sudo ff script docker-daemon-config base
|
||||
sudo ff script docker-daemon-config mirror
|
||||
sudo ff script docker-daemon-config all
|
||||
sudo ff script docker-daemon-config base --dry-run
|
||||
|
||||
选项:
|
||||
-h, --help 显示此帮助
|
||||
--list 列出可用类别
|
||||
--dry-run 只打印合并结果，不写文件
|
||||
--no-backup 不备份现有 daemon.json
|
||||
--no-restart 合并后不尝试 reload/restart docker
|
||||
"""
|
||||
|
||||
|
||||
def fill_missing(base: Any, patch: Any) -> Any:
|
||||
"""仅补 patch 中有、base 中没有的 key；已有 key 不覆盖。"""
|
||||
if not isinstance(patch, dict):
|
||||
return base
|
||||
if not isinstance(base, dict):
|
||||
return copy.deepcopy(patch)
|
||||
|
||||
out = copy.deepcopy(base)
|
||||
for key, patch_value in patch.items():
|
||||
if key not in out:
|
||||
out[key] = copy.deepcopy(patch_value)
|
||||
elif isinstance(out[key], dict) and isinstance(patch_value, dict):
|
||||
out[key] = fill_missing(out[key], patch_value)
|
||||
return out
|
||||
|
||||
|
||||
def load_json(path: Path) -> dict[str, Any]:
|
||||
text = path.read_text(encoding="utf-8").strip()
|
||||
if not text:
|
||||
return {}
|
||||
data = json.loads(text)
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError(f"{path} 根节点必须是 JSON 对象")
|
||||
return data
|
||||
|
||||
|
||||
def resolve_categories(names: list[str]) -> list[str]:
|
||||
if not names:
|
||||
return []
|
||||
if "all" in names:
|
||||
return list(CATEGORIES.keys())
|
||||
unknown = [name for name in names if name not in CATEGORIES]
|
||||
if unknown:
|
||||
raise ValueError(f"未知类别: {', '.join(unknown)}")
|
||||
# 去重并保持顺序
|
||||
seen: set[str] = set()
|
||||
ordered: list[str] = []
|
||||
for name in names:
|
||||
if name not in seen:
|
||||
seen.add(name)
|
||||
ordered.append(name)
|
||||
return ordered
|
||||
|
||||
|
||||
def patch_path(root: Path, category: str) -> Path:
|
||||
return root / PATCH_DIR_NAME / f"{category}.json"
|
||||
|
||||
|
||||
def collect_patches(root: Path, categories: list[str]) -> dict[str, dict[str, Any]]:
|
||||
patches: dict[str, dict[str, Any]] = {}
|
||||
for category in categories:
|
||||
path = patch_path(root, category)
|
||||
if not path.is_file():
|
||||
raise FileNotFoundError(f"缺少配置片段: {path}")
|
||||
patches[category] = load_json(path)
|
||||
return patches
|
||||
|
||||
|
||||
def diff_added(before: dict[str, Any], after: dict[str, Any], prefix: str = "") -> list[str]:
|
||||
added: list[str] = []
|
||||
for key, value in after.items():
|
||||
path = f"{prefix}{key}"
|
||||
if key not in before:
|
||||
added.append(path)
|
||||
continue
|
||||
if isinstance(value, dict) and isinstance(before.get(key), dict):
|
||||
added.extend(diff_added(before[key], value, prefix=f"{path}."))
|
||||
return added
|
||||
|
||||
|
||||
def validate_daemon_json(path: Path) -> None:
|
||||
with path.open(encoding="utf-8") as fh:
|
||||
json.load(fh)
|
||||
|
||||
dockerd = shutil.which("dockerd")
|
||||
if not dockerd:
|
||||
return
|
||||
try:
|
||||
subprocess.run(
|
||||
[dockerd, "--validate", "--config-file", str(path)],
|
||||
check=True,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
except subprocess.CalledProcessError as exc:
|
||||
detail = (exc.stderr or "").strip()
|
||||
raise RuntimeError(f"dockerd --validate 失败: {detail or exc}") from exc
|
||||
|
||||
|
||||
def reload_docker() -> None:
|
||||
if shutil.which("systemctl") is None:
|
||||
print("[INFO] 未找到 systemctl，请手动重启 Docker 使配置生效")
|
||||
return
|
||||
|
||||
for action in ("reload", "restart"):
|
||||
result = subprocess.run(
|
||||
["systemctl", action, "docker"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
)
|
||||
if result.returncode == 0:
|
||||
print(f"[INFO] 已执行 systemctl {action} docker")
|
||||
return
|
||||
|
||||
print("[WARN] systemctl reload/restart docker 失败，请手动重启 Docker")
|
||||
if result.stdout.strip():
|
||||
print(result.stdout.strip())
|
||||
|
||||
|
||||
def write_daemon_json(content: dict[str, Any], dry_run: bool, backup: bool) -> None:
|
||||
serialized = json.dumps(content, indent=2, ensure_ascii=False) + "\n"
|
||||
|
||||
if dry_run:
|
||||
print(serialized)
|
||||
return
|
||||
|
||||
if os.geteuid() != 0:
|
||||
print("[ERROR] 写入 /etc/docker/daemon.json 需要 root，请使用 sudo 运行")
|
||||
sys.exit(1)
|
||||
|
||||
DAEMON_JSON.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
tmp_path = DAEMON_JSON.with_suffix(".json.tmp")
|
||||
tmp_path.write_text(serialized, encoding="utf-8")
|
||||
try:
|
||||
validate_daemon_json(tmp_path)
|
||||
except Exception:
|
||||
tmp_path.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
if backup and DAEMON_JSON.is_file():
|
||||
stamp = datetime.now().strftime("%Y%m%d%H%M%S")
|
||||
backup_path = DAEMON_JSON.with_name(f"{DAEMON_JSON.name}.bak.{stamp}")
|
||||
shutil.copy2(DAEMON_JSON, backup_path)
|
||||
print(f"[INFO] 已备份到 {backup_path}")
|
||||
|
||||
tmp_path.replace(DAEMON_JSON)
|
||||
os.chmod(DAEMON_JSON, 0o644)
|
||||
print(f"[INFO] 已写入 {DAEMON_JSON}")
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(add_help=False)
|
||||
parser.add_argument("categories", nargs="*", help="要应用的类别")
|
||||
parser.add_argument("-h", "--help", action="store_true")
|
||||
parser.add_argument("--list", action="store_true", help=argparse.SUPPRESS)
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
parser.add_argument("--no-backup", action="store_true")
|
||||
parser.add_argument("--no-restart", action="store_true")
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
|
||||
if args.help:
|
||||
print(usage_text())
|
||||
return 0
|
||||
|
||||
if args.list:
|
||||
for name, desc in CATEGORIES.items():
|
||||
print(f"{name}\t{desc}")
|
||||
print("all\t应用全部类别")
|
||||
return 0
|
||||
|
||||
try:
|
||||
categories = resolve_categories(args.categories)
|
||||
except ValueError as exc:
|
||||
print(f"[ERROR] {exc}")
|
||||
print("可用类别:", ", ".join([*CATEGORIES.keys(), "all"]))
|
||||
return 1
|
||||
|
||||
if not categories:
|
||||
print(usage_text())
|
||||
return 1
|
||||
|
||||
root = dotfile_root()
|
||||
patch_root = root / PATCH_DIR_NAME
|
||||
if not patch_root.is_dir():
|
||||
print(f"[ERROR] 配置目录不存在: {patch_root}")
|
||||
return 1
|
||||
|
||||
existing: dict[str, Any] = {}
|
||||
if DAEMON_JSON.is_file():
|
||||
try:
|
||||
existing = load_json(DAEMON_JSON)
|
||||
except json.JSONDecodeError as exc:
|
||||
print(f"[ERROR] {DAEMON_JSON} 不是合法 JSON: {exc}")
|
||||
return 1
|
||||
|
||||
try:
|
||||
patches = collect_patches(root, categories)
|
||||
except (FileNotFoundError, ValueError) as exc:
|
||||
print(f"[ERROR] {exc}")
|
||||
return 1
|
||||
|
||||
merged = copy.deepcopy(existing)
|
||||
for category in categories:
|
||||
before = copy.deepcopy(merged)
|
||||
merged = fill_missing(merged, patches[category])
|
||||
added = diff_added(before, merged)
|
||||
if added:
|
||||
print(f"[INFO] 类别 {category} 新增: {', '.join(added)}")
|
||||
else:
|
||||
print(f"[INFO] 类别 {category} 无需变更（相关项已存在）")
|
||||
|
||||
if merged == existing:
|
||||
print("[INFO] daemon.json 无变化")
|
||||
return 0
|
||||
|
||||
try:
|
||||
write_daemon_json(merged, dry_run=args.dry_run, backup=not args.no_backup)
|
||||
except Exception as exc:
|
||||
print(f"[ERROR] {exc}")
|
||||
return 1
|
||||
|
||||
if not args.dry_run and not args.no_restart:
|
||||
reload_docker()
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user