CSHARP-5017: Retry KMS requests on transient errors #1541
There are no files selected for viewing
This file contains 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 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 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 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 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 |
|---|---|---|
|
|
@@ -211,22 +211,20 @@ private SslStreamSettings GetTlsStreamSettings(string kmsProvider) | |
|
|
||
| private void ProcessNeedKmsState(CryptContext context, CancellationToken cancellationToken) | ||
| { | ||
| while (context.GetNextKmsMessageRequest() is { } request) | ||
| { | ||
| SendKmsRequest(request, cancellationToken); | ||
| } | ||
| context.MarkKmsDone(); | ||
| } | ||
|
|
||
| private async Task ProcessNeedKmsStateAsync(CryptContext context, CancellationToken cancellationToken) | ||
| { | ||
| while (context.GetNextKmsMessageRequest() is { } request) | ||
| { | ||
| await SendKmsRequestAsync(request, cancellationToken).ConfigureAwait(false); | ||
| } | ||
| context.MarkKmsDone(); | ||
| } | ||
|
|
||
| private void ProcessNeedMongoKeysState(CryptContext context, CancellationToken cancellationToken) | ||
|
|
@@ -278,48 +276,90 @@ private static byte[] ProcessReadyState(CryptContext context) | |
|
|
||
| private void SendKmsRequest(KmsRequest request, CancellationToken cancellation) | ||
| { | ||
| try | ||
| { | ||
| var endpoint = CreateKmsEndPoint(request.Endpoint); | ||
|
|
||
| var tlsStreamSettings = GetTlsStreamSettings(request.KmsProvider); | ||
| var sslStreamFactory = new SslStreamFactory(tlsStreamSettings, _networkStreamFactory); | ||
| using var sslStream = sslStreamFactory.CreateStream(endpoint, cancellation); | ||
|
|
||
| var sleepMs = request.Sleep; | ||
| if (sleepMs > 0) | ||
| { | ||
| Thread.Sleep(sleepMs); | ||
| } | ||
|
|
||
| using var binary = request.GetMessage(); | ||
| var requestBytes = binary.ToArray(); | ||
| sslStream.Write(requestBytes, 0, requestBytes.Length); | ||
|
|
||
| while (request.BytesNeeded > 0) | ||
| { | ||
| var buffer = new byte[request.BytesNeeded]; // BytesNeeded is the maximum number of bytes that libmongocrypt wants to receive. | ||
| var count = sslStream.Read(buffer, 0, buffer.Length); | ||
|
|
||
| if (count == 0) | ||
| { | ||
| throw new IOException("Unexpected end of stream. No data was read from the SSL stream."); | ||
| } | ||
|
|
||
| var responseBytes = new byte[count]; | ||
| Buffer.BlockCopy(buffer, 0, responseBytes, 0, count); | ||
| request.Feed(responseBytes); | ||
| } | ||
| } | ||
| catch (Exception ex) when (ex is IOException or SocketException) | ||
| { | ||
| if (!request.Fail()) | ||
| { | ||
| throw; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private async Task SendKmsRequestAsync(KmsRequest request, CancellationToken cancellation) | ||
| { | ||
| try | ||
| { | ||
| var endpoint = CreateKmsEndPoint(request.Endpoint); | ||
|
|
||
| var tlsStreamSettings = GetTlsStreamSettings(request.KmsProvider); | ||
| var sslStreamFactory = new SslStreamFactory(tlsStreamSettings, _networkStreamFactory); | ||
| using var sslStream = await sslStreamFactory.CreateStreamAsync(endpoint, cancellation).ConfigureAwait(false); | ||
|
|
||
| var sleepMs = request.Sleep; | ||
| if (sleepMs > 0) | ||
| { | ||
| await Task.Delay(sleepMs, cancellation).ConfigureAwait(false); | ||
| } | ||
|
|
||
| using var binary = request.GetMessage(); | ||
| var requestBytes = binary.ToArray(); | ||
| await sslStream.WriteAsync(requestBytes, 0, requestBytes.Length).ConfigureAwait(false); | ||
|
|
||
| while (request.BytesNeeded > 0) | ||
| { | ||
| var buffer = new byte[request.BytesNeeded]; // BytesNeeded is the maximum number of bytes that libmongocrypt wants to receive. | ||
| var count = await sslStream.ReadAsync(buffer, 0, buffer.Length).ConfigureAwait(false); | ||
|
|
||
| if (count == 0) | ||
| { | ||
| throw new IOException("Unexpected end of stream. No data was read from the SSL stream."); | ||
| } | ||
|
|
||
| var responseBytes = new byte[count]; | ||
| Buffer.BlockCopy(buffer, 0, responseBytes, 0, count); | ||
| request.Feed(responseBytes); | ||
| } | ||
| } | ||
| catch (Exception ex) when (ex is IOException or SocketException) | ||
|
There was a problem hiding this comment. Not sure here which other kind of network-errors related exceptions we could get There was a problem hiding this comment. I saw some comment on the Drivers ticket for this feature that it only pertains to HTTP errors so for socket-level errors do we also retry on those? Other comments in the ticket seems to suggest we'll need to tell libmongocrypt to reset its state if we encounter socket errors while reading a kms response. Actually are http errors thrown as socket or IOException errors? There was a problem hiding this comment. Yes, I think that comment is a little bit misleading. So, as far as I have understood the situation is like this:
|
||
| { | ||
| if (!request.Fail()) | ||
| { | ||
| throw; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // nested type | ||
|
|
||
This file contains 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 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 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
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.
Agreed!
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.
Done :)