Handle throttling in FileDescriptor implementations
#1149
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| package com.kevel.apso.aws | ||
|
|
||
| import org.specs2.mutable.Specification | ||
| import software.amazon.awssdk.awscore.exception.AwsErrorDetails | ||
| import software.amazon.awssdk.core.exception.SdkClientException | ||
| import software.amazon.awssdk.services.s3.model.S3Exception | ||
|
|
||
| class S3BucketSpec extends Specification { | ||
|
|
||
| // The message the CRT-based S3 client builds for a failed request, as seen in production. | ||
| private def clientError(errorString: String) = | ||
| SdkClientException.create(s"Failed to send the request: $errorString") | ||
|
|
||
| private def serviceError(statusCode: Int, errorCode: String) = | ||
| S3Exception | ||
| .builder() | ||
| .statusCode(statusCode) | ||
| .awsErrorDetails(AwsErrorDetails.builder().errorCode(errorCode).serviceName("S3").build()) | ||
| .message("Boom") | ||
| .build() | ||
|
|
||
| "An S3Bucket" should { | ||
|
|
||
| "recognize a slow down" in { | ||
|
|
||
| "reported as a service error" in { | ||
| S3Bucket.isSlowDown(serviceError(503, "SlowDown")) must beTrue | ||
| } | ||
|
|
||
| "reported as a service error with a 429 status code" in { | ||
| S3Bucket.isSlowDown(serviceError(429, "TooManyRequestsException")) must beTrue | ||
| } | ||
|
|
||
| "reported by the CRT client as a client-side error" in { | ||
| S3Bucket.isSlowDown(clientError(S3Bucket.SlowDownErrorMessage)) must beTrue | ||
| } | ||
| } | ||
|
|
||
| "not recognize as a slow down" in { | ||
|
|
||
| // A 503 alone doesn't imply throttling, so it must keep being handled as a plain service error. | ||
| "a service error with a 503 status code but no throttling error code" in { | ||
| S3Bucket.isSlowDown(serviceError(503, "ServiceUnavailable")) must beFalse | ||
| } | ||
|
|
||
| "a service error reporting another failure" in { | ||
| S3Bucket.isSlowDown(serviceError(404, "NoSuchKey")) must beFalse | ||
| } | ||
|
|
||
| "a client-side error reporting another failure" in { | ||
| S3Bucket.isSlowDown(clientError("Socket closed")) must beFalse | ||
| } | ||
|
|
||
| "a client-side error without a message" in { | ||
| S3Bucket.isSlowDown(SdkClientException.builder().build()) must beFalse | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,9 +1,10 @@ | ||
| package com.kevel.apso | ||
|
|
||
| import scala.annotation.tailrec | ||
| import scala.concurrent.duration.* | ||
| import scala.concurrent.{ExecutionContext, Future, blocking} | ||
| import scala.util.control.NonFatal | ||
| import scala.util.{Failure, Random, Success, Try} | ||
|
|
||
| /** Utility object with retry mechanisms. | ||
| */ | ||
|
|
@@ -73,4 +74,86 @@ object Retry { | |
| */ | ||
| def retry[T](maxRetries: Int = 10, inBetweenSleep: FiniteDuration = 100.millis)(f: => T): Try[T] = | ||
| retry(maxRetries, Option(inBetweenSleep))(f) | ||
|
|
||
| /** Computes the duration to wait before the next attempt, growing exponentially with the number of attempts made. | ||
| * | ||
| * @param attempt | ||
| * the zero-based index of the attempt that failed | ||
| * @param base | ||
| * the base waiting duration | ||
| * @param max | ||
| * the optional duration with which to cap the exponential growth | ||
| * @param factor | ||
| * the factor of the exponential duration | ||
| * @param jitter | ||
| * the upper bound of the random duration added to the result | ||
| * @return | ||
| * the duration to wait before the next attempt | ||
| */ | ||
| private[apso] def exponentialBackOffDelay( | ||
| attempt: Int, | ||
| base: FiniteDuration, | ||
| max: Option[FiniteDuration] = None, | ||
| factor: Double = 2.0, | ||
| jitter: FiniteDuration = 1.second | ||
| ): FiniteDuration = { | ||
| val exponential = (base.toMillis * Math.pow(factor, attempt.toDouble)).toLong | ||
| val capped = max.fold(exponential)(m => Math.min(exponential, m.toMillis)) | ||
| (capped + (Random.nextDouble() * jitter.toMillis).toLong).millis | ||
|
Contributor
There was a problem hiding this comment. I wonder about this jitter approach. Reading through https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/, I think we could do better. The default is 1.0 second currently and it's add after the capped value, so, to deal with contention from multiple callers, it will kind-of evenly spread things out at a maximum of like 1sec at the maximum, so for hundreds of callers, they will fall on the same second. There's a few different ways we can take it. If we commit to a "full jitter" implementation we can simplify this a lot and remove the However, if we want to be able to toggle the jitterness, maybe something like: val jitterFraction = Math.min(1.0, Math.max(0.0, jitter)) // being extra careful, lol...
(
(1.0 - jitterFraction) * capped +
ThreadLocalRandom.current().nextDouble() * jitterFraction * capped
).toLong.millisThis seems like a better alternative, but I didn't test or benchmarks it. Thoughts?
Contributor
Author
There was a problem hiding this comment. Great suggestion! I hadn't thought much about the jitter since this was mostly copied over from another internal project. See 8397dc1
Contributor
There was a problem hiding this comment. Reading the javadocs of |
||
| } | ||
|
|
||
| /** Performs a function `f` until it succeeds or until maximum retries is reached, waiting between attempts for a | ||
| * duration that grows exponentially with the number of attempts made. | ||
| * | ||
| * @param maxRetries | ||
| * the number of retries | ||
| * @param base | ||
| * the base waiting duration | ||
| * @param max | ||
| * the optional duration with which to cap the exponential growth | ||
| * @param factor | ||
| * the factor of the exponential duration | ||
| * @param jitter | ||
| * the upper bound of the random duration added to each waiting duration | ||
| * @param retryWhen | ||
| * the predicate deciding whether a failure is worth retrying | ||
| * @param onRetry | ||
| * the function called before each retry with the failure being retried, the duration that will be waited for and | ||
| * the number of retries still left | ||
| * @param onMaxRetriesReached | ||
| * the function called when the max retries are reached with the latest failure | ||
| * @param f | ||
| * the function to retry | ||
| * @return | ||
| * a Try of the `f` function result | ||
| */ | ||
| def exponentialBackOff[T]( | ||
| maxRetries: Int, | ||
| base: FiniteDuration, | ||
| max: Option[FiniteDuration] = None, | ||
| factor: Double = 2.0, | ||
| jitter: FiniteDuration = 1.second, | ||
| retryWhen: Throwable => Boolean = _ => true, | ||
| onRetry: (Throwable, FiniteDuration, Int) => Unit = (_, _, _) => (), | ||
| onMaxRetriesReached: Throwable => Unit = _ => () | ||
| )(f: => T): Try[T] = { | ||
| @tailrec | ||
| def aux(attempt: Int): Try[T] = | ||
| Try(f) match { | ||
| case res @ Success(_) => res | ||
| case failure @ Failure(ex) => | ||
| if (!retryWhen(ex)) failure | ||
| else if (attempt >= maxRetries) { | ||
| onMaxRetriesReached(ex) | ||
| failure | ||
| } else { | ||
| val delay = exponentialBackOffDelay(attempt, base, max, factor, jitter) | ||
| onRetry(ex, delay, maxRetries - attempt) | ||
| Thread.sleep(delay.toMillis) | ||
|
Contributor
There was a problem hiding this comment. I came across this blog https://blog.damavis.com/en/blocking-calls-and-asynchronous-programming-with-scala/ and thought to share. I don't think it matter that much for the current state of things, as apso is providing a "sync" API so this wouldn't have an effect on the apps using it.
Contributor
Author
There was a problem hiding this comment. Yes, it's not ideal. Since this was already the behavior of the Retry module and the API is sync, I think it should be acceptable. I added the |
||
| aux(attempt + 1) | ||
| } | ||
| } | ||
|
|
||
| aux(0) | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
handleris a partial function (it already was before this PR), but it's being called normally. It might make sense to add a safeguard here, something like:ex => !handler.applyOrElse(ex, (_: Throwable) => false), to make a "bug" fail-fast.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.
See 3c2395d