#!/usr/bin/env python3 """LockedIn Labs: real local HTTP context propagation with the pinned OTel SDK. Install fde-first-trace-requirements.txt. Both endpoints are owned loopback listeners in this process. No collector, model, private data or remote target. """ import argparse from contextlib import ExitStack, contextmanager from http.client import HTTPConnection from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from importlib.metadata import version import json import platform import threading try: from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from opentelemetry.sdk.trace.sampling import ALWAYS_ON, ParentBased from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator except ImportError: raise SystemExit("Install the pinned fde-first-trace-requirements.txt in a virtual environment.") SCHEMA = "lockedin.http-trace-lab/v1" CASES = ("correct", "unsampled", "dropped", "reused-parent", "invalid-zero", "missing", "baggage") TRACE_ID = "1234567890abcdef1234567890abcdef" PARENT_ID = "1234567890abcdef" EXPECTED = {"opentelemetry-api": "1.45.1", "opentelemetry-sdk": "1.45.1", "opentelemetry-semantic-conventions": "0.66b1", "typing_extensions": "4.16.0"} PROPAGATOR = TraceContextTextMapPropagator() def identity(context): return {"valid": context.is_valid, "trace_id": f"{context.trace_id:032x}", "span_id": f"{context.span_id:016x}", "sampled": context.trace_flags.sampled, "remote": context.is_remote} def carrier(headers): return {key: headers[key] for key in ("traceparent", "tracestate", "baggage") if key in headers} def request(port, headers): """The only client target is a listener allocated by this fixture.""" client = HTTPConnection("127.0.0.1", port, timeout=3) try: client.request("GET", "/synthetic-task", headers=headers) response = client.getresponse() return response.status, response.read(65536).decode("utf-8") finally: client.close() @contextmanager def listener(handler): server = ThreadingHTTPServer(("127.0.0.1", 0), handler) thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}) thread.start() try: yield server.server_port finally: server.shutdown() server.server_close() thread.join(timeout=4) if thread.is_alive(): raise RuntimeError("owned_listener_did_not_stop") class TeachingHandler(BaseHTTPRequestHandler): def log_message(self, *_): pass def reply(self, status, value): body = json.dumps(value).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def run_case(name, sampled=1): if name not in CASES or sampled not in (0, 1): raise ValueError("unsupported_case_or_sampled_flag") exporter = InMemorySpanExporter() providers = [TracerProvider(sampler=ParentBased(ALWAYS_ON), resource=Resource.create({"service.name": service})) for service in ("synthetic-gateway", "synthetic-worker")] for provider in providers: provider.add_span_processor(SimpleSpanProcessor(exporter)) gateway_tracer, worker_tracer = [provider.get_tracer("lockedin.http-trace-lab") for provider in providers] observations, failures = {}, [] counts, count_lock = {"gateway": 0, "worker": 0}, threading.Lock() def observe(handler, tracer, role): with count_lock: counts[role] += 1 incoming = carrier(handler.headers) extracted = PROPAGATOR.extract(incoming) parent = trace.get_current_span(extracted).get_span_context() return incoming, extracted, parent, tracer.start_as_current_span(role + ".request", context=extracted, kind=trace.SpanKind.SERVER) class Worker(TeachingHandler): def do_GET(self): self.connection.settimeout(3) if self.path != "/synthetic-task": self.reply(404, {"error": "unknown_synthetic_path"}) return try: incoming, extracted, parent, span_scope = observe(self, worker_tracer, "worker") with span_scope as span: observations["worker"] = {"received": incoming, "parent": identity(parent), "span": identity(span.get_span_context()), "recording": span.is_recording(), "handler_entries": counts["worker"], "result": "SYNTHETIC-OK"} self.reply(200, {"result": "SYNTHETIC-OK"}) except Exception as error: failures.append(type(error).__name__) self.reply(500, {"error": "fixture_worker_failed"}) try: with ExitStack() as stack: worker_port = stack.enter_context(listener(Worker)) class Gateway(TeachingHandler): def do_GET(self): self.connection.settimeout(3) if self.path != "/synthetic-task": self.reply(404, {"error": "unknown_synthetic_path"}) return try: incoming, extracted, parent, span_scope = observe(self, gateway_tracer, "gateway") with span_scope as span: outgoing = {} PROPAGATOR.inject(outgoing) if name == "dropped": outgoing = {} elif name == "reused-parent": outgoing = {"traceparent": incoming["traceparent"]} observations["gateway"] = {"received": incoming, "parent": identity(parent), "span": identity(span.get_span_context()), "recording": span.is_recording(), "handler_entries": counts["gateway"], "sent": outgoing} status, body = request(worker_port, outgoing) observations["gateway"]["downstream_status"] = status self.reply(status, json.loads(body)) except Exception as error: failures.append(type(error).__name__) self.reply(500, {"error": "fixture_gateway_failed"}) gateway_port = stack.enter_context(listener(Gateway)) flag = 0 if name == "unsampled" else sampled incoming = {"traceparent": f"00-{TRACE_ID}-{PARENT_ID}-{flag:02x}"} if name == "invalid-zero": incoming["traceparent"] = f"00-{'0' * 32}-{PARENT_ID}-01" elif name == "missing": incoming = {} elif name == "baggage": incoming["baggage"] = "synthetic.department=practice" status, body = request(gateway_port, incoming) if failures or status != 200 or len(observations) != 2: raise RuntimeError("fixture_request_failed:" + ",".join(failures)) spans = [{"name": span.name, "service": span.resource.attributes["service.name"], "context": identity(span.context), "parent": identity(span.parent) if span.parent else None} for span in exporter.get_finished_spans()] a, b = observations["gateway"], observations["worker"] return {"case": name, "client_status": status, "client_result": json.loads(body), **observations, "handler_counts": dict(counts), "finished_spans": spans, "comparison": {"same_trace": a["span"]["trace_id"] == b["span"]["trace_id"], "worker_parent_is_gateway": b["parent"]["valid"] and b["parent"]["span_id"] == a["span"]["span_id"], "exported_span_count": len(spans)}, "owned_listeners_stopped": True} finally: for provider in providers: provider.shutdown() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--case", choices=CASES) parser.add_argument("--sampled", type=int, choices=(0, 1), default=1, help="Incoming sampled bit, except fixed unsampled and invalid-zero cases.") args = parser.parse_args() installed = {package: version(package) for package in EXPECTED} if installed != EXPECTED: parser.error("Use the exact pinned fde-first-trace-requirements.txt environment.") reports = [run_case(name, args.sampled) for name in ([args.case] if args.case else CASES)] print(json.dumps({"schema": SCHEMA, "python": platform.python_version(), "packages": installed, "scope": "Real loopback HTTP and SDK contexts in one process; separate providers, in-memory export. No collector, TLS, DNS capture, remote hosts, model, authentication or durable business effect.", "cases": reports}, indent=2)) if __name__ == "__main__": main()