"""Standalone standard-library client. Download and import as oopps_ai_client. Only usage metadata is serialized. SQLite is a durable delivery spool; HTTP failures retain events. Call flush in a background job, outside inference. """ import argparse import csv import io import json import os import sqlite3 import subprocess import time import urllib.request from datetime import datetime, timezone from urllib.parse import urlparse class Client: def __init__(self, base_url, token, spool='oopps-ai-usage.sqlite3'): parsed = urlparse(base_url) if parsed.scheme != 'https' and not (parsed.scheme == 'http' and parsed.hostname in {'localhost', '127.0.0.1'}): raise ValueError('Use HTTPS for AI FinOps delivery') self.base_url = base_url.rstrip('/') self.token = token self.spool = spool with sqlite3.connect(spool) as db: db.execute('CREATE TABLE IF NOT EXISTS pending (id INTEGER PRIMARY KEY AUTOINCREMENT, path TEXT NOT NULL, payload TEXT NOT NULL)') os.chmod(spool, 0o600) def enqueue(self, path, payload): if path not in {'/api/ai/ingest/usage'} and not path.startswith('/api/ai/ingest/deployments/'): raise ValueError('Unsupported accounting endpoint') with sqlite3.connect(self.spool) as db: db.execute('INSERT INTO pending(path,payload) VALUES(?,?)', (path,json.dumps(payload))) def record_chat(self, completion, *, started_at, event_key=None, operation_id=''): """Call for each actual attempt. For streaming pass the final usage chunk. Disable opaque SDK retries or instrument each attempt separately: a logical completion cannot reveal usage of hidden provider attempts. """ def get(obj, key, default=None): return obj.get(key,default) if isinstance(obj,dict) else getattr(obj,key,default) usage = get(completion,'usage') if usage is None: raise ValueError('Final usage is missing; cannot invent billed token counts') key = event_key or get(completion,'id') model = get(completion,'model') if not key or not model: raise ValueError('Actual attempt ID and served model are required') details = get(usage,'prompt_tokens_details') or {} self.enqueue('/api/ai/ingest/usage', {'events': [{ 'event_key': key,'model': model,'operation_id': operation_id, 'started_at': started_at.isoformat(),'finished_at': datetime.now(timezone.utc).isoformat(), 'input_tokens': get(usage,'prompt_tokens'),'cached_tokens': get(details,'cached_tokens',0) or 0, 'output_tokens': get(usage,'completion_tokens'),'outcome': 'success','complete': True, }]}) def flush(self, limit=500): """Send bounded batches. A crash after HTTP success is safe to replay.""" sent = 0 # Never hold a SQLite write transaction during HTTP: inference producers # must be able to enqueue while a slow network request is in flight. # Concurrent flushers may replay the same fact; server deduplication owns it. with sqlite3.connect(self.spool, timeout=30) as db: rows = db.execute('SELECT id,path,payload FROM pending ORDER BY id LIMIT ?', (limit,)).fetchall() position = 0 while position < len(rows): current = rows[position]; group = [current]; payload = json.loads(current[2]) if current[1] == '/api/ai/ingest/usage': while position+len(group)500: break payload['events'].extend(next_payload['events']); group.append(next_row) request = urllib.request.Request(self.base_url+current[1],data=json.dumps(payload).encode(), headers={'Content-Type':'application/json','Authorization':'Bearer '+self.token},method='POST') with urllib.request.urlopen(request,timeout=15) as result: if result.status != 200: raise RuntimeError('Usage delivery was not acknowledged') with sqlite3.connect(self.spool, timeout=30) as db: db.executemany('DELETE FROM pending WHERE id=? AND path=? AND payload=?', group) sent += len(group); position += len(group) return sent def read_gpus(uuids): result = subprocess.run(['nvidia-smi','--query-gpu=uuid,utilization.gpu,memory.used','--format=csv,noheader,nounits'], capture_output=True,text=True,check=True,timeout=10) rows = {row[0].strip(): row for row in csv.reader(io.StringIO(result.stdout))} if not set(uuids).issubset(rows): raise RuntimeError('Some configured physical GPUs were not found') selected = [rows[key] for key in uuids] return {'gpu_uuids': sorted(uuids), 'gpu_utilization': sum(float(r[1]) for r in selected)/len(selected), 'memory_used_bytes': sum(int(float(r[2]))*1024*1024 for r in selected)} def main(): parser=argparse.ArgumentParser(description='Oopps AI usage delivery / dedicated-GPU interval collector') parser.add_argument('command',choices=['flush','gpu']) parser.add_argument('--spool',default='oopps-ai-usage.sqlite3') parser.add_argument('--deployment',type=int) parser.add_argument('--gpu-uuids',help='Comma-separated physical UUIDs exclusively assigned to this deployment') parser.add_argument('--interval',type=int,default=60) args=parser.parse_args() client=Client(os.environ['OOPPS_URL'],os.environ['OOPPS_AI_TOKEN'],args.spool) if args.command=='flush': print(f'Delivered spool records: {client.flush()}'); return uuids=(args.gpu_uuids or '').split(',') if not args.deployment or not all(u.startswith('GPU-') for u in uuids) or not 10<=args.interval<=3600: parser.error('GPU collection requires --deployment, physical --gpu-uuids and an interval of 10..3600 seconds') read_gpus(uuids) start=datetime.now(timezone.utc) while True: time.sleep(args.interval) end=datetime.now(timezone.utc) metrics=read_gpus(uuids) client.enqueue(f'/api/ai/ingest/deployments/{args.deployment}/intervals',{ 'event_key': f'{args.deployment}:{start.isoformat()}', 'started_at': start.isoformat(),'finished_at': end.isoformat(), **metrics,'usage_complete': False, }) start=end try: client.flush() except Exception as exc: print(f'Delivery deferred ({type(exc).__name__}); source facts remain in spool',flush=True) if __name__=='__main__': main()