Add basic Open Telemetry instrumentation for all requests.

This commit wraps all requests in an Open Telemetry span that abides by
the semantic conventions for HTTP clients [0] (insofar as I understand
them). We also propagate the trace context [1] when there is one.

Right now this instrumentation is opt in: `otel` is in `Suggests`, and
tracing must be enabled (e.g. via the `OTEL_TRACES_EXPORTER` environment
variable). Otherwise this is costless at runtime.

For example:

    library(otelsdk)

    Sys.setenv(OTEL_TRACES_EXPORTER = "stderr")

    request("https://google.com") |>
      req_perform()

I'm not sure that `otel` needs to move to `Imports`, because by design
users actually need the `otelsdk` package to enable tracing anyway.

[0]: https://opentelemetry.io/docs/specs/semconv/http/http-spans/#http-client-span
[1]: https://www.w3.org/TR/trace-context/

Signed-off-by: Aaron Jacobs <[email protected]>

Author: atheriel

DESCRIPTION

@@ -39,6 +39,7 @@ Suggests:
     knitr,
     later (>= 1.4.0),
     nanonext,
+    otel (>= 0.0.0.9000),
     paws.common,
     promises,
     rmarkdown,
@@ -56,4 +57,5 @@ Encoding: UTF-8
 Roxygen: list(markdown = TRUE)
 RoxygenNote: 7.3.2
 Remotes:  
-    r-lib/webfakes
+    r-lib/webfakes,
+    r-lib/otel

NEWS.md

@@ -1,6 +1,7 @@
 # httr2 (development version)
 
 * `req_url_query()` now re-calculates n lengths when using `.multi = "explode"` to avoid select/recycling issues (@Kevanness, #719).
+* httr2 will now emit OpenTelemetry traces for all requests when tracing is enabled. Requires the `otelsdk` package (@atheriel, #729).
 
 # httr2 1.1.2
 

R/otel.R

@@ -0,0 +1,109 @@
+# Attaches an Open Telemetry span that abides by the semantic conventions for
+# HTTP clients to the request, including the associated W3C trace context
+# headers.
+#
+# See: https://opentelemetry.io/docs/specs/semconv/http/http-spans/#http-client-span
+req_with_span <- function(
+  req,
+  resend_count = 0,
+  tracer = NULL,
+  scope = parent.frame()
+) {
+  if (!is_installed("otel")) {
+    return(req)
+  }
+  tracer <- tracer %||% otel::get_tracer("httr2")
+  if (!tracer$is_enabled()) {
+    return(req)
+  }
+  parsed <- tryCatch(url_parse(req$url), error = function(cnd) NULL)
+  if (is.null(parsed)) {
+    # Don't create spans for invalid URLs.
+    return(req)
+  }
+  default_port <- 443L
+  if (parsed$scheme == "http") {
+    default_port <- 80L
+  }
+  # Follow the semantic conventions and redact credentials in the URL, when
+  # present.
+  if (!is.null(parsed$username)) {
+    parsed$username <- "REDACTED"
+  }
+  if (!is.null(parsed$password)) {
+    parsed$password <- "REDACTED"
+  }
+  method <- req_method_get(req)
+  span <- tracer$start_span(
+    name = method,
+    options = list(kind = "CLIENT"),
+    # Ensure we set attributes relevant to sampling at span creation time.
+    attributes = compact(list(
+      "http.request.method" = method,
+      "server.address" = parsed$hostname,
+      "server.port" = parsed$port %||% default_port,
+      "url.full" = url_build(parsed),
+      "http.request.resend_count" = if (resend_count > 1) resend_count
+    )),
+    scope = scope
+  )
+  ctx <- span$get_context()
+  req <- req_headers(req, !!!ctx$to_http_headers())
+  req$state$span <- span
+  req
+}
+
+# Ends the Open Telemetry span associated with this request, if any.
+req_end_span <- function(req, resp = NULL) {
+  span <- req$state$span
+  if (is.null(span) || !span$is_recording()) {
+    return()
+  }
+  if (is.null(resp)) {
+    span$end()
+    return()
+  }
+  if (is_error(resp)) {
+    span$record_exception(resp)
+    span$set_status("error")
+    # Surface the underlying curl error class.
+    span$set_attribute("error.type", class(resp$parent)[1])
+    span$end()
+    return()
+  }
+  span$set_attribute("http.response.status_code", resp_status(resp))
+  if (error_is_error(req, resp)) {
+    desc <- resp_status_desc(resp)
+    if (is.na(desc)) {
+      desc <- NULL
+    }
+    span$set_status("error", desc)
+    # The semantic conventions recommend using the status code as a string for
+    # these cases.
+    span$set_attribute("error.type", as.character(resp_status(resp)))
+  } else {
+    span$set_status("ok")
+  }
+  span$end()
+}
+
+# Replaces the existing Open Telemetry span on a request with a new one. Used
+# for retries.
+req_reset_span <- function(
+  req,
+  handle,
+  resend_count = 0,
+  tracer = NULL,
+  scope = parent.frame()
+) {
+  req <- req_with_span(req, resend_count, tracer, scope)
+  if (is.null(req$state$span)) {
+    return(req)
+  }
+  # Because the headers have changed, we need to re-sign the request and update
+  # stateful components (like the handle).
+  req <- auth_sign(req)
+  curl::handle_setheaders(handle, .list = headers_flatten(req$headers))
+  req$state$headers <- req$headers
+  req
+}

R/req-perform-connection.R

@@ -54,6 +54,7 @@ req_perform_connection <- function(req, blocking = TRUE, verbosity = NULL) {
   # verbosity checked in req_verbosity_connection
 
   req <- req_verbosity_connection(req, verbosity %||% httr2_verbosity())
+  req <- req_with_span(req)
   req_prep <- req_prepare(req)
   handle <- req_handle(req_prep)
   the$last_request <- req
@@ -71,7 +72,14 @@ req_perform_connection <- function(req, blocking = TRUE, verbosity = NULL) {
     if (!is.null(resp)) {
       close(resp)
     }
+
+    if (tries != 0) {
+      # Start a new span for retried requests.
+      req_prep <- req_reset_span(req_prep, handle, resend_count = tries)
+    }
+
     resp <- req_perform_connection1(req, handle, blocking = blocking)
+    req_completed(req_prep, resp)
 
     if (retry_is_transient(req, resp)) {
       tries <- tries + 1
@@ -82,7 +90,6 @@ req_perform_connection <- function(req, blocking = TRUE, verbosity = NULL) {
       break
     }
   }
-  req_completed(req)
 
   if (!is_error(resp) && error_is_error(req, resp)) {
     # Read full body if there's an error

R/req-perform.R

@@ -93,6 +93,9 @@ req_perform <- function(
     return(req)
   }
 
+  sys_sleep(throttle_delay(req), "for throttling delay")
+
+  req <- req_with_span(req)
   req_prep <- req_prepare(req)
   handle <- req_handle(req_prep)
   max_tries <- retry_max_tries(req)
@@ -101,17 +104,19 @@ req_perform <- function(
   n <- 0
   tries <- 0
   reauthed <- FALSE # only ever re-authenticate once
-
-  sys_sleep(throttle_delay(req), "for throttling delay")
-
   delay <- 0
   while (tries < max_tries && Sys.time() < deadline) {
     retry_check_breaker(req, tries, error_call = error_call)
     sys_sleep(delay, "for retry backoff")
     n <- n + 1
 
+    if (n != 1) {
+      # Start a new span for retried requests.
+      req_prep <- req_reset_span(req_prep, handle, resend_count = n)
+    }
+
     resp <- req_perform1(req, path = path, handle = handle)
-    req_completed(req_prep)
+    req_completed(req_prep, resp)
 
     if (retry_is_transient(req, resp)) {
       tries <- tries + 1
@@ -270,7 +275,8 @@ req_handle <- function(req) {
 
   handle
 }
-req_completed <- function(req) {
+req_completed <- function(req, resp = NULL) {
+  req_end_span(req, resp)
   req_policy_call(req, "done", list(), NULL)
 }