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.

The main subtlety is that I had to tweak some of httr2's internals so
that request signing can take into account new headers. Luckily there is
fairly comprehensive test coverage so I'm fairly sure at this point that
I haven't broken anything.

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.

Unit tests are included.

[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,8 @@ Suggests:
     knitr,
     later (>= 1.4.0),
     nanonext,
+    otel (>= 0.2.0),
+    otelsdk (>= 0.2.0),
     paws.common,
     promises,
     rmarkdown,
@@ -55,3 +57,5 @@ Config/testthat/start-first: resp-stream, req-perform
 Encoding: UTF-8
 Roxygen: list(markdown = TRUE)
 RoxygenNote: 7.3.2
+Remotes:
+    r-lib/otelsdk

NEWS.md

@@ -3,6 +3,9 @@
 * Refactor `url_modify()` to better retain exact formatting of URL components 
   that are not modified. (#788, #794)
 
+* httr2 will now emit OpenTelemetry traces for all requests when tracing is
+  enabled. Requires the `otelsdk` package (@atheriel, #729).
+
 # httr2 1.2.1
 
 * Colons in paths are no longer escaped.

R/otel.R

@@ -0,0 +1,114 @@
+# 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 = get_tracer(),
+  activation_scope = parent.frame(),
+  activate = TRUE
+) {
+  if (!is_tracing(tracer)) {
+    cli::cli_abort(
+      "Cannot create request span; tracing is not enabled",
+      .internal = TRUE
+    )
+  }
+  parsed <- tryCatch(url_parse(req$url), error = function(cnd) NULL)
+  if (is.null(parsed)) {
+    # Don't create spans for invalid URLs.
+    return(req)
+  }
+  if (!req_has_user_agent(req)) {
+    req <- req_user_agent(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_get_method(req)
+  # Set required (and some recommended) attributes, especially those 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,
+    "user_agent.original" = req$options$useragent
+  ))
+  span <- tracer$start_span(
+    name = method,
+    options = list(kind = "CLIENT"),
+    attributes = attributes
+  )
+  if (activate) {
+    span$activate(activation_scope, end_on_exit = TRUE)
+  }
+  req <- req_headers(req, !!!otel::pack_http_context())
+  req$state$span <- span
+  req
+}
+
+req_record_span_status <- function(req, resp = NULL) {
+  span <- req$state$span
+  if (is.null(span) || !span$is_recording()) {
+    return()
+  }
+  # For more accurate span timing, we end the span after the response has been
+  # received, rather than at the end of the associated scope.
+  on.exit(span$end())
+  if (is.null(resp)) {
+    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])
+    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")
+  }
+}
+
+get_tracer <- function() {
+  if (!is.null(the$tracer)) {
+    return(the$tracer)
+  }
+  if (!is_installed("otel")) {
+    return(NULL)
+  }
+  if (is_testing()) {
+    # Don't cache the tracer in unit tests. It interferes with tracer provider
+    # injection in otelsdk::with_otel_record().
+    return(otel::get_tracer("httr2"))
+  }
+  the$tracer <- otel::get_tracer("httr2")
+  the$tracer
+}
+
+is_tracing <- function(tracer = get_tracer()) {
+  !is.null(tracer) && tracer$is_enabled()
+}

R/pooled-request.R

@@ -66,6 +66,19 @@ PooledRequest <- R6Class(
 
       private$req_prep <- req_prepare(req)
       private$handle <- req_handle(private$req_prep)
+      if (is_tracing()) {
+        # Note: we need to do this before we call handle_preflight() so that
+        # request signing works correctly with the added headers.
+        #
+        # TODO: Support resend_count.
+        private$req_prep <- req_with_span(
+          private$req_prep,
+          # Pooled request spans should not become the active span; we want
+          # subsequent requests to be siblings rather than parents.
+          activate = FALSE
+        )
+      }
+      handle_preflight(private$req_prep, private$handle)
 
       curl::multi_add(
         handle = private$handle,
@@ -83,6 +96,9 @@ PooledRequest <- R6Class(
       if (!is.null(private$handle)) {
         curl::multi_cancel(private$handle)
       }
+      if (!is.null(private$req_prep)) {
+        req_record_span_status(private$req_prep)
+      }
     }
   ),
   private = list(
@@ -114,6 +130,7 @@ PooledRequest <- R6Class(
       }
 
       resp <- create_response(self$req, curl_data, body)
+      req_record_span_status(private$req_prep, resp)
       resp <- cache_post_fetch(self$req, resp, path = private$path)
       private$handle_response(resp, self$req)
     },
@@ -136,6 +153,7 @@ PooledRequest <- R6Class(
       curl_error <- error_cnd(message = msg, class = error_class, call = NULL)
       error <- curl_cnd(curl_error, call = private$error_call)
       error$request <- self$req
+      req_record_span_status(private$req_prep, error)
       private$on_error(error)
     }
   )

R/req-dry-run.R

@@ -61,7 +61,7 @@ req_dry_run <- function(
 
   req <- req_prepare(req)
   handle <- req_handle(req)
-  curl::handle_setopt(handle, url = req$url)
+  handle_preflight(req, handle)
   resp <- curl::curl_echo(handle, progress = FALSE)
   headers <- new_headers(
     as.list(resp$headers),

R/req-perform-connection.R

@@ -84,7 +84,13 @@ req_perform_connection <- function(
     if (!is.null(resp)) {
       close(resp)
     }
-    resp <- req_perform_connection1(req, handle, blocking = blocking)
+    resp <- req_perform_connection1(
+      req,
+      req_prep,
+      handle,
+      blocking = blocking,
+      resend_count = tries + 1L
+    )
 
     if (retry_is_transient(req, resp)) {
       tries <- tries + 1
@@ -95,7 +101,7 @@ req_perform_connection <- function(
       break
     }
   }
-  req_completed(req)
+  req_completed(req_prep)
 
   if (!is_error(resp) && error_is_error(req, resp)) {
     # Read full body if there's an error
@@ -135,10 +141,22 @@ req_verbosity_connection <- function(
   req
 }
 
-req_perform_connection1 <- function(req, handle, blocking = TRUE) {
+req_perform_connection1 <- function(
+  req,
+  req_prep,
+  handle,
+  blocking = TRUE,
+  resend_count = 0
+) {
   the$last_request <- req
   the$last_response <- NULL
   signal(class = "httr2_perform_connection")
+  if (is_tracing()) {
+    # Note: we need to do this before we call handle_preflight() so that request
+    # signing works correctly with the added headers.
+    req_prep <- req_with_span(req_prep, resend_count = resend_count)
+  }
+  handle_preflight(req_prep, handle)
 
   err <- capture_curl_error({
     conn <- curl::curl(req$url, handle = handle)
@@ -151,11 +169,14 @@ req_perform_connection1 <- function(req, handle, blocking = TRUE) {
     body <- StreamingBody$new(conn)
   })
   if (is_error(err)) {
+    req_record_span_status(req, err)
     return(err)
   }
 
   curl_data <- curl::handle_data(handle)
-  create_response(req, curl_data, body)
+  resp <- create_response(req, curl_data, body)
+  req_record_span_status(req, resp)
+  resp
 }
 
 # Make open mockable

R/req-perform.R

@@ -109,7 +109,13 @@ req_perform <- function(
     sys_sleep(delay, "for retry backoff")
     n <- n + 1
 
-    resp <- req_perform1(req, path = path, handle = handle)
+    resp <- req_perform1(
+      req,
+      req_prep,
+      path = path,
+      handle = handle,
+      resend_count = n
+    )
     req_completed(req_prep)
 
     if (retry_is_transient(req, resp)) {
@@ -174,23 +180,38 @@ resp_failure_cnd <- function(req, resp, error_call = caller_env()) {
   ))
 }
 
-req_perform1 <- function(req, path = NULL, handle = NULL) {
+req_perform1 <- function(
+  req,
+  req_prep,
+  path = NULL,
+  handle = NULL,
+  resend_count = 0
+) {
   the$last_request <- req
   the$last_response <- NULL
   signal(class = "httr2_perform")
+  if (is_tracing()) {
+    # Note: we need to do this before we call handle_preflight() so that request
+    # signing works correctly with the added headers.
+    req_prep <- req_with_span(req_prep, resend_count = resend_count)
+  }
+  handle_preflight(req_prep, handle)
 
   err <- capture_curl_error({
     fetch <- curl_fetch(handle, req$url, path)
   })
   if (is_error(err)) {
+    req_record_span_status(req, err)
     return(err)
   }
 
   # Ensure cookies are saved to disk now, not when request is finalised
   curl::handle_setopt(handle, cookielist = "FLUSH")
   curl::handle_setopt(handle, cookiefile = NULL, cookiejar = NULL)
 
-  create_response(req, fetch$curl_data, fetch$body)
+  resp <- create_response(req, fetch$curl_data, fetch$body)
+  req_record_span_status(req_prep, resp)
+  resp
 }
 
 curl_fetch <- function(handle, url, path) {
@@ -222,33 +243,37 @@ req_verbosity <- function(req, verbosity, error_call = caller_env()) {
 # Must call req_prepare(), then req_handle(), then after the request has been
 # performed, req_completed() (on the prepared requests)
 req_prepare <- function(req) {
-  req <- auth_sign(req)
   req <- req_method_apply(req)
   req <- req_body_apply(req)
-
-  # Save actually request headers so that req_verbose() can use them
-  req$state$headers <- req$headers
-
-  req
-}
-req_handle <- function(req) {
   if (!req_has_user_agent(req)) {
     req <- req_user_agent(req)
   }
+  req
+}
 
+req_handle <- function(req) {
   handle <- curl::new_handle()
   curl::handle_setopt(handle, url = req$url)
-  curl::handle_setheaders(
-    handle,
-    .list = headers_flatten(req$headers, redact = FALSE)
-  )
   curl::handle_setopt(handle, .list = req$options)
   if (length(req$fields) > 0) {
     curl::handle_setform(handle, .list = req$fields)
   }
 
   handle
 }
+
+# Called right before the request is sent, when the final headers are in place.
+handle_preflight <- function(req, handle) {
+  req <- auth_sign(req)
+  curl::handle_setheaders(
+    handle,
+    .list = headers_flatten(req$headers, redact = FALSE)
+  )
+  # Save final request headers so that req_verbose() can use them
+  req$state$headers <- req$headers
+  invisible(handle)
+}
+
 req_completed <- function(req) {
   req_policy_call(req, "done", list(), NULL)
 }