Skip to content
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

feat: support for OkHttp thread metrics #1022

Draft
wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

feat: support for OkHttp thread metrics #1022

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3d71280
(wip) support for OkHttp thread count
ianbotsf Jan 3, 2024
ea6d6d1
bumping to released aws-crt-kotlin version
ianbotsf Jan 5, 2024
6c8ce8d
Switch to idle/active threads as a dimension
ianbotsf Jan 12, 2024
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
2 changes: 1 addition & 1 deletion gradle/libs.versions.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ okio-version = "3.6.0"
otel-version = "1.32.0"
slf4j-version = "2.0.9"
slf4j-v1x-version = "1.7.36"
crt-kotlin-version = "0.8.2"
crt-kotlin-version = "0.8.3"

# codegen
smithy-version = "1.42.0"
Expand Down
28 changes: 21 additions & 7 deletions ...client-engine-okhttp/jvm/src/aws/smithy/kotlin/runtime/http/engine/okhttp/OkHttpEngine.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,15 @@ import aws.smithy.kotlin.runtime.http.HttpCall
import aws.smithy.kotlin.runtime.http.config.EngineFactory
import aws.smithy.kotlin.runtime.http.engine.*
import aws.smithy.kotlin.runtime.http.engine.internal.HttpClientMetrics
import aws.smithy.kotlin.runtime.http.engine.internal.ThreadState
import aws.smithy.kotlin.runtime.http.request.HttpRequest
import aws.smithy.kotlin.runtime.net.TlsVersion
import aws.smithy.kotlin.runtime.operation.ExecutionContext
import aws.smithy.kotlin.runtime.time.Instant
import aws.smithy.kotlin.runtime.time.fromEpochMilliseconds
import kotlinx.coroutines.job
import okhttp3.*
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit
import kotlin.time.toJavaDuration
import aws.smithy.kotlin.runtime.net.TlsVersion as SdkTlsVersion
Expand All @@ -40,8 +42,13 @@ public class OkHttpEngine(
override val engineConstructor: (OkHttpEngineConfig.Builder.() -> Unit) -> OkHttpEngine = ::invoke
}

private val metrics = HttpClientMetrics(TELEMETRY_SCOPE, config.telemetryProvider)
private val client = config.buildClient(metrics)
private val dispatcher = config.buildDispatcher()
private val metrics = HttpClientMetrics(
scope = TELEMETRY_SCOPE,
provider = config.telemetryProvider,
threadStateCallback = dispatcher::getThreadState,
)
private val client = config.buildClient(dispatcher, metrics)

override suspend fun roundTrip(context: ExecutionContext, request: HttpRequest): HttpCall {
val callContext = callContext()
Expand All @@ -68,10 +75,21 @@ public class OkHttpEngine(
}
}

private fun Dispatcher.getThreadState(): ThreadState? = (executorService as? ThreadPoolExecutor)?.let { executor ->
val active = executor.activeCount.toLong()
val idle = executor.poolSize.toLong() - active
ThreadState(idle, active)
}

private fun OkHttpEngineConfig.buildDispatcher() = Dispatcher().apply {
maxRequests = maxConcurrency.toInt()
maxRequestsPerHost = maxConcurrencyPerHost.toInt()
}

/**
* Convert SDK version of HTTP configuration to OkHttp specific configuration and return the configured client
*/
private fun OkHttpEngineConfig.buildClient(metrics: HttpClientMetrics): OkHttpClient {
private fun OkHttpEngineConfig.buildClient(dispatcher: Dispatcher, metrics: HttpClientMetrics): OkHttpClient {
val config = this

return OkHttpClient.Builder().apply {
Expand Down Expand Up @@ -100,10 +118,6 @@ private fun OkHttpEngineConfig.buildClient(metrics: HttpClientMetrics): OkHttpCl
)
connectionPool(pool)

val dispatcher = Dispatcher().apply {
maxRequests = config.maxConcurrency.toInt()
maxRequestsPerHost = config.maxConcurrencyPerHost.toInt()
}
dispatcher(dispatcher)

// Log events coming from okhttp. Allocate a new listener per-call to facilitate dedicated trace spans.
Expand Down
85 changes: 51 additions & 34 deletions ...ttp-client/common/src/aws/smithy/kotlin/runtime/http/engine/internal/HttpClientMetrics.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,13 @@ public object HttpClientMetricAttributes {
public val AcquiredConnection: Attributes = attributesOf { "state" to "acquired" }
public val QueuedRequest: Attributes = attributesOf { "state" to "queued" }
public val InFlightRequest: Attributes = attributesOf { "state" to "in-flight" }
public val IdleThread: Attributes = attributesOf { "state" to "idle" }
public val ActiveThread: Attributes = attributesOf { "state" to "active" }
}

@InternalApi
public data class ThreadState(val idle: Long, val active: Long)

/**
* Container for common HTTP engine related metrics. Engine implementations can re-use this and update
* the various fields in whatever manner fits best (increment/decrement vs current absolute value).
Expand All @@ -39,6 +44,7 @@ public object HttpClientMetricAttributes {
public class HttpClientMetrics(
scope: String,
public val provider: TelemetryProvider,
threadStateCallback: (() -> ThreadState?)? = null,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems different for no apparent reason. It's not wrong I guess but feels odd that we would special case it since each engine can adapt however they want to the current way metrics are exposed.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes I concede it's very different from what we have elsewhere and feels very specialized right now.

I think one problem is that we don't directly control the thread pool. We don't decide when new threads are created or when they swap from idle to active. All we can do is inspect the state at any given moment to know the current value. That sounds like what an AsyncUpDownCounter is intended to do via a callback.

We could start a background thread in the OkHttp engine to periodically inspect the thread pool and stuff the idle/active counts into HttpClientMetrics, which would then be read when the callback is run. But this means the value will be out of sync by whatever interval the background thread uses. If, for instance, the background thread runs every 10 seconds but the meter polls every 5 seconds, then up to two datapoints will be reported inaccurately.

Is your greater concern that this change exposes the concept of callbacks outside of HttpClientMetrics and inverts the collection model from a push to a pull? Is it the inconsistency of some metrics being exposed outside of HttpClientMetrics as callbacks while others are mutable properties inside? Or is it that this may not be flexible enough to adapt to other engines where thread updates are more deterministic?

) : Closeable {
private val meter = provider.meterProvider.getOrCreateMeter(scope)

Expand Down Expand Up @@ -76,34 +82,6 @@ public class HttpClientMetrics(
"The amount of time a connection has been open",
)

private val connectionLimitHandle = meter.createAsyncUpDownCounter(
"smithy.client.http.connections.limit",
{ it.record(_connectionsLimit.value) },
"{connection}",
"Max connections configured for the HTTP client",
)

private val connectionUsageHandle = meter.createAsyncUpDownCounter(
"smithy.client.http.connections.usage",
::recordConnectionState,
"{connection}",
"Current state of connections (idle, acquired)",
)

private val requestsConcurrencyLimitHandle = meter.createAsyncUpDownCounter(
"smithy.client.http.requests.limit",
{ it.record(_requestConcurrencyLimit.value) },
"{request}",
"Max concurrent requests configured for the HTTP client",
)

private val requestsHandle = meter.createAsyncUpDownCounter(
"smithy.client.http.requests.usage",
::recordRequestsState,
"{request}",
"The current state of HTTP client request concurrency (queued, in-flight)",
)

public val bytesSent: MonotonicCounter = meter.createMonotonicCounter(
"smithy.client.http.bytes_sent",
"By",
Expand All @@ -122,6 +100,41 @@ public class HttpClientMetrics(
"The amount of time after a request has been sent spent waiting on a response from the remote server",
)

private val asyncHandles = listOfNotNull(
meter.createAsyncUpDownCounter(
"smithy.client.http.connections.limit",
{ it.record(_connectionsLimit.value) },
"{connection}",
"Max connections configured for the HTTP client",
),
meter.createAsyncUpDownCounter(
"smithy.client.http.connections.usage",
::recordConnectionState,
"{connection}",
"Current state of connections (idle, acquired)",
),
meter.createAsyncUpDownCounter(
"smithy.client.http.requests.limit",
{ it.record(_requestConcurrencyLimit.value) },
"{request}",
"Max concurrent requests configured for the HTTP client",
),
meter.createAsyncUpDownCounter(
"smithy.client.http.requests.usage",
::recordRequestsState,
"{request}",
"The current state of HTTP client request concurrency (queued, in-flight)",
),
threadStateCallback?.let { callback ->
meter.createAsyncUpDownCounter(
"smithy.client.http.threads.count",
{ measurement -> recordThreadState(measurement, callback) },
"{thread}",
"Current state of HTTP engine threads (idle, active)",
)
},
)

/**
* The maximum number of connections configured for the client
*/
Expand Down Expand Up @@ -186,13 +199,17 @@ public class HttpClientMetrics(
measurement.record(acquiredConnections, HttpClientMetricAttributes.AcquiredConnection)
}

private fun recordThreadState(measurement: LongAsyncMeasurement, callback: () -> ThreadState?) {
callback()?.let { threadState ->
measurement.record(threadState.idle, HttpClientMetricAttributes.IdleThread)
measurement.record(threadState.active, HttpClientMetricAttributes.ActiveThread)
}
}

override fun close() {
val exceptions = listOf(
runCatching(connectionLimitHandle::stop),
runCatching(connectionUsageHandle::stop),
runCatching(requestsHandle::stop),
runCatching(requestsConcurrencyLimitHandle::stop),
).mapNotNull(Result<*>::exceptionOrNull)
val exceptions = asyncHandles
.map { handle -> runCatching { handle.stop() } }
.mapNotNull(Result<*>::exceptionOrNull)

exceptions.firstOrNull()?.let { first ->
exceptions.drop(1).forEach(first::addSuppressed)
Expand Down
Loading