SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Add HTTP resiliency support using Polly and pipeline builder

Browse source 
Introduce configurable resiliency mechanisms for HTTP requests, including retries, circuit breakers, and timeouts, leveraging Microsoft.Extensions.Resilience and Polly. Refactor `SendHttpRequestBase` to include an `EnableResiliency` input and encapsulate resiliency logic in a dedicated pipeline. Update project references to include necessary dependencies.
This commit is contained in:
Sipke Schoorstra 2025-02-24 22:30:56 +01:00
parent a24ca0435f
commit 6ba50020fe
No known key found for this signature in database
GPG key ID: 5C10502B28A4268F
2 changed files with 86 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
src/modules/Elsa.Http/Activities/SendHttpRequestBase.cs
View file

@ -1,3 +1,4 @@
using System.Net;
using System.Net.Http.Headers;
using Elsa.Extensions;
using Elsa.Http.ContentWriters;
@ -7,6 +8,7 @@ using Elsa.Workflows.Attributes;
using Elsa.Workflows.UIHints;
using Elsa.Workflows.Models;
using Microsoft.Extensions.Logging;
using Polly;
using HttpHeaders = Elsa.Http.Models.HttpHeaders;
namespace Elsa.Http;
@ -25,8 +27,7 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
/// <summary>
/// The URL to send the request to.
/// </summary>
[Input]
public Input<Uri?> Url { get; set; } = default!;
[Input] public Input<Uri?> Url { get; set; } = default!;
/// <summary>
/// The HTTP method to use when sending the request.
@ -81,6 +82,11 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
)]
public Input<HttpHeaders?> RequestHeaders { get; set; } = new(new HttpHeaders());
/// <summary>
/// Indicates whether resiliency mechanisms should be enabled for the HTTP request.
/// </summary>
public Input<bool> EnableResiliency { get; set; } = default!;
/// <summary>
/// The HTTP response status code
/// </summary>
@ -122,15 +128,16 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
private async Task TrySendAsync(ActivityExecutionContext context)
{
var request = PrepareRequest(context);
var logger = (ILogger)context.GetRequiredService(typeof(ILogger<>).MakeGenericType(GetType()));
var httpClientFactory = context.GetRequiredService<IHttpClientFactory>();
var httpClient = httpClientFactory.CreateClient(nameof(SendHttpRequestBase));
var cancellationToken = context.CancellationToken;
var resiliencyEnabled = EnableResiliency.GetOrDefault(context, () => false);
try
{
var response = await httpClient.SendAsync(request, cancellationToken);
var response = await SendRequestAsync();
var parsedContent = await ParseContentAsync(context, response);
var statusCode = (int)response.StatusCode;
var responseHeaders = new HttpHeaders(response.Headers);
@ -147,7 +154,7 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
logger.LogWarning(e, "An error occurred while sending an HTTP request");
context.AddExecutionLogEntry("Error", e.Message, payload: new
{
StackTrace = e.StackTrace
e.StackTrace
});
context.JournalData.Add("Error", e.Message);
await HandleRequestExceptionAsync(context, e);
@ -157,11 +164,30 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
logger.LogWarning(e, "An error occurred while sending an HTTP request");
context.AddExecutionLogEntry("Error", e.Message, payload: new
{
StackTrace = e.StackTrace
e.StackTrace
});
context.JournalData.Add("Cancelled", true);
await HandleTaskCanceledExceptionAsync(context, e);
}
return;
async Task<HttpResponseMessage> SendRequestAsync()
{
if (resiliencyEnabled)
{
var pipeline = BuildResiliencyPipeline(context);
return await pipeline.ExecuteAsync(async ct => await SendRequestAsyncCore(ct), cancellationToken);
}
return await SendRequestAsyncCore();
}
async Task<HttpResponseMessage> SendRequestAsyncCore(CancellationToken ct = default)
{
var request = PrepareRequest(context);
return await httpClient.SendAsync(request, ct);
}
}
private async Task<object?> ParseContentAsync(ActivityExecutionContext context, HttpResponseMessage httpResponse)
@ -195,7 +221,7 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
{
var method = Method.GetOrDefault(context) ?? "GET";
var url = Url.Get(context);
var request = new HttpRequestMessage(new HttpMethod(method), url);
var request = new HttpRequestMessage(new(method), url);
var headers = context.GetHeaders(RequestHeaders);
var authorization = Authorization.GetOrDefault(context);
var addAuthorizationWithoutValidation = DisableAuthorizationHeaderValidation.GetOrDefault(context);
@ -218,7 +244,7 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
var factory = SelectContentWriter(contentType, factories);
request.Content = factory.CreateHttpContent(content, contentType);
}
return request;
}
@ -230,4 +256,45 @@ public abstract class SendHttpRequestBase : Activity<HttpResponseMessage>
var parsedContentType = new System.Net.Mime.ContentType(contentType);
return factories.FirstOrDefault(httpContentFactory => httpContentFactory.SupportedContentTypes.Any(c => c == parsedContentType.MediaType)) ?? new JsonContentFactory();
}
private ResiliencePipeline<HttpResponseMessage> BuildResiliencyPipeline(ActivityExecutionContext context)
{
var pipelineBuilder = new ResiliencePipelineBuilder<HttpResponseMessage>()
.AddRetry(new()
{
ShouldHandle = new PredicateBuilder<HttpResponseMessage>()
.Handle<TimeoutException>() // Specific timeout exception
.Handle<HttpRequestException>(ex => IsTransientStatusCode(ex.StatusCode)) // Network errors or transient HTTP codes
.HandleResult(response => IsTransientStatusCode(response.StatusCode)),
MaxRetryAttempts = 3,
Delay = TimeSpan.FromSeconds(Math.Min(Random.Shared.NextDouble() * 2, 8)), // Jittered delay capped at 8 secs
BackoffType = DelayBackoffType.Exponential
})
.AddCircuitBreaker(new()
{
FailureRatio = 0.5,
SamplingDuration = TimeSpan.FromSeconds(30),
MinimumThroughput = 10,
BreakDuration = TimeSpan.FromSeconds(60)
})
.AddTimeout(TimeSpan.FromSeconds(60)); // Outer timeout
return pipelineBuilder.Build();
}
// Helper method to identify transient status codes.
private static bool IsTransientStatusCode(HttpStatusCode? statusCode)
{
if (!statusCode.HasValue) return true; // No status code (e.g., network failure) is worth retrying
return statusCode switch
{
HttpStatusCode.RequestTimeout => true, // 408
HttpStatusCode.TooManyRequests => true, // 429
HttpStatusCode.InternalServerError => true, // 500
HttpStatusCode.BadGateway => true, // 502
HttpStatusCode.ServiceUnavailable => true, // 503
HttpStatusCode.GatewayTimeout => true, // 504
_ => false
};
}
}

20
src/modules/Elsa.Http/Elsa.Http.csproj
View file

@ -8,20 +8,22 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="FluentStorage" />
<PackageReference Include="FluentStorage"/>
<PackageReference Include="Microsoft.Extensions.Http.Resilience"/>
<PackageReference Include="Microsoft.Extensions.Resilience"/>
</ItemGroup>
<!--Overridden for vulnaribility reasons with dependencies referencing older versions.-->
<ItemGroup>
<PackageReference Include="System.Text.Json" VersionOverride="$(SystemTextJsonVersion)" />
<PackageReference Include="System.Text.Json" VersionOverride="$(SystemTextJsonVersion)"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Elsa.Liquid\Elsa.Liquid.csproj" />
<ProjectReference Include="..\Elsa.SasTokens\Elsa.SasTokens.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Management\Elsa.Workflows.Management.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
<ProjectReference Include="..\Elsa.JavaScript\Elsa.JavaScript.csproj" />
<ProjectReference Include="..\Elsa.Liquid\Elsa.Liquid.csproj"/>
<ProjectReference Include="..\Elsa.SasTokens\Elsa.SasTokens.csproj"/>
<ProjectReference Include="..\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj"/>
<ProjectReference Include="..\Elsa.Workflows.Management\Elsa.Workflows.Management.csproj"/>
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj"/>
<ProjectReference Include="..\Elsa.JavaScript\Elsa.JavaScript.csproj"/>
</ItemGroup>
</Project>