"""Three path scopes and filesystem identity. Never equate a directory with a mounted disk.""" from __future__ import annotations import os from pathlib import Path import shutil from .core import absolute, require, run, parse_json, write_json, atomic_write, KitError LAYOUT = { 'site': ('/etc/contextra', '설정·인증 파일. 비밀값 공유/삭제 금지'), 'runtime': ('/var/lib/contextra/runtime', '실행 상태. 서비스 중 수동 수정/삭제 금지'), 'models': ('/opt/contextra/models', '모델 원본·ONNX·엔진. model 명령으로 변경'), 'artifacts': ('/data/contextra/artifacts', '운영 Event 자료. 원장과 연결되므로 수동 삭제 금지'), 'debug': ('/data/contextra/debug', '진단 스냅샷/클립. 실행 중 자료 덮어쓰기 금지'), 'offline': ('/data/contextra/offline', 'inputs 원본과 results 시험 결과를 분리'), 'diagnostics': ('/data/contextra/diagnostics', '민감정보를 제외한 진단 요약'), } def mount_info(path, runner=run): p = absolute(path, no_symlinks=True) while not p.exists(): p = p.parent _, raw = runner(['findmnt', '--json', '--target', str(p), '--output', 'TARGET,SOURCE,FSTYPE,UUID,OPTIONS']) rows = parse_json(raw).get('filesystems', []) require(len(rows) == 1, 'MOUNT_UNKNOWN', '실제 저장 디스크를 확인할 수 없습니다.') row = rows[0] return {k: row.get(k) or '' for k in ('target', 'source', 'fstype', 'uuid', 'options')} def capture(name, physical, data_root, *, allow_root=False, runner=run): require(name in LAYOUT, 'STORAGE_NAME', '알 수 없는 저장소 종류입니다.') p = absolute(physical, no_symlinks=True) logical = absolute(data_root, no_symlinks=True) / name info = mount_info(p, runner) require(allow_root or info['target'] != '/', 'ROOT_STORAGE_CONFIRMATION', f'{name}: 현재 시스템 디스크 위의 폴더입니다. 별도 디스크가 맞는지 확인하세요.', 'deployctl configure (의도한 root 저장소만 명시 승인)') require(info['uuid'] or info['source'].startswith('/dev/'), 'MOUNT_ID_UNAVAILABLE', 'v1은 UUID 또는 /dev 장치로 식별 가능한 로컬 filesystem만 지원합니다.') return dict(physical=str(p), logical=str(logical), container=LAYOUT[name][0], expected={k: info[k] for k in ('target', 'source', 'fstype', 'uuid')}, allow_root=allow_root) def check_entry(entry, *, runner=run, writable=False): p = absolute(entry['physical'], no_symlinks=True) e, a = entry['expected'], mount_info(p, runner) same = a['target'] == e['target'] and a['fstype'] == e['fstype'] same = same and (a['uuid'] == e['uuid'] if e.get('uuid') else a['source'] == e['source']) require(same, 'STORAGE_MOUNT_MISMATCH', f'{p}: 폴더는 있어도 등록된 디스크와 다릅니다. 시스템 SSD에 대신 저장하지 않습니다.', 'findmnt -T ' + str(p) + ' ; deployctl storage show') require(a['target']!='/' or entry.get('allow_root') is True,'ROOT_STORAGE_CONFIRMATION','시스템 디스크 저장은 최초 명시 승인이 필요합니다.') require('ro' not in a['options'].split(','), 'STORAGE_READ_ONLY', '저장 디스크가 읽기 전용입니다.') if writable and p.exists(): require(os.access(p, os.W_OK | os.X_OK), 'STORAGE_PERMISSION', '저장 폴더의 소유자/쓰기 권한을 확인하세요.') nearest = p while not nearest.exists(): nearest = nearest.parent return dict(physical=str(p), logical=entry['logical'], container=entry['container'], filesystem=a, free_bytes=shutil.disk_usage(nearest).free) def validate_map(site): require(site.get('schema') == 1 and set(site.get('storage', {})) == set(LAYOUT), 'STORAGE_SCHEMA', '7개 저장소가 포함된 schema=1 site 설정이 필요합니다.') install, data = absolute(site['install_root'], no_symlinks=True), absolute(site['data_root'], no_symlinks=True) require(install != data and not install.is_relative_to(data) and not data.is_relative_to(install), 'ROOTS_OVERLAP', '프로그램 설치 root와 데이터 root를 서로 분리하세요.') paths = [] for name, entry in site['storage'].items(): p = absolute(entry['physical'], no_symlinks=True) logical = absolute(entry['logical'], no_symlinks=True) require(logical == data / name and entry['container'] == LAYOUT[name][0], 'STORAGE_SCOPE', '호스트 통합경로와 고정 컨테이너 경로가 맞지 않습니다.') require(not p.is_relative_to(install) and not install.is_relative_to(p), 'ROOTS_OVERLAP', '영속 데이터를 release/설치 폴더 안에 두지 마세요.') require(all(not p.is_relative_to(x) and not x.is_relative_to(p) for x in paths), 'STORES_OVERLAP', '서로 다른 저장소 폴더를 겹쳐 지정하지 마세요.') paths.append(p) return site def checks(site, **kwargs): validate_map(site) return {name: check_entry(entry, **kwargs) for name, entry in site['storage'].items()} def prepare(site, *, bind_views=False, runner=run): """Only called after approval. Physical paths remain Docker source of truth.""" checks(site, runner=runner) root = absolute(site['data_root'], no_symlinks=True) root.mkdir(parents=True, exist_ok=True) text = '# Contextra 데이터 안내\n\n실제 파일은 storage-map의 디스크에 저장됩니다.\n' for name, entry in site['storage'].items(): physical, logical = Path(entry['physical']), Path(entry['logical']) physical.mkdir(parents=True, exist_ok=True, mode=0o700 if name == 'site' else 0o750) text += f'\n- {name}: {LAYOUT[name][1]}\n' readme = physical / 'README.ko.md' if not readme.exists(): atomic_write(readme, f'# {name}\n\n용도: {LAYOUT[name][1]}\n\n' f'HOST 실제: `{physical}`\n\nHOST 통합: `{logical}`\n\n' f'CONTAINER: `{entry["container"]}`\n\n' '두 호스트 경로는 복사본이 아닙니다. 어느 쪽에서 삭제해도 같은 파일이 삭제됩니다.\n' '실제 위치 확인: `deployctl storage show`\n문제 확인: `deployctl doctor`\n', new=True, mode=0o644) if bind_views and physical != logical: logical.mkdir(parents=True, exist_ok=True) mounted = mount_info(logical, runner) if mounted['target'] == str(logical): require(os.path.samefile(physical, logical), 'LOGICAL_MOUNT_CONFLICT', '통합경로가 다른 폴더에 연결돼 있습니다.') else: require(not any(logical.iterdir()), 'LOGICAL_NOT_EMPTY', '통합경로에 기존 파일이 있습니다. 가려서 덮지 않습니다.') runner(['mount', '--bind', str(physical), str(logical)]) require(os.path.samefile(physical, logical), 'BIND_FAILED', '통합경로 연결을 확인하지 못했습니다.') for suffix in ('snapshots', 'clips'): (Path(site['storage']['debug']['physical']) / suffix).mkdir(exist_ok=True) for suffix in ('inputs', 'results'): (Path(site['storage']['offline']['physical']) / suffix).mkdir(exist_ok=True) if not (root / 'README.ko.md').exists(): atomic_write(root / 'README.ko.md', text, new=True, mode=0o644) def move_plan(site, name, destination): require(name in LAYOUT and name not in ('site', 'runtime'), 'MOVE_MANUAL_REQUIRED', '인증·실행 상태 저장소는 별도 정지/복구 계획으로 이전하세요.') checks(site) dest = absolute(destination, no_symlinks=True) src = Path(site['storage'][name]['physical']) require(dest != src and not dest.is_relative_to(src) and not src.is_relative_to(dest), 'MOVE_OVERLAP', '복사 원본과 목적지가 겹칩니다.') require(not dest.exists() or not any(dest.iterdir()), 'DESTINATION_NOT_EMPTY', '빈 전용 목적지 폴더를 사용하세요.') return dict(status='PLAN_ONLY', source=str(src), destination=str(dest), copy=['rsync', '-aH', '--numeric-ids', '--', str(src) + '/', str(dest) + '/'], verify=['rsync', '-aHnci', '--numeric-ids', '--', str(src) + '/', str(dest) + '/'], steps=['관련 채널 Stop/reap와 모든 writer 정지 확인', '복사', 'checksum 비교 결과0 확인', 'configure로 새 filesystem 등록', '컨테이너 재생성', '원본 보존'], automatic_move=False, reason='v1은 무단 대용량 이전/원본 삭제를 실행하지 않습니다.')