Skip to content

Add per-transaction HTTP metrics #15

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
o-nnerb merged 5 commits into from
Oct 2, 2026
Merged

Add per-transaction HTTP metrics #15

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ec05cfd
Add per-transaction HTTP metrics (request/response phases and connect…
o-nnerb Oct 2, 2026
c81ad0e
Add connection setup metrics (DNS, connect, TLS, proxy)
o-nnerb Oct 2, 2026
211f8e6
Add byte counts, TLS version and cipher suite to the transaction metrics
o-nnerb Oct 2, 2026
0bfbb2b
Make the raw responder of the byte metrics test Sendable-clean
o-nnerb Oct 2, 2026
bf72233
Reword a comment that the unacceptable language check rejects
o-nnerb Oct 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 82 additions & 5 deletions Sources/AsyncHTTPClient/AsyncAwait/HTTPClient+execute.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
//===----------------------------------------------------------------------===//

import Logging
import NIOConcurrencyHelpers
import NIOCore
import NIOHTTP1
import Tracing
Expand Down Expand Up @@ -46,7 +47,40 @@ extension HTTPClient {
request,
deadline: deadline,
logger: logger ?? Self.loggingDisabled,
redirectMode: self.configuration.redirectConfiguration.mode
redirectMode: self.configuration.redirectConfiguration.mode,
metricsHandler: nil
)
}
}

/// Execute arbitrary HTTP requests and receive the timing and connection information of every transaction.
///
/// - Parameters:
/// - request: HTTP request to execute.
/// - deadline: Point in time by which the request must complete.
/// - logger: The logger to use for this request.
/// - metrics: Called once per HTTP transaction, i.e. once per followed redirect plus once for the final response.
/// The calls are made in order and may happen on any thread. Each call happens when the transaction ended,
/// which for the last one can be after this method returned, because the response body is still being streamed.
/// A transaction that fails is reported as well, with ``HTTPClientTransactionMetrics/error`` set.
///
/// - warning: This method may violates Structured Concurrency because it returns a `HTTPClientResponse` that needs to be
/// streamed by the user. This means the request, the connection and other resources are still alive when the request returns.
///
/// - Returns: The response to the request. Note that the `body` of the response may not yet have been fully received.
public func execute(
_ request: HTTPClientRequest,
deadline: NIODeadline,
logger: Logger? = nil,
metrics: @escaping @Sendable (HTTPClientTransactionMetrics) -> Void
) async throws -> HTTPClientResponse {
try await withRequestSpan(request) {
try await self.executeAndFollowRedirectsIfNeeded(
request,
deadline: deadline,
logger: logger ?? Self.loggingDisabled,
redirectMode: self.configuration.redirectConfiguration.mode,
metricsHandler: metrics
)
}
}
Expand Down Expand Up @@ -78,6 +112,34 @@ extension HTTPClient {
logger: logger
)
}

/// Execute arbitrary HTTP requests and receive the timing and connection information of every transaction.
///
/// See ``execute(_:deadline:logger:metrics:)`` for when and how often `metrics` is called.
///
/// - Parameters:
/// - request: HTTP request to execute.
/// - timeout: time the the request has to complete.
/// - logger: The logger to use for this request.
/// - metrics: Called once per HTTP transaction.
///
/// - warning: This method may violates Structured Concurrency because it returns a `HTTPClientResponse` that needs to be
/// streamed by the user. This means the request, the connection and other resources are still alive when the request returns.
///
/// - Returns: The response to the request. Note that the `body` of the response may not yet have been fully received.
public func execute(
_ request: HTTPClientRequest,
timeout: TimeAmount,
logger: Logger? = nil,
metrics: @escaping @Sendable (HTTPClientTransactionMetrics) -> Void
) async throws -> HTTPClientResponse {
try await self.execute(
request,
deadline: .now() + timeout,
logger: logger,
metrics: metrics
)
}
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
Expand All @@ -88,7 +150,8 @@ extension HTTPClient {
_ request: HTTPClientRequest,
deadline: NIODeadline,
logger: Logger,
redirectMode: HTTPClient.Configuration.RedirectConfiguration.Mode
redirectMode: HTTPClient.Configuration.RedirectConfiguration.Mode,
metricsHandler: (@Sendable (HTTPClientTransactionMetrics) -> Void)?
) async throws -> HTTPClientResponse {
var currentRequest = request
var currentRedirectState = RedirectState(redirectMode, initialURL: request.url)
Expand All @@ -97,6 +160,8 @@ extension HTTPClient {

// this loop is there to follow potential redirects
while true {
let hopMetricsHandler = metricsHandler.map { HTTPHopMetricsHandler($0) }

let preparedRequest =
try HTTPClientRequest.Prepared(
currentRequest,
Expand All @@ -105,7 +170,12 @@ extension HTTPClient {
tracing: self.configuration.tracing
)
let response = try await {
var response = try await self.executeCancellable(preparedRequest, deadline: deadline, logger: logger)
var response = try await self.executeCancellable(
preparedRequest,
deadline: deadline,
logger: logger,
metricsHandler: hopMetricsHandler?.transactionHandler
)

history.append(
.init(
Expand Down Expand Up @@ -217,6 +287,11 @@ extension HTTPClient {
currentRequest = newRequest
}
}

// Only the paths that follow a redirect get here. `response` must stay alive until the
// flag is set, otherwise its body could be dropped, and the hop cancelled, before.
hopMetricsHandler?.redirectWillBeFollowed()
withExtendedLifetime(response) {}
}
}

Expand All @@ -225,7 +300,8 @@ extension HTTPClient {
private func executeCancellable(
_ request: HTTPClientRequest.Prepared,
deadline: NIODeadline,
logger: Logger
logger: Logger,
metricsHandler: (@Sendable (HTTPClientTransactionMetrics) -> Void)?
) async throws -> HTTPClientResponse {
let cancelHandler = TransactionCancelHandler()

Expand All @@ -246,7 +322,8 @@ extension HTTPClient {
logger: logger,
connectionDeadline: .now() + (self.configuration.timeout.connectionCreationTimeout),
preferredEventLoop: eventLoop,
responseContinuation: continuation
responseContinuation: continuation,
metricsHandler: metricsHandler
)

cancelHandler.registerTransaction(transaction)
Expand Down
74 changes: 67 additions & 7 deletions Sources/AsyncHTTPClient/AsyncAwait/Transaction.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,20 +36,38 @@ final class Transaction:

private let state: NIOLockedValueBox<StateMachine>

private let metrics: NIOLockedValueBox<HTTPRequestMetricsRecorder>
private let metricsHandler: (@Sendable (HTTPClientTransactionMetrics) -> Void)?

init(
request: HTTPClientRequest.Prepared,
requestOptions: RequestOptions,
logger: Logger,
connectionDeadline: NIODeadline,
preferredEventLoop: EventLoop,
responseContinuation: CheckedContinuation<HTTPClientResponse, Error>
responseContinuation: CheckedContinuation<HTTPClientResponse, Error>,
metricsHandler: (@Sendable (HTTPClientTransactionMetrics) -> Void)? = nil
) {
self.request = request
self.requestOptions = requestOptions
self.logger = logger
self.connectionDeadline = connectionDeadline
self.preferredEventLoop = preferredEventLoop
self.state = NIOLockedValueBox(StateMachine(responseContinuation))
self.metrics = NIOLockedValueBox(HTTPRequestMetricsRecorder(url: request.url))
self.metricsHandler = metricsHandler
}

/// Hands the recorded metrics to the handler. Only the first call has an effect, so every path
/// that ends the transaction can call this without coordinating with the others.
private func deliverMetrics(error: (any Error)?) {
guard let metricsHandler = self.metricsHandler else {
return
}
let metrics = self.metrics.withLockedValue { $0.finish(error: error) }
if let metrics {
metricsHandler(metrics)
}
}

func cancel() {
Expand All @@ -67,6 +85,7 @@ final class Transaction:

switch writeAction {
case .writeAndWait(let executor), .writeAndContinue(let executor):
self.metrics.withLockedValue { $0.requestBodyBytesWritten(byteBuffer.readableBytes) }
executor.writeRequestBodyPart(.byteBuffer(byteBuffer), request: self, promise: nil)

case .fail:
Expand Down Expand Up @@ -114,6 +133,7 @@ final class Transaction:

switch action {
case .writeAndContinue(let executor):
self.metrics.withLockedValue { $0.requestBodyBytesWritten(part.readableBytes) }
executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil)
case .writeAndWait:
// Holding the lock here *should* be safe but because of a bug in the runtime
Expand All @@ -135,9 +155,11 @@ final class Transaction:

switch action {
case .writeAndContinue(let executor):
self.metrics.withLockedValue { $0.requestBodyBytesWritten(part.readableBytes) }
executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil)
continuation.resume()
case .writeAndWait(let executor):
self.metrics.withLockedValue { $0.requestBodyBytesWritten(part.readableBytes) }
executor.writeRequestBodyPart(.byteBuffer(part), request: self, promise: nil)
case .fail:
continuation.resume(throwing: BreakTheWriteLoopError())
Expand Down Expand Up @@ -179,6 +201,8 @@ extension Transaction: HTTPSchedulableRequest {
var requiredEventLoop: EventLoop? { nil }

func requestWasQueued(_ scheduler: HTTPRequestScheduler) {
let time = NIODeadline.now()
self.metrics.withLockedValue { $0.requestWasQueued(at: time) }
self.state.withLockedValue { state in
state.requestWasQueued(scheduler)
}
Expand All @@ -193,6 +217,10 @@ extension Transaction: HTTPExecutableRequest {

// MARK: Request

func connectionAcquired(_ info: HTTPConnectionMetricsInfo) {
self.metrics.withLockedValue { $0.connectionAcquired(info) }
}

func willExecuteRequest(_ executor: HTTPRequestExecutor) {
let action = self.state.withLockedValue { state in
state.willExecuteRequest(executor)
Expand All @@ -203,14 +231,24 @@ extension Transaction: HTTPExecutableRequest {
executor.cancelRequest(self)
case .cancelAndFail(let executor, let continuation, with: let error):
executor.cancelRequest(self)
self.deliverMetrics(error: error)
continuation.resume(throwing: error)
case .none:
break
}
}

func requestHeadSent() {
// protocol requirement. Intentionally not needed.
// A request without a body is complete as soon as its head was sent, `requestBodyStreamSent`
// is not called for it.
let isComplete = self.requestFramingMetadata.body == .fixedSize(0)
let time = NIODeadline.now()
self.metrics.withLockedValue {
$0.requestHeadSent()
if isComplete {
$0.requestEnded(at: time)
}
}
}

func resumeRequestBodyStream() {
Expand Down Expand Up @@ -256,6 +294,9 @@ extension Transaction: HTTPExecutableRequest {
}

func requestBodyStreamSent() {
let time = NIODeadline.now()
self.metrics.withLockedValue { $0.requestEnded(at: time) }

let action = self.state.withLockedValue { state in
state.requestBodyStreamSent()
}
Expand All @@ -271,6 +312,9 @@ extension Transaction: HTTPExecutableRequest {
// MARK: Response

func receiveResponseHead(_ head: HTTPResponseHead) {
let time = NIODeadline.now()
self.metrics.withLockedValue { $0.responseHeadReceived(at: time) }

let action = self.state.withLockedValue { state in
state.receiveResponseHead(head, delegate: self)
}
Expand All @@ -294,6 +338,9 @@ extension Transaction: HTTPExecutableRequest {
}

func receiveResponseBodyParts(_ buffer: CircularBuffer<ByteBuffer>) {
let bytes = buffer.reduce(0) { $0 + $1.readableBytes }
self.metrics.withLockedValue { $0.responseBodyBytesDelivered(bytes) }

let action = self.state.withLockedValue { state in
state.receiveResponseBodyParts(buffer)
}
Expand All @@ -311,11 +358,19 @@ extension Transaction: HTTPExecutableRequest {
}

func receiveResponseEnd(_ buffer: CircularBuffer<ByteBuffer>?, trailers: HTTPHeaders?) {
let time = NIODeadline.now()
let bytes = buffer?.reduce(0) { $0 + $1.readableBytes } ?? 0
self.metrics.withLockedValue {
$0.responseEnded(at: time)
$0.responseBodyBytesDelivered(bytes)
}

let receiveResponseEndAction = self.state.withLockedValue { state in
state.receiveResponseEnd(buffer, trailers: trailers)
}
switch receiveResponseEndAction {
case .finishResponseStream(let source, let finalResponse):
self.deliverMetrics(error: nil)
if let finalResponse = finalResponse {
_ = source.yield(contentsOf: finalResponse)
}
Expand Down Expand Up @@ -353,22 +408,26 @@ extension Transaction: HTTPExecutableRequest {
break

case .failResponseHead(let continuation, let error, let scheduler, let executor, let bodyStreamContinuation):
continuation.resume(throwing: error)
bodyStreamContinuation?.resume(throwing: error)
scheduler?.cancelRequest(self) // NOTE: scheduler and executor are exclusive here
executor?.cancelRequest(self)
self.deliverMetrics(error: error)
continuation.resume(throwing: error)
bodyStreamContinuation?.resume(throwing: error)

case .failResponseStream(let source, let error, let executor, let requestBodyStreamContinuation):
executor.cancelRequest(self)
self.deliverMetrics(error: error)
source.finish(error)
requestBodyStreamContinuation?.resume(throwing: error)
executor.cancelRequest(self)

case .failRequestStreamContinuation(let bodyStreamContinuation, let error, let executor):
bodyStreamContinuation.resume(throwing: error)
executor.cancelRequest(self)
self.deliverMetrics(error: error)
bodyStreamContinuation.resume(throwing: error)

case .cancelExecutor(let executor):
executor.cancelRequest(self)
self.deliverMetrics(error: HTTPClientError.cancelled)
}
}

Expand All @@ -382,9 +441,10 @@ extension Transaction: HTTPExecutableRequest {
private func performDeadlineExceededAction(_ action: StateMachine.DeadlineExceededAction) {
switch action {
case .cancel(let requestContinuation, let scheduler, let executor, let bodyStreamContinuation):
requestContinuation.resume(throwing: HTTPClientError.deadlineExceeded)
scheduler?.cancelRequest(self)
executor?.cancelRequest(self)
self.deliverMetrics(error: HTTPClientError.deadlineExceeded)
requestContinuation.resume(throwing: HTTPClientError.deadlineExceeded)
bodyStreamContinuation?.resume(throwing: HTTPClientError.deadlineExceeded)
case .cancelSchedulerOnly(let scheduler):
scheduler.cancelRequest(self)
Expand Down
Loading
Loading