feat: emit metrics from CRT HTTP engine #1017
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| { | ||
| "id": "e7c3c7ab-749e-4371-8b25-42ea76aa870d", | ||
| "type": "feature", | ||
| "description": "Emit metrics from CRT HTTP engine", | ||
| "issues": [ | ||
| "https://github.com/awslabs/smithy-kotlin/issues/893" | ||
| ] | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -13,9 +13,12 @@ import aws.smithy.kotlin.runtime.crt.SdkDefaultIO | |
| import aws.smithy.kotlin.runtime.http.HttpErrorCode | ||
| import aws.smithy.kotlin.runtime.http.HttpException | ||
| import aws.smithy.kotlin.runtime.http.engine.ProxyConfig | ||
| import aws.smithy.kotlin.runtime.http.engine.internal.HttpClientMetrics | ||
| import aws.smithy.kotlin.runtime.http.request.HttpRequest | ||
| import aws.smithy.kotlin.runtime.io.Closeable | ||
| import aws.smithy.kotlin.runtime.net.TlsVersion | ||
| import aws.smithy.kotlin.runtime.telemetry.metrics.measureSeconds | ||
| import kotlinx.atomicfu.atomic | ||
| import kotlinx.coroutines.TimeoutCancellationException | ||
| import kotlinx.coroutines.sync.Mutex | ||
| import kotlinx.coroutines.sync.Semaphore | ||
|
|
@@ -27,8 +30,10 @@ import aws.smithy.kotlin.runtime.net.TlsVersion as SdkTlsVersion | |
|
|
||
| internal class ConnectionManager( | ||
| private val config: CrtHttpEngineConfig, | ||
| private val metrics: HttpClientMetrics, | ||
| ) : Closeable { | ||
| private val leases = Semaphore(config.maxConnections.toInt()) | ||
| private val pending = atomic(0L) | ||
|
|
||
| private val crtTlsContext: TlsContext = TlsContextOptionsBuilder() | ||
| .apply { | ||
|
|
@@ -61,16 +66,22 @@ internal class ConnectionManager( | |
| val manager = getManagerForUri(request.uri, proxyConfig) | ||
| var leaseAcquired = false | ||
|
|
||
| metrics.queuedRequests = pending.incrementAndGet() | ||
|
|
||
| return try { | ||
| metrics.requestsQueuedDuration.measureSeconds { | ||
| // wait for an actual connection | ||
| val conn = withTimeout(config.connectionAcquireTimeout) { | ||
| // get a permit to acquire a connection (limits overall connections since managers are per/host) | ||
| leases.acquire() | ||
| leaseAcquired = true | ||
| metrics.connectionAcquireDuration.measureSeconds { | ||
| manager.acquireConnection() | ||
| } | ||
| } | ||
|
|
||
| LeasedConnection(conn) | ||
| } | ||
| } catch (ex: Exception) { | ||
| if (leaseAcquired) { | ||
| leases.release() | ||
|
|
@@ -82,8 +93,18 @@ internal class ConnectionManager( | |
| } | ||
|
|
||
| throw httpEx | ||
| } finally { | ||
| metrics.queuedRequests = pending.decrementAndGet() | ||
| emitConnections() | ||
| } | ||
| } | ||
|
|
||
| private fun emitConnections() { | ||
|
There was a problem hiding this comment. nit / naming: There was a problem hiding this comment. Since the class already has the word "connection" in it, renaming to simply |
||
| val idleConnections = leases.availablePermits.toLong() | ||
|
There was a problem hiding this comment. correctness: this semaphore isn't tied in anyway to actual connections in CRT, you could have 100 max connections configured but that doesn't mean you have 100 idle connections already connected waiting to be used. I'm thinking this would have to come from the actual CRT connection manager. There was a problem hiding this comment. Oh jeez, good point. Yes, this is clearly flawed. That brings up an interesting wrinkle, though. We don't have just one CRT connection manager, we have one for each host. Does the Smithy metric for acquired connections equal the sum of acquired connections across all underlying CRT connection managers? Moreover, we have a CRT config param called There was a problem hiding this comment. The fact CRT uses connection manager per host is an implementation detail, the config setting is at the engine level. This is what the The reason I allowed each connection manager to be configured with There was a problem hiding this comment. I see this as an area that may cause confusion later, especially if users begin correlating the idle connections metric with connections reported by the OS or JVM. I don't have a better suggestion though given how the CRT connection manager works. I'll switch the implementation to calculate the idle connections metric as There was a problem hiding this comment. Oh and interrogating the number of active connections from the CRT connection managers requires another revision of aws-crt-kotlin: awslabs/aws-crt-kotlin#88 |
||
| metrics.idleConnections = idleConnections | ||
| metrics.acquiredConnections = config.maxConnections.toLong() - idleConnections | ||
| } | ||
|
|
||
| private suspend fun getManagerForUri(uri: Uri, proxyConfig: ProxyConfig): HttpClientConnectionManager = mutex.withLock { | ||
| connManagers.getOrPut(uri.authority) { | ||
| val connOpts = options.apply { | ||
|
|
@@ -105,6 +126,7 @@ internal class ConnectionManager( | |
| HttpClientConnectionManager(connOpts) | ||
| } | ||
| } | ||
|
|
||
| override fun close() { | ||
| connManagers.forEach { entry -> entry.value.close() } | ||
| crtTlsContext.close() | ||
|
|
@@ -117,6 +139,7 @@ internal class ConnectionManager( | |
| delegate.close() | ||
| } finally { | ||
| leases.release() | ||
| emitConnections() | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
There was a problem hiding this comment. comment: These new body types are missing tests |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package aws.smithy.kotlin.runtime.http.engine.crt.io | ||
|
|
||
| import aws.smithy.kotlin.runtime.io.SdkBuffer | ||
| import aws.smithy.kotlin.runtime.io.SdkByteReadChannel | ||
| import aws.smithy.kotlin.runtime.telemetry.metrics.MonotonicCounter | ||
|
|
||
| private class ReportingByteReadChannel( | ||
| val delegate: SdkByteReadChannel, | ||
| val metric: MonotonicCounter, | ||
| ) : SdkByteReadChannel by delegate { | ||
| override suspend fun read(sink: SdkBuffer, limit: Long): Long = delegate.read(sink, limit).also { | ||
| if (it > 0) metric.add(it) | ||
| } | ||
| } | ||
|
|
||
| internal fun SdkByteReadChannel.reportingTo(metric: MonotonicCounter): SdkByteReadChannel = | ||
| ReportingByteReadChannel(this, metric) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package aws.smithy.kotlin.runtime.http.engine.crt.io | ||
|
|
||
| import aws.smithy.kotlin.runtime.io.SdkBuffer | ||
| import aws.smithy.kotlin.runtime.io.SdkSource | ||
| import aws.smithy.kotlin.runtime.telemetry.metrics.MonotonicCounter | ||
|
|
||
| private class ReportingSource(val delegate: SdkSource, val metric: MonotonicCounter) : SdkSource by delegate { | ||
| override fun read(sink: SdkBuffer, limit: Long): Long = delegate.read(sink, limit).also { | ||
| if (it > 0) metric.add(it) | ||
| } | ||
| } | ||
|
|
||
| internal fun SdkSource.reportingTo(metric: MonotonicCounter): SdkSource = ReportingSource(this, metric) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
correctness: If we are trying to acquire a connection then it's not queued right? I think queued would be before we hit the requestLimiter.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
OK, if "queued" means "before we attempt to acquire a connection" then I'm guessing that the
requestsQueuedDurationmeasurement below is also wrong. I'll move it too.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Had to look at the definitions I gave them again and I think this would be inline with what it says
If we are to the point of acquiring a connection we are "actively processing" the request. I can see where the definition could be interpreted differently though as in "I have all the resources needed at this point to execute the request" but I think establishing a connection (or waiting on one to be available) is part of overall request processing. Curious if others disagree.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This still doesn't look correct, I don't think
queuedRequestsis calculated in connection manager.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ah I think I was confusing the semaphore inside
ConnectionManagerwith the one insideCrtHttpEngine. Switching.