#!/usr/bin/env python3 """ Production-ready asynchronous orchestrator. - Uses a SQLite DB (WAL) for jobs & metrics - Parses env files (static) for candidates (no 'source') - Communicates with local adapter via UDS or localhost token - Dry-run default, live mode requires enabling via enable_live.sh """ import os, json, asyncio, time, sqlite3, logging, hashlib from pathlib import Path import httpx from prometheus_client import start_http_server, Counter, Gauge from filelock import FileLock BASE = Path(__file__).resolve().parent.parent CFG = json.load(open(BASE / 'orchestrator' / 'config.json')) DB_PATH = Path(CFG['db_path']).resolve() LOG_FILE = Path(CFG['log_file']).resolve() ADAPTER_SOCKET = Path(CFG['adapter_socket']).resolve() ADAPTER_TOKEN_FILE = Path(CFG['adapter_token_file']).resolve() # logging logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s', handlers=[logging.FileHandler(LOG_FILE), logging.StreamHandler()]) # metrics JOBS_PROCESSED = Counter('hypothalumus_jobs_processed_total', 'Total jobs processed') JOB_FAILURES = Counter('hypothalumus_job_failures_total', 'Job failures') ACTIVE_JOBS = Gauge('hypothalumus_active_jobs', 'Currently active jobs') # DB helper def init_db(): conn = sqlite3.connect(DB_PATH, timeout=30, isolation_level=None) conn.execute('PRAGMA journal_mode=WAL;') c = conn.cursor() c.execute(''' CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, var TEXT, url TEXT, created_at INTEGER, status TEXT, attempts INTEGER, last_response TEXT, created_src TEXT ) ''') c.execute(''' CREATE TABLE IF NOT EXISTS metrics ( url TEXT PRIMARY KEY, success INTEGER, failure INTEGER, ewma_latency REAL, last_failure INTEGER, open_until INTEGER ) ''') conn.commit() return conn # safe env parsing (static): read wrappers or desktop envs for *_API_ entries def load_candidates(env_dir): candidates = {} for f in Path(env_dir).glob('*.env'): for line in f.read_text(encoding='utf8').splitlines(): line=line.strip() if not line or line.startswith('#'): continue if line.lower().startswith('export '): line = line[7:].strip() if '=' not in line: continue key,val=line.split('=',1) key=key.strip(); val=val.strip().strip('\'"') if '_API_' in key: prefix=key.split('_API_')[0].lower() candidates.setdefault(prefix,[]).append((key,val)) return candidates # insert job to DB (idempotent) def insert_job(conn, jid, var, url, src): ts=int(time.time()) with conn: conn.execute('INSERT OR IGNORE INTO jobs (id,var,url,created_at,status,attempts,created_src) VALUES (?,?,?,?,?,?,?)', (jid,var,url,ts,'pending',0,src)) # simple circuit breaker helpers def is_open(conn, url): r = conn.execute('SELECT open_until FROM metrics WHERE url=?', (url,)).fetchone() if not r: return False return (r[0] or 0) > int(time.time()) def mark_success(conn, url, latency): row = conn.execute('SELECT success,failure,ewma_latency FROM metrics WHERE url=?',(url,)).fetchone() if row: s,f,ew=row s=s+1 ew = (0.2*latency) + (0.8*(ew or latency)) conn.execute('UPDATE metrics SET success=?, ewma_latency=? WHERE url=?',(s,ew,url)) else: conn.execute('INSERT INTO metrics(url,success,failure,ewma_latency,last_failure,open_until) VALUES (?,?,?,?,?,?)', (url,1,0,latency, None, 0)) def mark_failure(conn, url): row = conn.execute('SELECT success,failure FROM metrics WHERE url=?',(url,)).fetchone() if row: s,f=row f=f+1 open_until = int(time.time()) + CFG['circuit_open_seconds'] if f >= CFG['failures_to_open'] else 0 conn.execute('UPDATE metrics SET failure=?, last_failure=?, open_until=? WHERE url=?',(f,int(time.time()),open_until,url)) else: conn.execute('INSERT INTO metrics(url,success,failure,ewma_latency,last_failure,open_until) VALUES (?,?,?,?,?,?)', (url,0,1,None,int(time.time()),0)) # perform the probe / action async def call_url(client, url): try: r = await client.get(url, timeout=CFG['http_timeout']) return True, r.status_code, r.elapsed.total_seconds(), r.text[:1000] except Exception as e: return False, None, None, str(e) # call adapter (for privileged actions) via HTTP-to-UDS or direct HTTP with token async def call_adapter(action, payload): # Adapter expects JSON: {"action":"restart_network","params":{},"id":"..."} token = ADAPTER_TOKEN_FILE.read_text().strip() if ADAPTER_TOKEN_FILE.exists() else None headers = {} if token: headers['X-Adapter-Token'] = token # adapter may be on UDS; but httpx needs http://localhost + transport url = f"http://unix:{ADAPTER_SOCKET}:/{action}" async with httpx.AsyncClient() as client: try: r = await client.post(url, json=payload, headers=headers, timeout=CFG['http_timeout']) return True, r.status_code, r.text[:2000] except Exception as e: return False, None, str(e) async def process_job(conn, job): jid, var, url = job logging.info(f"PROCESS JOB {jid} -> {url}") ACTIVE_JOBS.inc() try: if is_open(conn, url): logging.warning("Circuit open for %s; skipping", url) return async with httpx.AsyncClient() as client: ok, status, latency, body = await call_url(client, url) if ok and (status is None or status < 400): logging.info("Probe ok for %s", url) mark_success(conn, url, latency or 0.1) conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', ('ok', json.dumps({'status':status,'lat':latency}), jid)) JOBS_PROCESSED.inc() else: logging.warning("Probe failed for %s : %s", url, body) mark_failure(conn, url) # If not dry_run, issue adapter or remote action depending on var prefix rule if not CFG.get('dry_run', True): # Example adaptability: if var prefix 'network' then call adapter restart prefix = var.split('_API_')[0].lower() if prefix in ('network','device','chrome','safari'): action = prefix + "_repair" payload = {"url":url,"var":var,"job_id":jid} ok2, code2, resp2 = await call_adapter(action, payload) if ok2 and (code2 is None or int(code2) < 400): conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', ('actioned', json.dumps({'adapter':resp2}), jid)) else: conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', ('failed', json.dumps({'error':resp2}), jid)) JOB_FAILURES.inc() else: conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', ('failed', json.dumps({'probe_error':body}), jid)) JOB_FAILURES.inc() else: # dry-run: just mark dry-run-failed conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', ('dryrun-failed', json.dumps({'probe':body}), jid)) JOB_FAILURES.inc() finally: ACTIVE_JOBS.dec() async def worker_loop(): conn = sqlite3.connect(DB_PATH, timeout=30, isolation_level=None) while True: rows = conn.execute("SELECT id,var,url FROM jobs WHERE status IN ('pending','dryrun-failed') ORDER BY created_at LIMIT 20").fetchall() if not rows: await asyncio.sleep(CFG['jobs_poll_seconds']) continue tasks=[] sem = asyncio.Semaphore(CFG['max_parallel']) async with httpx.AsyncClient() as client: loop = asyncio.get_event_loop() for r in rows: jid,var,url = r # run process_job in task t = loop.create_task(process_job(conn, (jid,var,url))) tasks.append(t) if tasks: await asyncio.gather(*tasks) await asyncio.sleep(0.1) def ensure_db_jobs_from_env(conn): # parse env dir wrappers for *_API_* and insert into jobs table if missing env_dir = BASE / 'env' for f in env_dir.glob('*.env'): for line in f.read_text(encoding='utf8').splitlines(): l=line.strip() if not l or l.startswith('#'): continue if l.lower().startswith('export '): l = l[7:].strip() if '=' not in l: continue key,val = l.split('=',1) if '_API_' in key: key=key.strip() val=val.strip().strip('\'"') jid = hashlib.sha1((key+val).encode()).hexdigest()[:20] insert_job(conn, jid, key, val, f.name) def main(): DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = init_db() ensure_db_jobs_from_env(conn) # start prometheus metrics try: start_http_server(CFG.get('prometheus_port',9001)) except Exception as e: logging.warning("Prometheus port may already be in use: %s", e) # main loop asyncio.run(worker_loop()) if __name__ == '__main__': main()