auto-instrument the std network clients when observability is on

std/net/grpc.pith

@@ -36,6 +36,9 @@ import std.bytes as bytes
 import std.bits as bits
 import std.encoding as encoding
 import std.net.tls as tls
+import std.trace as trace
+import std.metrics as metrics
+import std.time as time
 
 # grpc status codes (the grpc-status trailer). 0 is success; the rest are
 # errors. see https://grpc.io/docs/guides/status-codes/.
@@ -83,11 +86,28 @@ impl Conn:
     # are the serialized response message. fails with a GrpcError on a non-OK
     # grpc-status, a malformed response, or a transport error.
     pub fn unary(full_method: String, request: Bytes) -> Bytes!GrpcError:
+        if not trace.is_active():
+            framed := frame_message(request)
+            resp := self.client.request("POST", self.authority, full_method, grpc_headers(), framed)
+            if resp.is_err:
+                fail GrpcError(GRPC_UNAVAILABLE, "grpc transport error: {resp.err}")
+            return handle_response(resp.ok)!
+        # observability on: a CLIENT span, a traceparent injected into the request
+        # headers, and RED metrics — all invisible to the caller.
+        span := trace.start_kind(full_method, trace.CLIENT)
+        span.set_attr("rpc.system", "grpc").set_attr("rpc.method", full_method)
         framed := frame_message(request)
-        resp := self.client.request("POST", self.authority, full_method, grpc_headers(), framed)
+        started := time.mono_nanos()
+        resp := self.client.request("POST", self.authority, full_method, headers_with_traceparent(span.context()), framed)
         if resp.is_err:
+            finish_client_span(span, full_method, GRPC_UNAVAILABLE, started)
             fail GrpcError(GRPC_UNAVAILABLE, "grpc transport error: {resp.err}")
-        return handle_response(resp.ok)!
+        result := handle_response(resp.ok)
+        mut status := GRPC_OK
+        if result.is_err:
+            status = result.err.status
+        finish_client_span(span, full_method, status, started)
+        return result!
 
     # close the underlying http/2 connection. the channel cannot be used after.
     pub fn close():
@@ -131,6 +151,39 @@ REQUEST_HEADERS := [hpack.header_field("content-type", "application/grpc"), hpac
 fn grpc_headers() -> List[hpack.HeaderField]:
     return REQUEST_HEADERS
 
+# the standard grpc headers plus a W3C traceparent for `ctx`, so the server joins
+# this trace. builds a fresh list rather than pushing onto the shared cached one.
+fn headers_with_traceparent(ctx: trace.SpanContext) -> List[hpack.HeaderField]:
+    mut fields: List[hpack.HeaderField] := []
+    for f in grpc_headers():
+        fields.push(f)
+    fields.push(hpack.header_field("traceparent", trace.format_traceparent(ctx)))
+    return fields
+
+# the headers for a streaming open: the base grpc headers, plus a traceparent
+# carrying the current trace context when one is active — so a streaming rpc
+# started inside a span propagates it. off (or no active span) => base headers.
+fn stream_headers() -> List[hpack.HeaderField]:
+    if not trace.is_active():
+        return grpc_headers()
+    ctx := trace.current_context()
+    if not trace.context_is_valid(ctx):
+        return grpc_headers()
+    return headers_with_traceparent(ctx)
+
+# set a grpc client span's status from the grpc-status, end it, and record the RED
+# metrics (a request counter and a duration histogram) for the call.
+fn finish_client_span(span: trace.Span, full_method: String, status: Int, start_ns: Int):
+    if status == GRPC_OK:
+        span.set_status(trace.STATUS_OK, "")
+    else:
+        span.set_status(trace.STATUS_ERROR, status_name(status))
+    span.set_attr("rpc.grpc.status_code", status.to_string())
+    span.end()
+    elapsed_ms := (time.mono_nanos() - start_ns) / 1000000
+    metrics.counter("rpc_client_requests_total").labels(["rpc_method", full_method, "grpc_status", status_name(status)]).inc()
+    metrics.histogram("rpc_client_duration_ms").labels(["rpc_method", full_method]).observe(elapsed_ms)
+
 # frame a message for the wire: a one-byte compression flag (0 = identity), a
 # big-endian uint32 length, then the message bytes. the five prefix bytes are
 # written directly rather than through an intermediate list. writes to a fresh
@@ -381,7 +434,7 @@ impl Conn:
     # returned ServerStream to pull response messages until it reports done; a
     # non-OK grpc-status surfaces as a GrpcError from recv().
     pub fn server_stream(full_method: String, request: Bytes) -> ServerStream!GrpcError:
-        opened := self.client.open_stream("POST", self.authority, full_method, grpc_headers(), false)
+        opened := self.client.open_stream("POST", self.authority, full_method, stream_headers(), false)
         if opened.is_err:
             fail GrpcError(GRPC_UNAVAILABLE, "grpc: cannot open stream: {opened.err}")
         stream := opened.ok
@@ -460,7 +513,7 @@ impl Conn:
     # one response. `full_method` is "/package.Service/Method". send() each
     # request, close_send() to finish, then recv_response().
     pub fn client_stream(full_method: String) -> ClientStream!GrpcError:
-        opened := self.client.open_stream("POST", self.authority, full_method, grpc_headers(), false)
+        opened := self.client.open_stream("POST", self.authority, full_method, stream_headers(), false)
         if opened.is_err:
             fail GrpcError(GRPC_UNAVAILABLE, "grpc: cannot open stream: {opened.err}")
         return ClientStream(opened.ok, new_stream_reader())
@@ -470,7 +523,7 @@ impl Conn:
     # drive the request stream, recv() drains the response stream — the two can run
     # in separate tasks.
     pub fn bidi_stream(full_method: String) -> BidiStream!GrpcError:
-        opened := self.client.open_stream("POST", self.authority, full_method, grpc_headers(), false)
+        opened := self.client.open_stream("POST", self.authority, full_method, stream_headers(), false)
         if opened.is_err:
             fail GrpcError(GRPC_UNAVAILABLE, "grpc: cannot open stream: {opened.err}")
         return BidiStream(opened.ok, new_stream_reader())

std/net/http.pith

@@ -22,6 +22,7 @@ import std.net.tls as tls
 import std.os.path as path_mod
 import std.text.scanner as scanner
 import std.time as time_mod
+import std.trace as trace
 import std.uuid as uuid
 from std.concurrent import Context, BlockingError, blocking_cancelled, blocking_deadline_exceeded, blocking_failed
 from std.io import BufferedBytesReader, BufferedBytesWriter, BufferedStringReader, TcpStream, FileStream
@@ -2354,6 +2355,90 @@ fn send_request_following_redirects_with_tls_config(req: ClientRequest, tls_conf
         current = redirect_request(current, resp)!
         redirects_left = redirects_left - 1
 
+# the tls send path shared by send_tls_with_config and its traced wrapper: force
+# tls on, then dispatch to the plain-host redirect path or a single sni request.
+fn send_tls_request(req: ClientRequest, server_name: String, config: tls.Config) -> HttpResponse!:
+    current := ClientRequest(req.method, req.host, req.port, req.path, copy_headers_map(req.headers), req.body, true, req.timeout_ms, req.redirect_limit)
+    if server_name != req.host:
+        return send_request_tls(current, server_name, config)
+    return send_request_following_redirects_with_tls_config(current, config)
+
+# --- client-side tracing ----------------------------------------------------
+# these run only when tracing is active (the send methods gate on
+# trace.is_active() first), so an uninstrumented program pays a single bool check.
+
+# start a CLIENT span for an outbound request and tag it with the method and url.
+fn start_http_client_span(method: String, host: String, path: String) -> trace.Span:
+    span := trace.start_kind("HTTP " + method, trace.CLIENT)
+    span.set_attr("http.request.method", method).set_attr("http.url", host + path)
+    return span
+
+# set a client span's status from the response (>= 400, or `status` < 0 for a
+# transport error, is an error), end it, and record the RED metrics.
+fn finish_http_client_span(span: trace.Span, method: String, status: Int, start_ns: Int):
+    mut status_label := "error"
+    if status < 0:
+        span.set_status(trace.STATUS_ERROR, "transport error")
+    else:
+        status_label = status.to_string()
+        span.set_attr("http.response.status_code", status_label)
+        if status >= 400:
+            span.set_status(trace.STATUS_ERROR, "http status " + status_label)
+        else:
+            span.set_status(trace.STATUS_OK, "")
+    span.end()
+    elapsed_ms := (time_mod.mono_nanos() - start_ns) / 1000000
+    metrics.counter("http_client_requests_total").labels(["method", method, "status", status_label]).inc()
+    metrics.histogram("http_client_duration_ms").labels(["method", method]).observe(elapsed_ms)
+
+# --- server-side tracing ----------------------------------------------------
+
+# begin a SERVER span for an inbound request. it extracts any W3C `traceparent`
+# header so the span joins the caller's trace, then starts a span named
+# "METHOD path". pair it with end_server_span once the response status is known;
+# there is no server framework, so a handler wires the two calls in by hand:
+#
+#     span := http.begin_server_span(req)
+#     resp := handle(req)
+#     http.end_server_span(span, resp.status)
+#
+# a cheap no-op (a non-recording span) when tracing is inactive.
+pub fn begin_server_span(req: HttpRequestBytes) -> trace.Span:
+    if not trace.is_active():
+        return trace.start_kind("", trace.SERVER)
+    incoming := req.header("traceparent") catch ""
+    if incoming != "":
+        parsed := trace.parse_traceparent(incoming)
+        if parsed != none:
+            trace.with_context(parsed.value())
+    span := trace.start_kind(req.method_text + " " + req.path_text, trace.SERVER)
+    span.set_attr("http.request.method", req.method_text).set_attr("url.path", req.path_text)
+    return span
+
+# finish a server span from begin_server_span: set its status from the response
+# code (>= 500 is an error), end it, and record the http_server RED metrics
+# labeled by method, route, and status. a no-op when tracing is inactive.
+pub fn end_server_span(span: trace.Span, status: Int):
+    if not trace.is_active():
+        return
+    status_label := status.to_string()
+    span.set_attr("http.response.status_code", status_label)
+    if status >= 500:
+        span.set_status(trace.STATUS_ERROR, "http status " + status_label)
+    else:
+        span.set_status(trace.STATUS_OK, "")
+    span.end()
+    # the span name is "METHOD path"; split it back into the metric labels.
+    space := span.name.index_of(" ")
+    mut method := span.name
+    mut route := ""
+    if space >= 0:
+        method = span.name.substring(0, space)
+        route = span.name.substring(space + 1, span.name.len())
+    elapsed_ms := time_mod.now() - (span.start_unix_ns / 1000000)
+    metrics.counter("http_server_requests_total").labels(["method", method, "route", route, "status", status_label]).inc()
+    metrics.histogram("http_server_duration_ms").labels(["method", method, "route", route]).observe(elapsed_ms)
+
 fn send_request_following_redirects_into(req: ClientRequest, writer: BufferedBytesWriter) -> StreamedHttpResponse!:
     mut current := req
     mut redirects_left := req.redirect_limit
@@ -2478,7 +2563,17 @@ impl ClientRequest:
         return buf.bytes()
 
     fn send() -> HttpResponse!:
-        return send_request_following_redirects(self)
+        if not trace.is_active():
+            return send_request_following_redirects(self)
+        span := start_http_client_span(self.method, self.host, self.path)
+        traced := self.header("traceparent", trace.format_traceparent(span.context()))
+        started := time_mod.mono_nanos()
+        result := send_request_following_redirects(traced)
+        if result.is_err:
+            finish_http_client_span(span, self.method, - 1, started)
+        else:
+            finish_http_client_span(span, self.method, result.ok.status, started)
+        return result!
 
     fn send_tls() -> HttpResponse!:
         return self.send_tls_with_config(self.host, tls.client_config()!)
@@ -2487,10 +2582,17 @@ impl ClientRequest:
         return self.send_tls()
 
     fn send_tls_with_config(server_name: String, config: tls.Config) -> HttpResponse!:
-        current := ClientRequest(self.method, self.host, self.port, self.path, copy_headers_map(self.headers), self.body, true, self.timeout_ms, self.redirect_limit)
-        if server_name != self.host:
-            return send_request_tls(current, server_name, config)
-        return send_request_following_redirects_with_tls_config(current, config)
+        if not trace.is_active():
+            return send_tls_request(self, server_name, config)
+        span := start_http_client_span(self.method, self.host, self.path)
+        traced := self.header("traceparent", trace.format_traceparent(span.context()))
+        started := time_mod.mono_nanos()
+        result := send_tls_request(traced, server_name, config)
+        if result.is_err:
+            finish_http_client_span(span, self.method, - 1, started)
+        else:
+            finish_http_client_span(span, self.method, result.ok.status, started)
+        return result!
 
     fn send_into(writer: BufferedBytesWriter) -> StreamedHttpResponse!:
         return send_request_following_redirects_into(self, writer)

std/trace.pith

@@ -15,7 +15,7 @@
 # environment), so start() is a cheap no-op in an uninstrumented program. ended
 # spans collect in a buffer that the OTLP exporter drains; see std.otlp.
 
-import std.log as log
+import std.uuid as uuid
 import std.time as time
 
 # span kinds (OTel SpanKind).
@@ -102,10 +102,22 @@ pub struct Span:
     saved_span_id: String
     saved_sampled: Bool
 
+# a compact 32-hex-character id from a uuid (dashes stripped) — the trace-id
+# width; a 16-hex span id is its first half. these match std.log's id format so
+# a trace id lines up whether it came from here or from a log record.
+fn compact_uuid() -> String:
+    raw := uuid.v7() catch (uuid.v4() catch "")
+    return raw.replace("-", "")
+
 fn new_id(width: Int) -> String:
+    hex := compact_uuid()
     if width == 32:
-        return log.new_trace_id() catch "00000000000000000000000000000000"
-    return log.new_span_id() catch "0000000000000000"
+        if hex.len() == 32:
+            return hex
+        return "00000000000000000000000000000000"
+    if hex.len() >= 16:
+        return hex.substring(0, 16)
+    return "0000000000000000"
 
 # unix time in nanoseconds, at millisecond resolution (the wall clock is ms; a
 # span's duration is measured separately with the monotonic clock for precision).