Add destination audit wizard with selective deletion

New top-level wizard option (2) lets users inspect and clean up an
existing destination before migration. Queries all SMB shares, NFS
exports, iSCSI objects, datasets, and zvols; displays a structured
inventory report; then offers per-category deletion with escalating
warnings — standard confirm for shares/iSCSI, explicit "DELETE" phrase
required for zvols and datasets to guard against accidental data loss.

Adds to client.py: query_destination_inventory, delete_smb_shares,
delete_nfs_exports, delete_zvols, delete_datasets.
Adds to cli.py: _fmt_bytes, _print_inventory_report, _run_audit_wizard.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-05 15:56:10 -05:00
parent 3fd9e6b6a8
commit c28ce9e3b8
2 changed files with 413 additions and 2 deletions

View File

@@ -382,3 +382,107 @@ async def create_missing_zvols(
) as client:
for name, volsize in zvols.items():
await create_zvol(client, name, volsize)
# ─────────────────────────────────────────────────────────────────────────────
# Destination inventory
# ─────────────────────────────────────────────────────────────────────────────
async def query_destination_inventory(client: TrueNASClient) -> dict[str, list]:
"""
Query all current configuration from the destination system.
Returns a dict with keys: smb_shares, nfs_exports, datasets, zvols,
iscsi_extents, iscsi_initiators, iscsi_portals, iscsi_targets, iscsi_targetextents.
Each value is a list (may be empty if the query fails or returns nothing).
"""
result: dict[str, list] = {}
for key, method, params in [
("smb_shares", "sharing.smb.query", None),
("nfs_exports", "sharing.nfs.query", None),
("datasets", "pool.dataset.query", [[["type", "=", "FILESYSTEM"]]]),
("zvols", "pool.dataset.query", [[["type", "=", "VOLUME"]]]),
("iscsi_extents", "iscsi.extent.query", None),
("iscsi_initiators", "iscsi.initiator.query", None),
("iscsi_portals", "iscsi.portal.query", None),
("iscsi_targets", "iscsi.target.query", None),
("iscsi_targetextents", "iscsi.targetextent.query", None),
]:
try:
result[key] = await client.call(method, params) or []
except RuntimeError as exc:
log.warning("Could not query %s: %s", key, exc)
result[key] = []
return result
async def delete_smb_shares(
client: TrueNASClient, shares: list[dict]
) -> tuple[int, int]:
"""Delete SMB shares by ID. Returns (deleted, failed)."""
deleted = failed = 0
for share in shares:
try:
await client.call("sharing.smb.delete", [share["id"]])
log.info(" Deleted SMB share %r", share.get("name"))
deleted += 1
except RuntimeError as exc:
log.error(" Failed to delete SMB share %r: %s", share.get("name"), exc)
failed += 1
return deleted, failed
async def delete_nfs_exports(
client: TrueNASClient, exports: list[dict]
) -> tuple[int, int]:
"""Delete NFS exports by ID. Returns (deleted, failed)."""
deleted = failed = 0
for export in exports:
try:
await client.call("sharing.nfs.delete", [export["id"]])
log.info(" Deleted NFS export %r", export.get("path"))
deleted += 1
except RuntimeError as exc:
log.error(" Failed to delete NFS export %r: %s", export.get("path"), exc)
failed += 1
return deleted, failed
async def delete_zvols(
client: TrueNASClient, zvols: list[dict]
) -> tuple[int, int]:
"""Delete zvols. Returns (deleted, failed)."""
deleted = failed = 0
for zvol in zvols:
try:
await client.call("pool.dataset.delete", [zvol["id"], {"recursive": True}])
log.info(" Deleted zvol %r", zvol["id"])
deleted += 1
except RuntimeError as exc:
log.error(" Failed to delete zvol %r: %s", zvol["id"], exc)
failed += 1
return deleted, failed
async def delete_datasets(
client: TrueNASClient, datasets: list[dict]
) -> tuple[int, int]:
"""
Delete datasets deepest-first to avoid parent-before-child errors.
Skips pool root datasets (no '/' in the dataset name).
Returns (deleted, failed).
"""
sorted_ds = sorted(
(d for d in datasets if "/" in d["id"]),
key=lambda d: d["id"].count("/"),
reverse=True,
)
deleted = failed = 0
for ds in sorted_ds:
try:
await client.call("pool.dataset.delete", [ds["id"], {"recursive": True}])
log.info(" Deleted dataset %r", ds["id"])
deleted += 1
except RuntimeError as exc:
log.error(" Failed to delete dataset %r: %s", ds["id"], exc)
failed += 1
return deleted, failed