"""Analytics-only CLAIM/retention/Event-time rollout using the installed ct contract.""" import ast import copy import json import os from pathlib import Path import subprocess import sys import time from trace_install import Installer, command, image_payload, load, sha class ArtifactInstaller(Installer): def __init__(self, root, expected): super().__init__(root, expected) self.tx = self.state/'transactions'/('artifact-'+self.root.name) def build(self): plan, _ = self.guard() if (self.root/'built.json').exists(): raise ValueError('PRESERVE_EXISTING_BUILD') ctx=self.root/'build-analytics';ctx.mkdir(mode=0o700,exist_ok=True) if any(not n.startswith('app/') for n in self.m['after']): raise ValueError('ANALYTICS_ONLY_PAYLOAD') original=plan['before']['analytics']['image'] base=self.m.get('layerParent',original) if base!=original: old=json.loads(command(['docker','image','inspect',original]))[0] parent=json.loads(command(['docker','image','inspect',base]))[0] if old['Parent']!=base or old['Config']!=parent['Config'] or old['RootFS']['Layers'][:-1]!=parent['RootFS']['Layers']: raise ValueError('UNVERIFIED_LAYER_PARENT') upper=Path(old['GraphDriver']['Data']['UpperDir']) files={str(p.relative_to(upper/'opt/contextra')):sha(p) for p in upper.rglob('*') if p.is_file()} if files!=self.m['carry']:raise ValueError('OLD_LAYER_CONTENT_DRIFT') for name,value in files.items(): if sha(self.root/'carry'/name)!=value:raise ValueError('CARRY_HASH_MISMATCH') image_payload(self.root/'carry',ctx/'payload',list(files)) image_payload(self.root/'payload',ctx/'payload',list(self.m['after'])) self.fp.atomic(ctx/'Dockerfile',('FROM '+base+'\nCOPY payload/ /opt/contextra/\n').encode()) tag='contextra-artifact:'+self.m['server']+'-'+self.root.name with (self.root/'build.log').open('wb') as log: proc=subprocess.run(['docker','build','--pull=false','--network=none','--memory=2g', '--cpu-period=100000','--cpu-quota=200000','-t',tag,str(ctx)], stdout=log,stderr=subprocess.STDOUT,timeout=300,env=dict(os.environ,DOCKER_BUILDKIT='0')) if proc.returncode:raise RuntimeError('ARTIFACT_BUILD_FAILED') image=json.loads(command(['docker','image','inspect',tag]))[0] old=json.loads(command(['docker','image','inspect',base]))[0] if image['Config']!=old['Config'] or image['RootFS']['Layers'][:-1]!=old['RootFS']['Layers']: raise ValueError('PINNED_BASE_OR_CONFIG_CHANGED') name='ct-artifact-import-'+self.m['server'] try: command(['docker','run','--rm','--runtime=runc','--pull=never','--name',name,'--network=none', '--read-only','--memory=512m','--cpus=1','--pids-limit=64','--tmpfs','/tmp:size=16m', '-e','NVIDIA_VISIBLE_DEVICES=void','-e','PYTHONDONTWRITEBYTECODE=1', '-e','OPENBLAS_NUM_THREADS=1','--entrypoint','python3',image['Id'],'-c', 'import app.artifact_completion,app.artifact_retention,app.event_capture,app.event_artifact_service'],30) finally: check=subprocess.run(['docker','inspect',name],capture_output=True,timeout=10) if check.returncode==0 and json.loads(check.stdout)[0]['Image']==image['Id']: command(['docker','rm','-f',name],15) self.save('built',dict(image=image['Id'],tag=tag,base=base,previousImage=original)) print(json.dumps(load(self.root/'built.json'))) def indexes(self): _,rows=self.guard() tree=ast.parse((self.root/'payload/app/event_artifact_schema.py').read_text()) queries=[node.value for node in ast.walk(tree) if isinstance(node,ast.Constant) and isinstance(node.value,str) and node.value.startswith('CREATE INDEX IF NOT EXISTS contextra_artifact_cleanup_')] if len(queries)!=2:raise ValueError('EXPECTED_TWO_ADDITIVE_INDEXES') code="""import json,os,psycopg2 c=psycopg2.connect(os.environ['DATABASE_URL'],connect_timeout=3,options='-c statement_timeout=120000 -c lock_timeout=3000') c.autocommit=True try: with c.cursor() as q: for sql in QUERIES:q.execute(sql.replace('CREATE INDEX IF','CREATE INDEX CONCURRENTLY IF',1)) q.execute("SELECT indexrelid::regclass::text,indisvalid,indisready FROM pg_index WHERE indexrelid::regclass::text LIKE 'contextra_artifact_cleanup_%' ORDER BY 1") rows=q.fetchall() assert len(rows)==2 and all(v[1] and v[2] for v in rows) print(json.dumps(rows)) finally:c.close() """.replace('QUERIES',repr(queries)) self.save('indexes',json.loads(self.python(rows['analytics']['Id'],code,250))) print('Two additive cleanup indexes are ready.') def activate(self): plan,rows=self.guard() if not (self.root/'indexes.json').exists():raise ValueError('INDEX_READINESS_REQUIRED') if rows['ai-gateway']['State']['Running']: command(['docker','stop','--signal','SIGTERM','--timeout','30',rows['ai-gateway']['Id']],40) self.save('transition',dict(phase='GATEWAY_HELD',cameraStart=False)) if rows['analytics']['State']['Running']: out=command(['docker','exec','-i',rows['analytics']['Id'],'python3','-B','-'],240, (self.tools/'stop_channels.py').read_bytes()) self.fp.atomic(self.root/'stop-channels.log',out) status=self.runtime(rows['analytics']['Id']) if any(status['streamCounts'][k] for k in ('running','starting','error')): raise ValueError('CHANNEL_DRAIN_FAILED') self.drain_events(rows['analytics']['Id']) command(self.cb.compose_args(plan['active']['physical'])+['stop','--timeout','90','analytics'],120) self.finish_activation() def finish_activation(self): plan,rows=self.guard() if not (self.root/'transition.json').exists() or rows['ai-gateway']['State']['Running']: raise ValueError('GATEWAY_HOLD_REQUIRED') roles=['analytics'] if plan['before']['analytics']['running'] else [] self.check_stopped(rows,roles) active,_,_,approved,live,_=self.cb.inspect() if approved!=live:raise ValueError('COMPOSE_DRIFT') new=copy.deepcopy(approved);new['services']['analytics']['image']=load(self.root/'built.json')['image'] self.tx.mkdir(mode=0o700) for name,value in [('compose.json',new),('approved.json',new), ('compose.physical.json',self.s.physical(new)),('previous-active.json',active)]: self.fp.atomic(self.tx/name,value) from apply_source import normalize rendered=json.loads(command(self.cb.compose_args(self.tx/'compose.physical.json')+['config','--format','json'])) if normalize(self.s.logical(rendered))!=normalize(new):raise ValueError('COMPOSE_ROUNDTRIP') updated=dict(active,transaction=self.tx.name,compose=str(self.tx/'compose.json'), physical=str(self.tx/'compose.physical.json'),compose_semantic_sha256=self.cb.canonical(new)) self.save('activation',dict(before=active,after=updated)) if self.cb.active()!=active:raise ValueError('ACTIVE_CAS_FAILED') self.fp.atomic(self.state/'active.json',updated);self.cb.inspect();self.cb.guard_images(new) command(self.cb.compose_args(updated['physical'])+['up','--no-start','--no-deps','--no-build','--pull','never','analytics'],90) if roles:command(self.cb.compose_args(updated['physical'])+['start','analytics'],120) self.save('deployed',dict(active=updated,startedRoles=roles,gatewayHeld=plan['before']['ai-gateway']['running'])) print('Analytics replaced; Gateway held for readiness. No camera Start.') def verify(self): plan=load(self.root/'plan.json');deployed=load(self.root/'deployed.json') if self.cb.active()!=deployed['active']:raise ValueError('ACTIVE_DRIFT') self.cb.inspect();rows=self.rows();result={} for role in deployed['startedRoles']: until=time.monotonic()+120 while True: current=json.loads(command(['docker','inspect',rows[role]['Id']]))[0] if current['State'].get('Health',{}).get('Status')=='healthy':break if not current['State']['Running'] or time.monotonic()>until: raise ValueError('READINESS_FAILED '+role) time.sleep(2) for role in ('ai-control-sidecar','redis','postgresql','ai-gateway'): if rows[role]['Id']!=plan['before'][role]['id']:raise ValueError('UNRELATED_CONTAINER_CHANGED') for path,value in plan['protected'].items(): if sha(path)!=value:raise ValueError('PROTECTED_CONFIG_CHANGED') actual=json.loads(self.python(rows['analytics']['Id'], 'import hashlib,json\nfrom pathlib import Path\nnames='+repr(list(self.m['after']))+ '\nprint(json.dumps({n:hashlib.sha256((Path("/opt/contextra")/n).read_bytes()).hexdigest() for n in names}))')) if actual!=self.m['after']:raise ValueError('DEPLOYED_SOURCE_HASH_MISMATCH') result['sourceHashesVerified']=True result['runtimeBeforeGateway']=self.runtime(rows['analytics']['Id']) result['models']=json.loads(self.python(rows['analytics']['Id'],"""import json,urllib.request req=urllib.request.Request('http://127.0.0.1:8000/v2/repository/index',data=b'{}',headers={'Content-Type':'application/json'}) with urllib.request.urlopen(req,timeout=5) as r:rows=json.load(r) expected={'RFDETRMedium','RFDETRKeypointPreview','RFDETRFireSmoke','Awiros_person_and_head_detection'} assert expected.issubset({v['name'] for v in rows if v['state']=='READY'}),rows print(json.dumps(rows)) """)) if deployed['gatewayHeld']:command(['docker','start',rows['ai-gateway']['Id']],30) result['services']={k:dict(running=r['State']['Running'],image=r['Image']) for k,r in self.rows().items()} self.save('verified',result) print(json.dumps(dict(sourceHashesVerified=True,models=result['models'],services=result['services']))) if __name__=='__main__': root,expected,action=sys.argv[1:] if action not in ('prepare','build','indexes','activate','finish_activation','verify'): raise ValueError('UNSUPPORTED_ACTION') getattr(ArtifactInstaller(root,expected),action)()