"""CTCLR-06: only the selected local stack and its existing storage are mounted.""" from contextlib import contextmanager import json import os from pathlib import Path import time import compose_backend as backend import field_profiles as fp import server_context as server from data_safety import atomic, digest, safe class Host: def __init__(self): if not server.portable():raise ValueError('SERVER_ADAPTER_REVIEW_REQUIRED') server.guard_disks();server.verify_tools() a,site,cp,approved,live,source=backend.inspect() if approved!=live:raise ValueError('REVIEW_PENDING_CONFIG_FIRST') self.doc=approved;self.active=a self.env=approved['services']['analytics'].get('environment',{}) self.mounts=approved['services']['analytics'].get('volumes',[]) self.image=approved['services']['analytics']['image'] self.root=fp.STATE/'data-jobs' self.root.mkdir(mode=0o700,parents=True,exist_ok=True) def host_path(self,inside): p=Path(inside) matches=[m for m in self.mounts if m['type']=='bind' and (p==Path(m['target']) or Path(m['target']) in p.parents)] if len(matches)!=1:raise ValueError('OWNED_BIND_REQUIRED') m=matches[0];path=safe(Path(m['source'])/p.relative_to(m['target'])) if not path.exists():raise ValueError('DATA_PATH_MISSING') if not any(path==Path(d['path']) or Path(d['path']) in path.parents for d in server.CURRENT['disks']): raise ValueError('UNGUARDED_DATA_PATH') return path def paths(self): return {k:self.host_path(self.env[v]) for k,v in ( ('artifact','CONTEXTRA_ARTIFACT_ROOT'),('training','CONTEXTRA_TRAINING_ROOT'),('camera','STREAMS_RUNTIME_PATH'))} def writers(self): roles={server.service(x) for x in ('analytics','ai-gateway','ai-control-sidecar')} return [r for r in backend.containers() if r['State']['Running'] and r['Config']['Labels']['com.docker.compose.service'] in roles] @contextmanager def dependencies(self,kind): if self.writers():raise ValueError('STOP_ANALYSIS_BEFORE_MAINTENANCE') roles=[] if kind in ('data','artifact','db'):roles.append('postgresql') if kind=='redis':roles.append('redis') started=[] try: rows=backend.containers() for role in roles: found=[r for r in rows if r['Config']['Labels']['com.docker.compose.service']==server.service(role)] if len(found)!=1:raise ValueError('SERVICE_IDENTITY') row=found[0] if not row['State']['Running']: backend.execute(['docker','start',row['Id']],30);started.append(row['Id']) until=time.monotonic()+45 while True: now=json.loads(backend.execute(['docker','inspect',row['Id']]))[0] if now['State'].get('Health',{}).get('Status')=='healthy':break if not now['State']['Running'] or time.monotonic()>until:raise ValueError('DEPENDENCY_NOT_READY') time.sleep(1) yield finally: for ident in reversed(started): backend.execute(['docker','stop','--timeout','30',ident],40) r=json.loads(backend.execute(['docker','inspect',ident]))[0]['State'] if r['Running'] or r['ExitCode']==137:raise RuntimeError('DEPENDENCY_STOP_FAILED') def run_worker(self,job,request): from maintenance_process import run if self.writers():raise ValueError('WRITER_RESTARTED') if backend.active()!=self.active:raise ValueError('ACTIVE_CHANGED') server.guard_disks() rows=backend.containers() peers=[r for r in rows if r['Config']['Labels']['com.docker.compose.service']==server.service('postgresql')] networks=list(peers[0]['NetworkSettings']['Networks']) if len(peers)==1 else [] if len(networks)!=1:raise ValueError('PRIVATE_NETWORK_REQUIRED') names=('DATABASE_URL','CONTEXTRA_ARTIFACT_ROOT','CONTEXTRA_ARTIFACT_PUBLIC_BASE','CONTEXTRA_EVENT_ENDPOINT', 'CONTEXTRA_EVENT_ARTIFACT_ENABLED','CONTEXTRA_ARTIFACT_HTTP_ENABLED','CONTEXTRA_TRAINING_ROOT', 'STREAMS_RUNTIME_PATH','CONTEXTRA_METADATA_REDIS_URL','CONTEXTRA_METADATA_NAMESPACE') values={k:v for k,v in self.env.items() if k in names or k.startswith('CONTEXTRA_ARTIFACT_')} if any('\n' in v or '\r' in v for v in values.values()):raise ValueError('ENV_LINE') envfile=job/'worker.env';atomic(envfile,('\n'.join(k+'='+v for k,v in values.items())+'\n').encode()) name='ct-data-'+str(os.getpid())+'-'+str(time.time_ns()) args=['docker','run','-i','--name',name,'--pull=never','--read-only','--init','--cpus=2','--memory=768m','--pids-limit=64', '--security-opt=no-new-privileges','--cap-drop=ALL','--network='+networks[0], '--label=contextra.maintenance.job='+name,'--env-file',str(envfile), '-e','PYTHONPATH=/opt/contextra:/ct-tools','-e','OPENBLAS_NUM_THREADS=1','-e','NVIDIA_VISIBLE_DEVICES=void', '--tmpfs','/tmp:rw,nosuid,nodev,noexec,size=64m', '--mount','type=bind,src='+str(Path(__file__).parent)+',dst=/ct-tools,readonly', '--mount','type=bind,src='+str(job)+',dst=/ct-job'] seen=set();groups=set() for host in self.paths().values(): m=next(m for m in self.mounts if m['type']=='bind' and (Path(m['source'])==host or Path(m['source']) in host.parents)) if m['target'] in seen:continue if ',' in m['source'] or ',' in m['target']:raise ValueError('MOUNT_SEPARATOR') # Root with ALL capabilities dropped still needs the data owner's group. gid=str(Path(m['source']).stat().st_gid) if gid not in groups: args+=['--group-add',gid];groups.add(gid) args+=['--mount','type=bind,src='+m['source']+',dst='+m['target']];seen.add(m['target']) args+=['--entrypoint','python3',self.image,'-B','/ct-tools/data_worker.py'] try: r=run(args,input=json.dumps(request).encode(),capture_output=True,timeout=240) try:out=json.loads(r.stdout) except (ValueError,UnicodeError):raise ValueError('WORKER_OUTPUT_INVALID') from None if r.returncode: atomic(job/'error.json',out) raise ValueError(json.dumps(out,ensure_ascii=True)) return out finally: check=run(['docker','inspect',name],capture_output=True,timeout=10) if check.returncode==0: row=json.loads(check.stdout)[0] if row['Config'].get('Labels',{}).get('contextra.maintenance.job')!=name:raise ValueError('HELPER_IDENTITY') run(['docker','rm','-f',row['Id']],capture_output=True,check=True,timeout=15) envfile.unlink(missing_ok=True)