-
Notifications
You must be signed in to change notification settings - Fork 15
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Ensure we track retry attempts per request (#146)
As the pipeline is now created once per transport, we need to track the retires outside of the pipeline. It also fixes two other bugs where we don't null check correctly. Closes #145
- Loading branch information
1 parent
1bd2db0
commit 6dc8ff9
Showing
7 changed files
with
181 additions
and
39 deletions.
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
24 changes: 24 additions & 0 deletions
24
src/Elastic.Transport/Components/TransportClient/RequestInvokerHelpers.cs
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 |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| // Licensed to Elasticsearch B.V under one or more agreements. | ||
| // Elasticsearch B.V licenses this file to you under the Apache 2.0 License. | ||
| // See the LICENSE file in the project root for more information | ||
|
|
||
| using System.Diagnostics; | ||
| using Elastic.Transport.Diagnostics; | ||
|
|
||
| namespace Elastic.Transport; | ||
|
|
||
| internal static class RequestInvokerHelpers | ||
| { | ||
| public static void SetOtelAttributes<TResponse>(BoundConfiguration boundConfiguration, TResponse response) where TResponse : TransportResponse | ||
| { | ||
| if (!OpenTelemetry.CurrentSpanIsElasticTransportOwnedAndHasListeners || (!(Activity.Current?.IsAllDataRequested ?? false))) | ||
| return; | ||
|
|
||
| var attributes = boundConfiguration.ConnectionSettings.ProductRegistration.ParseOpenTelemetryAttributesFromApiCallDetails(response.ApiCallDetails); | ||
|
|
||
| if (attributes is null) return; | ||
|
|
||
| foreach (var attribute in attributes) | ||
| Activity.Current?.SetTag(attribute.Key, attribute.Value); | ||
| } | ||
| } |
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
60 changes: 60 additions & 0 deletions
60
tests/Elastic.Transport.Tests/Components/NodePool/StaticNodePoolTests.cs
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 |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| // Licensed to Elasticsearch B.V under one or more agreements. | ||
| // Elasticsearch B.V licenses this file to you under the Apache 2.0 License. | ||
| // See the LICENSE file in the project root for more information | ||
|
|
||
| using Xunit; | ||
| using System; | ||
| using FluentAssertions; | ||
| using System.Linq; | ||
|
|
||
| namespace Elastic.Transport.Tests.Components.NodePool | ||
| { | ||
| public class StaticNodePoolTests | ||
| { | ||
| [Fact] | ||
| public void MultipleRequests_WhenOnlyASingleEndpointIsConfigured_AndTheEndpointIsUnavailable_DoNotThrowAnException() | ||
| { | ||
| Node[] nodes = [new Uri("http://localhost:9200")]; | ||
| var pool = new StaticNodePool(nodes); | ||
| var transport = new DistributedTransport(new TransportConfiguration(pool)); | ||
|
|
||
| var response = transport.Request<StringResponse>(HttpMethod.GET, "/", null, null); | ||
|
|
||
| response.ApiCallDetails.SuccessOrKnownError.Should().BeFalse(); | ||
| response.ApiCallDetails.AuditTrail.Count.Should().Be(1); | ||
|
|
||
| var audit = response.ApiCallDetails.AuditTrail.First(); | ||
| audit.Event.Should().Be(Diagnostics.Auditing.AuditEvent.BadRequest); | ||
| audit.Node.FailedAttempts.Should().Be(1); | ||
| audit.Node.IsAlive.Should().BeFalse(); | ||
|
|
||
| response = transport.Request<StringResponse>(HttpMethod.GET, "/", null, null); | ||
|
|
||
| response.ApiCallDetails.SuccessOrKnownError.Should().BeFalse(); | ||
|
|
||
| var eventCount = 0; | ||
|
|
||
| foreach (var a in response.ApiCallDetails.AuditTrail) | ||
| { | ||
| eventCount++; | ||
|
|
||
| if (eventCount == 1) | ||
| { | ||
| a.Event.Should().Be(Diagnostics.Auditing.AuditEvent.AllNodesDead); | ||
| } | ||
|
|
||
| if (eventCount == 2) | ||
| { | ||
| a.Event.Should().Be(Diagnostics.Auditing.AuditEvent.Resurrection); | ||
| } | ||
|
|
||
| if (eventCount == 3) | ||
| { | ||
| a.Event.Should().Be(Diagnostics.Auditing.AuditEvent.BadRequest); | ||
| audit.Node.FailedAttempts.Should().Be(2); | ||
| audit.Node.IsAlive.Should().BeFalse(); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
83 changes: 83 additions & 0 deletions
83
tests/Elastic.Transport.Tests/Components/TransportClient/RequestInvokerTests.cs
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 |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| // Licensed to Elasticsearch B.V under one or more agreements. | ||
| // Elasticsearch B.V licenses this file to you under the Apache 2.0 License. | ||
| // See the LICENSE file in the project root for more information | ||
|
|
||
| using Xunit; | ||
| using System; | ||
| using FluentAssertions; | ||
| using System.Threading.Tasks; | ||
| using System.Threading; | ||
| using System.Diagnostics; | ||
| using System.Collections.Generic; | ||
| using System.IO; | ||
| using Elastic.Transport.Diagnostics; | ||
| using System.Net.NetworkInformation; | ||
|
|
||
| namespace Elastic.Transport.Tests.Components.TransportClient | ||
| { | ||
| public class RequestInvokerTests | ||
| { | ||
| [Fact] | ||
| public void NoExceptionShouldBeThrown_WhenHttpResponseDoesNotIncludeCloudHeaders() | ||
| { | ||
| // This test validates that if `ProductRegistration.ParseOpenTelemetryAttributesFromApiCallDetails` returns null, | ||
| // no exception is thrown and attributes are not set. | ||
|
|
||
| using var listener = new ActivityListener | ||
| { | ||
| ActivityStarted = _ => { }, | ||
| ActivityStopped = activity => { }, | ||
| ShouldListenTo = activitySource => activitySource.Name == OpenTelemetry.ElasticTransportActivitySourceName, | ||
| Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllData | ||
| }; | ||
| ActivitySource.AddActivityListener(listener); | ||
|
|
||
| var requestInvoker = new HttpRequestInvoker(new TestResponseFactory()); | ||
| var pool = new SingleNodePool(new Uri("http://localhost:9200")); | ||
| var config = new TransportConfiguration(pool, requestInvoker); | ||
| var transport = new DistributedTransport(config); | ||
|
|
||
| var response = transport.Head("/"); | ||
| response.ApiCallDetails.HttpStatusCode.Should().Be(200); | ||
| } | ||
|
|
||
| private sealed class TestResponseFactory : ResponseFactory | ||
| { | ||
| public override TResponse Create<TResponse>( | ||
| Endpoint endpoint, | ||
| BoundConfiguration boundConfiguration, | ||
| PostData postData, | ||
| Exception ex, | ||
| int? statusCode, | ||
| Dictionary<string, IEnumerable<string>> headers, | ||
| Stream responseStream, | ||
| string contentType, | ||
| long contentLength, | ||
| IReadOnlyDictionary<string, ThreadPoolStatistics> threadPoolStats, | ||
| IReadOnlyDictionary<TcpState, int> tcpStats) => CreateResponse<TResponse>(); | ||
|
|
||
| public override Task<TResponse> CreateAsync<TResponse>( | ||
| Endpoint endpoint, | ||
| BoundConfiguration boundConfiguration, | ||
| PostData postData, | ||
| Exception ex, | ||
| int? statusCode, | ||
| Dictionary<string, IEnumerable<string>> headers, | ||
| Stream responseStream, | ||
| string contentType, | ||
| long contentLength, | ||
| IReadOnlyDictionary<string, ThreadPoolStatistics> threadPoolStats, | ||
| IReadOnlyDictionary<TcpState, int> tcpStats, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| var response = CreateResponse<TResponse>(); | ||
| return Task.FromResult(response); | ||
| } | ||
|
|
||
| private static TResponse CreateResponse<TResponse>() where TResponse : TransportResponse, new() => new TResponse | ||
| { | ||
| ApiCallDetails = new() { HttpStatusCode = 200, Uri = new Uri("http://localhost/") } | ||
| }; | ||
| } | ||
| } | ||
| } |