#!/usr/bin/env python3 """Upload labeled synthetic performance samples with a durable local outbox (stdlib only).""" import argparse import json import math import os import re import sqlite3 import stat import sys import time from datetime import UTC, datetime from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError("Upload redirects are rejected to protect the source credential") def open_outbox(path, endpoint): path = Path(path) if not path.exists(): descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) os.close(descriptor) info = path.lstat() if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o077: raise ValueError("Outbox must be a private regular file (mode 0600)") database = sqlite3.connect(path) database.execute("pragma synchronous=FULL") database.execute("create table if not exists state " "(id integer primary key check(id=1),endpoint text not null,sequence integer)") database.execute("create table if not exists pending (sequence integer primary key,body text)") database.execute("insert or ignore into state values(1,?,0)", (endpoint,)) database.commit() if database.execute("select endpoint from state").fetchone()[0] != endpoint: database.close() raise ValueError("Outbox belongs to another source; use a separate file") return database def enqueue(database): with database: sequence = database.execute("select sequence from state where id=1").fetchone()[0] + 1 sample = {"sequence": sequence, "observedAt": datetime.now(UTC).isoformat(), "values": {"cpu": round(40 + 20 * math.sin(sequence / 8), 2), "memory": round(55 + 10 * math.sin(sequence / 17), 2)}, "quality": "good"} database.execute("insert into pending values(?,?)", (sequence, json.dumps(sample))) database.execute("update state set sequence=? where id=1", (sequence,)) def upload(database, endpoint, token, opener): rows = database.execute("select sequence,body from pending order by sequence limit 50").fetchall() if not rows: return 0 body = json.dumps({"schemaVersion": "odriva.source.v1", "synthetic": True, "samples": [json.loads(row[1]) for row in rows]}).encode() request = Request(endpoint, data=body, headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"}, method="POST") with opener.open(request, timeout=10) as response: if response.status != 202: raise ValueError("Unexpected ingest status; pending data retained") reply = response.read(65537) if len(reply) > 65536: raise ValueError("Oversized acknowledgment; pending data retained") ack = json.loads(reply) expected = [row[0] for row in rows] if not isinstance(ack, dict): raise ValueError("Invalid durable acknowledgment; pending data retained") acknowledged = ack.get("acknowledged", []) if not isinstance(acknowledged, list) or not all( isinstance(entry, dict) and type(entry.get("sequence")) is int for entry in acknowledged ): raise ValueError("Invalid durable acknowledgment; pending data retained") if (ack.get("sourceId") != urlsplit(endpoint).path.split("/")[-2] or ack.get("accepted") != len(expected) or [entry.get("sequence") for entry in acknowledged] != expected or any(entry.get("disposition") not in {"INSERTED", "DUPLICATE"} for entry in acknowledged)): raise ValueError("Invalid durable acknowledgment; pending data retained") with database: database.executemany("delete from pending where sequence=?", [(seq,) for seq in expected]) return len(rows) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--endpoint", required=True) parser.add_argument("--outbox", type=Path, default=Path("odriva-source-outbox.db")) parser.add_argument("--duration", type=float, default=900) parser.add_argument("--interval", type=float, default=5) parser.add_argument("--drain-timeout", type=float, default=180) parser.add_argument("--allow-local-http", action="store_true") args = parser.parse_args() parts = urlsplit(args.endpoint) allowed_http = (args.allow_local_http and parts.scheme == "http" and parts.hostname in {"localhost", "127.0.0.1", "::1"}) if (not (parts.scheme == "https" or allowed_http) or not parts.hostname or parts.username or parts.password or parts.query or parts.fragment or not re.fullmatch(r"/api/odriva/v1/sources/[0-9a-f-]{36}/events", parts.path)): parser.error("Use the source HTTPS ingest URL; HTTP is allowed only explicitly on loopback") if args.duration < 0 or args.interval <= 0 or args.drain_timeout <= 0: parser.error("Duration must be nonnegative; interval and drain timeout must be positive") token = os.environ.get("ODRIVA_SOURCE_TOKEN", "") if not 40 <= len(token) <= 128 or any(not 33 <= ord(char) <= 126 for char in token): parser.error("Set ODRIVA_SOURCE_TOKEN to this source's credential (never pass it in argv)") opener = build_opener(NoRedirect()) with open_outbox(args.outbox, args.endpoint) as database: started = time.monotonic() end = started + args.duration sample_at, upload_at, failures = started, started, 0 sent = 0 while True: now = time.monotonic() if now < end and now >= sample_at: enqueue(database) sample_at = now + args.interval pending = database.execute("select count(*) from pending").fetchone()[0] if now >= end and not pending: print(json.dumps({"pending": 0, "acknowledged": sent, "synthetic": True, "elapsedSeconds": round(now-started, 2)})) break if now >= end + args.drain_timeout: raise ValueError("Drain timed out; pending samples are retained for retry") if pending and now >= upload_at: try: sent += upload(database, args.endpoint, token, opener) failures = 0 upload_at = time.monotonic() except HTTPError as error: if error.code not in {408, 429, 500, 502, 503, 504}: raise ValueError(f"Ingest rejected HTTP {error.code}; pending data retained") failures += 1 retry = error.headers.get("Retry-After", "") upload_at = time.monotonic() + (min(int(retry), 60) if retry.isdigit() else min(2 ** min(failures, 5), 30)) except (URLError, TimeoutError): failures += 1 upload_at = time.monotonic() + min(2 ** min(failures, 5), 30) time.sleep(min(args.interval, 0.25)) if __name__ == "__main__": try: main() except (ValueError, OSError, sqlite3.Error) as error: # Only our fixed messages; transport/SQLite errors may include private paths or URLs. print(str(error) if isinstance(error, ValueError) else "Upload failed; outbox retained", file=sys.stderr) sys.exit(1)