Add basic Open Telemetry instrumentation for all requests

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)
 }