链路追踪连接一次请求中已被记录的操作

import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from opentelemetry.context import Context
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.trace import SpanKind, Status, StatusCode
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

exporter = InMemorySpanExporter()
providers = [TracerProvider(resource=Resource({"service.name": name}))
             for name in ("caller", "worker")]
for provider in providers:
    provider.add_span_processor(SimpleSpanProcessor(exporter))
caller, worker = [p.get_tracer("trace-lab") for p in providers]
propagator = TraceContextTextMapPropagator()


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_GET(self):
        carrier = {key.lower(): value for key, value in self.headers.items()}
        parent = propagator.extract(carrier, context=Context())
        with worker.start_as_current_span("handle" + self.path,
                                         context=parent, kind=SpanKind.SERVER) as span:
            status = 200
            if self.path == "/fail":
                try:
                    raise ValueError("synthetic inventory failure")
                except ValueError as error:
                    span.record_exception(error)
                    span.set_status(Status(StatusCode.ERROR, str(error)))
                    status = 503
            body = b"lab response\n"
            self.send_response(status)
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)


server = HTTPServer(("127.0.0.1", 0), Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
base = f"http://127.0.0.1:{server.server_port}"
try:
    with caller.start_as_current_span("request", context=Context()):
        with caller.start_as_current_span("outbound", kind=SpanKind.CLIENT):
            headers = {}
            propagator.inject(headers)
            with urlopen(Request(base + "/ok", headers=headers), timeout=5) as response:
                assert response.status == 200
                response.read()
    try:
        with urlopen(base + "/fail", timeout=5) as response:
            raise AssertionError("Expected HTTP 503")
    except HTTPError as response:
        with response:
            assert response.code == 503
            response.read()
finally:
    server.shutdown()
    thread.join(timeout=5)
    server.server_close()

for provider in providers:
    assert provider.force_flush()
spans = {span.name: span for span in exporter.get_finished_spans()}
assert len(spans) == 4
root, client, received, orphan = [spans[name]
    for name in ("request", "outbound", "handle/ok", "handle/fail")]
assert root.parent is None
assert root.context.trace_id == client.context.trace_id == received.context.trace_id
assert client.parent.span_id == root.context.span_id
assert received.parent.span_id == client.context.span_id and received.parent.is_remote
assert orphan.parent is None and orphan.context.trace_id != root.context.trace_id
assert orphan.status.status_code == StatusCode.ERROR
assert [event.name for event in orphan.events] == ["exception"]
assert received.resource.attributes["service.name"] == "worker"
assert client.resource.attributes["service.name"] == "caller"
print(json.dumps({"exported_spans": len(spans), "linked_spans": 3,
                  "unpropagated_request_started_new_trace": True,
                  "failed_span_status": orphan.status.status_code.name,
                  "exception_events": len(orphan.events)}, indent=2))
for provider in providers:
    provider.shutdown()