"""Bundled adapter, not the installed ct executable. No network configuration changes.""" import copy import importlib import json import os from pathlib import Path import stat import sys import time from ..core import absolute, canonical, docker, read_json, require, run, sha256 class Host: def __init__(self): root = Path(__file__).resolve().parents[2] tools = root/'tools/icerti_offline/maintenance' require(tools.is_dir(), 'TOOLKIT_MISSING', '패키지 전체 폴더를 반입하세요.') sys.path.insert(0, str(tools)) self.server = importlib.import_module('server_context') self.server.CURRENT = self.server.load() require(self.server.CURRENT is not None, 'LEGACY_INVENTORY_REQUIRED', '서버 정의 없는 과거 현장판입니다. 적용하지 말고 설치 정보를 반출하세요.') require(self.server.portable(), 'BACKEND_UNSUPPORTED', 'transaction 현장판은 이번 test 적용 대상이 아닙니다. 호환 패키지가 필요합니다.') # Paths are initialized after the validated host identity, never by the USB directory name. self.fp = importlib.import_module('field_profiles') self.cb = importlib.import_module('compose_backend') # This process uses the bundled adapter, pinned to the local daemon even with a remote Docker context. self.cb.execute = self.execute_local self.ds = importlib.import_module('data_safety') self.s = self.server.CURRENT self.state = self.cb.state_path() def execute_local(self, args, timeout=30): from ..compose_snapshot import command_environment require(args and args[0]=='docker','COMMAND_SCOPE','이 adapter는 로컬 Docker 명령만 허용합니다.') _,output=run(['docker','--host','unix:///var/run/docker.sock',*args[1:]], timeout=timeout,limit=8*1024**2,environment=command_environment()) return output def lock(self): return self.ds.lock(self.fp.STATE/'data-maintenance.lock') def rows(self): return {x['Config']['Labels']['com.docker.compose.service']: x for x in self.cb.containers()} def snapshot(self): self.server.verify_tools(); self.server.guard_disks() a, site, cp, approved, live, source = self.cb.inspect() require(approved == live and self.cb.canonical(source) == self.cb.canonical( read_json(self.state/'input-baseline.json', 8*1024**2)), 'SOURCE_DRIFT', '지원 밖 설정 차이가 있습니다. 자동 승인하지 않습니다.') rows = self.rows() require(set(rows) == set(self.s['services'].values()), 'SERVICE_SCOPE', '5개 서비스 구성을 확인하세요.') images = {r: rows[self.server.service(r)]['Image'] for r in self.server.ROLES} require(all(approved['services'][r]['image'] == images[r] for r in images), 'IMAGE_DRIFT', '실행 image와 관리 기록이 다릅니다.') paths = [self.server.CONFIG, self.state/'active.json', self.state/'site.json', self.state/'protected-bindings.json', self.state/'input-baseline.json', Path(a['compose']), Path(a['physical']), Path(a['compose']).with_name('approved.json'), Path(self.s['paths']['tools'])/'installed-files.json'] return dict(active=a, images=images, files={str(p): sha256(p) for p in paths}, definition_sha256=sha256(self.server.CONFIG), source_sha256=self.cb.canonical(source), server=self.s['server'], backend=self.s['backend']) def stopped(self): require(not any(r['State']['Running'] or r['State']['Status'] in ('paused','restarting','removing') for r in self.rows().values()), 'MAINTENANCE_REQUIRED', '사용자가 기존 ct로 채널 정리와 서비스 정지를 완료한 뒤 적용하세요. 도구가 대신 Stop하지 않습니다.') def pending_clear(self): for name in ('pending.json', 'apply-pending.json', 'network-pending.json'): require(not (self.fp.STATE/name).exists(), 'PENDING_OPERATION', '이전 유지보수 기록을 먼저 확인하세요: '+name) def host_path(self, target): absolute(target) cfg = read_json(self.cb.current_path(), 8*1024**2)['services']['analytics'] matches = [v for v in cfg.get('volumes', []) if v['type']=='bind' and (target==v['target'] or target.startswith(v['target']+'/'))] require(matches, 'CONFIG_BIND', '설정의 실제 host bind를 찾지 못했습니다.') m = max(matches, key=lambda v:len(v['target'])) return absolute(Path(m['source'])/Path(target).relative_to(m['target']), no_symlinks=True) def settings_paths(self): cfg = read_json(self.cb.current_path(), 8*1024**2)['services']['analytics'] env = cfg.get('environment', {}) result = [] for key, kind, fmt in [('CONTEXTRA_CONTROL_TEMPLATE_FILE','template','json'), ('STREAMS_CONFIG_PATH','base','yaml'), ('STREAMS_RUNTIME_PATH','runtime','yaml')]: require(env.get(key), 'CONFIG_PATH', '설정 위치가 없습니다: '+key) path = self.host_path(env[key]); require(path.is_file(), 'CONFIG_PATH', '기존 설정 파일이 없습니다.') result.append(dict(path=str(path), kind=kind, format=fmt)) policy = env.get('CONTEXTRA_FIGHT_DISPLAY_POLICY') or env.get('CONTEXTRA_ARTIFACT_ROOT', '/data/contextra/artifacts')+'/.fight-display-policy.json' policy=self.host_path(policy) require(len({x['path'] for x in result}|{str(policy)})==4,'CONFIG_ALIAS','설정/정책 경로가 겹칩니다.') return result, policy def compose(self, path, arguments, timeout=120): from ..compose_snapshot import command_environment args=self.cb.compose_args(path) return run(['docker','--host','unix:///var/run/docker.sock',*args[1:],*arguments], timeout=timeout,environment=command_environment(),limit=8*1024**2) def validate_transaction(self, path, expected): from apply_source import normalize _,raw=self.compose(path,['config','--format','json']) require(normalize(self.server.logical(json.loads(raw)))==normalize(expected), 'COMPOSE_ROUNDTRIP','Compose 해석 결과가 계획과 다릅니다. 환경변수/경로를 검토하세요.') def stage_settings(self, private, image, toolkit): name = 'ct-offline-settings-'+str(os.getpid()) argv = ['docker','--host','unix:///var/run/docker.sock','run','--rm','--pull=never','--runtime=runc', '--name',name,'--network=none','--read-only','--cpus=2','--memory=1g','--pids-limit=128', '--cap-drop=ALL','--security-opt=no-new-privileges','--tmpfs','/tmp:rw,size=32m', '-e','NVIDIA_VISIBLE_DEVICES=void','-e','PYTHONDONTWRITEBYTECODE=1','-e','OPENBLAS_NUM_THREADS=1', '--mount','type=bind,src='+str(private)+',dst=/job', '--mount','type=bind,src='+str(toolkit)+',dst=/toolkit,readonly', '--entrypoint','python3',image,'-B','/toolkit/deployment_kit/offline/settings.py','/job'] try: run(argv, timeout=120) finally: code, raw = docker('inspect', name, check=False) if code == 0: row = json.loads(raw)[0] require(row['Image'] == image, 'TRIAL_IDENTITY', '시험 컨테이너 소유권이 달라졌습니다.') docker('rm', '-f', name) def transaction(self, release_id, image): a = self.cb.active(); new = copy.deepcopy(read_json(self.cb.current_path(), 8*1024**2)) new['services']['analytics']['image'] = image tx = self.state/'transactions'/('offline-'+release_id) physical = self.server.physical(new) active = dict(a, transaction=tx.name, compose=str(tx/'compose.json'), physical=str(tx/'compose.physical.json'), compose_semantic_sha256=self.cb.canonical(new)) return tx, {'compose.json': new, 'approved.json': new, 'compose.physical.json': physical}, active def load_images(self, package, archives, expected): present=True for image in expected.values(): code,raw=docker('image','inspect','--format','{{.Id}}',image,check=False) present = present and code==0 and raw.strip()==image if present:return import shutil total=sum((package/n).stat().st_size for n in archives) _,docker_root=docker('info','--format','{{.DockerRootDir}}') require(shutil.disk_usage(docker_root.strip()).free > total+15*1024**3, 'IMAGE_LOAD_SPACE','Docker 반입 임시 공간과15GiB 여유가 부족합니다. 자동 정리하지 않습니다.') for archive in archives: # A daemon load timeout is an unknown outcome, not a cancelled transaction. docker('load', '-i', str(package/archive), timeout=1800, limit=8*1024**2) for image in expected.values(): _, raw = docker('image','inspect','--format','{{.Id}}',image) require(raw.strip()==image, 'IMAGE_LOAD', '반입 image ID가 다릅니다.') def create_containers(self): a = self.cb.active() # Atomic host file replacement requires a new container even with the same image ID. self.compose(a['physical'],['up','--no-start','--force-recreate','--no-deps','--no-build','--pull','never',self.server.service('analytics')],timeout=180) def start_and_verify(self, expected): self.server.guard_disks() a,site,cp,base,live,source = self.cb.inspect() require(a.get('enrolled') and live==base, 'START_CONTRACT', '기존 연결/설정 승인이 필요합니다.') import field_network import field_runtime profile = self.fp.load(a['profile']) require(site['ai_ip'] in field_network.addresses(profile['network']['interface']), 'ADDRESS_DRIFT', '현재 주소와 승인 주소가 다릅니다. NIC를 바꾸지 않습니다.') field_runtime.stage_ready(Path(self.fp.read_json(self.fp.STATE/'last-applied.json')['stage'])) self.cb.guard_images(base) roles = ['postgresql','redis','analytics','ai-control-sidecar'] self.compose(a['physical'],['up','-d','--no-build','--pull','never',*[self.server.service(r) for r in roles]],timeout=240) end = time.monotonic()+120 while True: rows=self.rows() ready=all(rows[self.server.service(r)]['State']['Running'] and rows[self.server.service(r)]['State'].get('Health',{}).get('Status')=='healthy' for r in roles) if ready:break require(time.monotonic()