diff --git a/std/net/grpc.pith b/std/net/grpc.pith index f0eff4ba..96c96bd7 100644 --- a/std/net/grpc.pith +++ b/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()) diff --git a/std/net/http.pith b/std/net/http.pith index f0cc601f..8aa3688e 100644 --- a/std/net/http.pith +++ b/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) diff --git a/std/trace.pith b/std/trace.pith index b47612a3..f371b5ca 100644 --- a/std/trace.pith +++ b/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).