from ops import * import yaml,signal,sys BASE=HERE if len(sys.argv)>1: HERE=BASE/sys.argv[1];HERE.mkdir(mode=0o700) def save(name,data,private=False): with ((PRIVATE if private else HERE)/name).open('x') as f:json.dump(data,f,indent=2) ORDER=[3,5,6,8,9,10,11,12,13,14,15,16,17,18,19,20,21] LEVELS=[1,2,4,8,16,17] PROBE='/opt/contextra/tools/provider_scale_probe.py' original=yaml.safe_load((PRIVATE/'runtime-original.yaml').read_text())['added'] specs={c:next(s for s in original if str(s.get('camera_id'))==str(c)) for c in ORDER} sids={c:specs[c]['stream_id'] for c in ORDER};reverse={sid:c for c,sid in sids.items()} FIELDS=('intrusion','loitering','fall','fight','fire','crowd','tracking') if (PRIVATE/'settings-original.json').exists(): saved={int(k):v for k,v in json.loads((PRIVATE/'settings-original.json').read_text()).items()} else: saved={c:api('/api/v1/streams/'+sids[c]+'/settings')['configured'] for c in ORDER} save('settings-original.json',saved,private=True) starts={};touched=[];launched=[];stages=[];begin=time.monotonic();initial_oom=None;control_pid=None def event(name,**value): row=dict(event=name,wall_ms=time.time_ns()//1000000,t=time.monotonic());row.update(value) with (HERE/'journal.jsonl').open('a') as f:f.write(json.dumps(row)+'\n') def interrupt(signum,frame):raise RuntimeError('SIGNAL_'+str(signum)) signal.signal(signal.SIGTERM,interrupt);signal.signal(signal.SIGINT,interrupt) def patch_settings(c,changes): current=api('/api/v1/streams/'+sids[c]+'/settings')['configured'] return request('/api/v1/streams/'+sids[c]+'/settings',dict(expected_revision=current['revision'],**changes),method='PATCH') def rows(): source=api('/api/v1/streams')['streams'];out={} for r in source: if r['stream_id'] not in reverse:continue c=reverse[r['stream_id']] if c not in launched:continue out[str(c)]={k:r.get(k) for k in ('state','run_id','run_generation','last_error','cleanup_error','cleanup_pending','degraded','scenarios','publisher_metrics')} out[str(c)]['providers']={p:{k:v.get(k) for k in ('state','child_pid','last_error','cleanup_error','last_cleanup','metadata_path','publisher_metrics')} for p,v in r.get('provider_processes',{}).items()} return out def basic_safety(current): if time.monotonic()-begin>1000:raise RuntimeError('TOTAL_TIME_BOUND') if shutil.disk_usage('/').free<15*1024**3:raise RuntimeError('ROOT_FREE_BOUND') available=int(next(l for l in Path('/proc/meminfo').read_text().splitlines() if l.startswith('MemAvailable:')).split()[1])*1024 if available<2*1024**3:raise RuntimeError('HOST_RAM_FREE_BOUND') for c,r in current.items(): if r['last_error'] or r['cleanup_error'] or r['degraded']:raise RuntimeError('CHANNEL_ERROR_'+c) for p,v in r['providers'].items(): if v['last_error'] or v['cleanup_error'] or v['state']=='error':raise RuntimeError('PROVIDER_ERROR_'+c+'_'+p) def sample(): global initial_oom,control_pid started=time.monotonic();r=rows() mapping={str(v['child_pid']):c+':'+p for c,s in r.items() for p,v in s['providers'].items() if v.get('child_pid')} if control_pid is None and mapping: pid=next(iter(mapping)) code="from pathlib import Path;import sys;print(next(s.split()[1] for s in (Path('/proc')/sys.argv[1]/'status').read_text().splitlines() if s.startswith('PPid:')))" control_pid=cmd(['docker','exec','contextra-analytics-1','python3','-B','-c',code,pid],timeout=5).strip() if control_pid:mapping[control_pid]='channel-control' resources=json.loads(cmd(['docker','exec','-i','contextra-analytics-1','python3','-B',PROBE,'resources'],input=json.dumps(mapping),timeout=15)) a=api('/api/v1/metadata/status')['artifacts'] gpu=cmd(['nvidia-smi','--query-gpu=utilization.gpu,utilization.memory,memory.used,power.draw,clocks.sm,clocks.mem','--format=csv,noheader,nounits'],timeout=6).strip() value=dict(t=started,finished=time.monotonic(),wall_ms=time.time_ns()//1000000,channels=r,resources=resources,gpu=gpu, artifacts={k:a.get(k) for k in ('rendering','active','activity','counts','peaks','error','storage_warning')}) oom=dict(l.split() for l in resources['cgroup'].get('memory.events','').splitlines()) if initial_oom is None:initial_oom=oom if any(int(oom.get(k,0))>int(initial_oom.get(k,0)) for k in ('oom','oom_kill')): save('oom-bound.json',value);raise RuntimeError('CGROUP_OOM') return value def model_config(): return json.loads(cmd(['docker','exec','contextra-analytics-1','python3','-B',PROBE,'models'],timeout=15)) def frame_probe(cameras): path=Path('/workspace/contextra-runtime-config/metadata-active/reader/reader-credentials.json') assert not path.is_symlink() and path.stat().st_mode&0o077==0 return json.loads(cmd(['docker','exec','-i','contextra-analytics-1','python3','-B',PROBE,'frames','--cameras',','.join(map(str,cameras))],input=path.read_text(),timeout=15)) assert not inspect('contextra-ai-gateway-1')['State']['Running'] result=dict(stages=[],starts=starts,errors=[],cleanup=None) try: configs=model_config();save('model-configs.json',configs) conditions=[] for c in ORDER: desired={k:dict(enabled=True) for k in ('intrusion','loitering','fall','fight','crowd')} desired.update(fire=dict(enabled=True,scenario=dict(enabled=True)),tracking=dict(enabled=True)) touched.append(c);v=patch_settings(c,desired)['configured'] assert [p['key'] for p in v['execution_plan']['providers']]==['pose','fire','crowd'] assert v['analysisFpsTarget']==10 conditions.append(dict(camera=c,source_sha256=hashlib.sha256(specs[c]['uri'].encode()).hexdigest(), source_options_sha256=hashlib.sha256(json.dumps(specs[c].get('source_options',{}),sort_keys=True).encode()).hexdigest(), original_revision=saved[c]['revision'],trial_revision=v['revision'],fps=v['analysisFpsTarget'], scenarios={k:v[k] for k in FIELDS},pose=v['pose'],display_score=v['display_score'],display_categories=v['display_categories'])) save('conditions.json',dict(order=ORDER,levels=LEVELS,conditions=conditions, resource_bounds=dict(root_free_gib=15,host_available_ram_gib=2),input_expectations=dict(fire=[5,6],fall=[3,8,9,10,11,12,16,21],fight=[13,14,15,17,18,19,20]))) for level in LEVELS: stage=dict(level=level,started_wall_ms=time.time_ns()//1000000,samples=[],startup=[],metric_errors=[],status='RUNNING') result['stages'].append(stage);event('stage_begin',level=level) for c in ORDER[:level]: if c in launched:continue basic_safety(rows()) launched.append(c);sent=time.monotonic();wall=time.time_ns()//1000000 response=request('/api/v1/streams/'+sids[c]+'/start',{},timeout=20) starts[c]=dict(sent=sent,wall_ms=wall,ack_at=time.monotonic(),request_state=response.get('state')) event('start_ack',camera=c,**starts[c]) deadline=time.monotonic()+70 while True: current=rows();basic_safety(current);r=current[str(c)] ready=len(r['providers'])==3 and all(v.get('metadata_path',{}).get('providers',{}).get(p,{}).get('responses',0)>=2 for p,v in r['providers'].items()) if ready: first={p:v['metadata_path']['providers'][p].get('first_response_at') for p,v in r['providers'].items()} starts[c]['first_response_at']=first;starts[c]['ready_observed_at']=time.monotonic();break if time.monotonic()>deadline: save('startup-failure-'+str(c)+'.json',current);raise RuntimeError('NO_PROVIDER_RESPONSE_'+str(c)) time.sleep(1) stage['startup'].append(dict(camera=c,**starts[c])) time.sleep(5) current=rows();basic_safety(current) for c in ORDER[:level]: try: request('/api/v1/streams/'+sids[c]+'/metrics',dict(run_generation=current[str(c)]['run_generation'],enabled=True,seconds=60),timeout=15) except Exception as e:stage['metric_errors'].append(dict(camera=c,error=type(e).__name__)) start=time.monotonic() for i in range(7): time.sleep(max(0,start+5*i-time.monotonic())) value=sample();stage['samples'].append(value);basic_safety(value['channels']) event('sample',level=level,index=i,channels=len(value['channels']),sample_seconds=round(value['finished']-value['t'],3),gpu=value['gpu']) stage['frames']=frame_probe(ORDER[:level]) assert configs==model_config() stage['status']='PASS';stage['finished_wall_ms']=time.time_ns()//1000000 save('stage-'+str(level)+'.json',stage) event('stage_complete',level=level) print(json.dumps(dict(level=level,status='PASS',children=sum(len(r['providers']) for r in stage['samples'][-1]['channels'].values()))),flush=True) except Exception as e: result['errors'].append(dict(type=type(e).__name__,reason=str(e)[:200])) event('trial_error',**result['errors'][-1]) try:save('failure-state.json',rows()) except Exception:pass finally: for c in launched: try:request('/api/v1/streams/'+sids[c]+'/stop',{},timeout=20) except Exception as e:result['errors'].append(dict(type=type(e).__name__,reason='STOP_REQUEST_'+str(c))) deadline=time.monotonic()+90 while True: r=rows() reaped=all(not v['cleanup_pending'] and all(not p.get('child_pid') for p in v['providers'].values()) for v in r.values()) if reaped or time.monotonic()>deadline:break time.sleep(1) result['cleanup']=dict(reaped=reaped,states=r) if reaped: for c in touched: try: changes={k:saved[c][k] for k in FIELDS} if 'scenario' not in changes['fire']: changes['fire']=dict(changes['fire'],scenario={'enabled':False}) patch_settings(c,changes) except Exception as e:result['errors'].append(dict(type=type(e).__name__,reason='RESTORE_SETTINGS_'+str(c))) result['finished_wall_ms']=time.time_ns()//1000000 save('trial-result.json',result) print(json.dumps(dict(completed=[s['level'] for s in result['stages'] if s['status']=='PASS'],errors=result['errors'],reaped=reaped)),flush=True) if result['errors'] or not reaped:raise SystemExit(1)