-
Notifications
You must be signed in to change notification settings - Fork 1.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'ThreeMammals:develop' into feat_enableHttp2
- Loading branch information
Showing
25 changed files
with
1,970 additions
and
1,378 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
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,115 @@ | ||
| using Ocelot.Logging; | ||
|
|
||
| namespace Ocelot.Infrastructure.DesignPatterns; | ||
|
|
||
| /// <summary> | ||
| /// Basic <seealso href="https://www.bing.com/search?q=Retry+pattern">Retry pattern</seealso> for stabilizing integrated services. | ||
| /// </summary> | ||
| /// <remarks>Docs: | ||
| /// <list type="bullet"> | ||
| /// <item><see href="https://learn.microsoft.com/en-us/azure/architecture/patterns/retry">Microsoft Learn | Retry pattern</see></item> | ||
| /// </list> | ||
| /// </remarks> | ||
| public static class Retry | ||
| { | ||
| public const int DefaultRetryTimes = 3; | ||
| public const int DefaultWaitTimeMilliseconds = 25; | ||
|
|
||
| private static string GetMessage<T>(T operation, int retryNo, string message) | ||
| where T : Delegate | ||
| => $"Ocelot {nameof(Retry)} strategy for the operation of '{operation.GetType()}' type -> {nameof(Retry)} No {retryNo}: {message}"; | ||
|
|
||
| /// <summary> | ||
| /// Retry a synchronous operation when an exception occurs or predicate is true, then delay and retry again. | ||
| /// </summary> | ||
| /// <typeparam name="TResult">Type of the result of the sync operation.</typeparam> | ||
| /// <param name="operation">Required Func-delegate of the operation.</param> | ||
| /// <param name="predicate">Predicate to check, optionally.</param> | ||
| /// <param name="retryTimes">Number of retries.</param> | ||
| /// <param name="waitTime">Waiting time in milliseconds.</param> | ||
| /// <param name="logger">Concrete logger from upper context.</param> | ||
| /// <returns>A <typeparamref name="TResult"/> value as the result of the sync operation.</returns> | ||
| public static TResult Operation<TResult>( | ||
| Func<TResult> operation, | ||
| Predicate<TResult> predicate = null, | ||
| int retryTimes = DefaultRetryTimes, int waitTime = DefaultWaitTimeMilliseconds, | ||
| IOcelotLogger logger = null) | ||
| { | ||
| for (int n = 1; n < retryTimes; n++) | ||
| { | ||
| TResult result; | ||
| try | ||
| { | ||
| result = operation.Invoke(); | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| logger?.LogError(() => GetMessage(operation, n, $"Caught exception of the {e.GetType()} type -> Message: {e.Message}."), e); | ||
| Thread.Sleep(waitTime); | ||
| continue; // the result is unknown, so continue to retry | ||
| } | ||
|
|
||
| // Apply predicate for known result | ||
| if (predicate?.Invoke(result) == true) | ||
| { | ||
| logger?.LogWarning(() => GetMessage(operation, n, $"The predicate has identified erroneous state in the returned result. For further details, implement logging of the result's value or properties within the predicate method.")); | ||
| Thread.Sleep(waitTime); | ||
| continue; // on erroneous state | ||
| } | ||
|
|
||
| // Happy path | ||
| return result; | ||
| } | ||
|
|
||
| // Last retry should generate native exception or other erroneous state(s) | ||
| logger?.LogDebug(() => GetMessage(operation, retryTimes, $"Retrying lastly...")); | ||
| return operation.Invoke(); // also final result must be analyzed in the upper context | ||
| } | ||
|
|
||
| /// <summary> | ||
| /// Retry an asynchronous operation when an exception occurs or predicate is true, then delay and retry again. | ||
| /// </summary> | ||
| /// <typeparam name="TResult">Type of the result of the async operation.</typeparam> | ||
| /// <param name="operation">Required Func-delegate of the operation.</param> | ||
| /// <param name="predicate">Predicate to check, optionally.</param> | ||
| /// <param name="retryTimes">Number of retries.</param> | ||
| /// <param name="waitTime">Waiting time in milliseconds.</param> | ||
| /// <param name="logger">Concrete logger from upper context.</param> | ||
| /// <returns>A <typeparamref name="TResult"/> value as the result of the async operation.</returns> | ||
| public static async Task<TResult> OperationAsync<TResult>( | ||
| Func<Task<TResult>> operation, // required operation delegate | ||
| Predicate<TResult> predicate = null, // optional retry predicate for the result | ||
| int retryTimes = DefaultRetryTimes, int waitTime = DefaultWaitTimeMilliseconds, // retrying options | ||
| IOcelotLogger logger = null) // static injections | ||
| { | ||
| for (int n = 1; n < retryTimes; n++) | ||
| { | ||
| TResult result; | ||
| try | ||
| { | ||
| result = await operation?.Invoke(); | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| logger?.LogError(() => GetMessage(operation, n, $"Caught exception of the {e.GetType()} type -> Message: {e.Message}."), e); | ||
| await Task.Delay(waitTime); | ||
| continue; // the result is unknown, so continue to retry | ||
| } | ||
|
|
||
| // Apply predicate for known result | ||
| if (predicate?.Invoke(result) == true) | ||
| { | ||
| logger?.LogWarning(() => GetMessage(operation, n, $"The predicate has identified erroneous state in the returned result. For further details, implement logging of the result's value or properties within the predicate method.")); | ||
| await Task.Delay(waitTime); | ||
| continue; // on erroneous state | ||
| } | ||
|
|
||
| // Happy path | ||
| return result; | ||
| } | ||
|
|
||
| // Last retry should generate native exception or other erroneous state(s) | ||
| logger?.LogDebug(() => GetMessage(operation, retryTimes, $"Retrying lastly...")); | ||
| return await operation?.Invoke(); // also final result must be analyzed in the upper context | ||
| } | ||
| } |
116 changes: 57 additions & 59 deletions
116
src/Ocelot/LoadBalancer/LoadBalancers/CookieStickySessions.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 |
|---|---|---|
| @@ -1,85 +1,83 @@ | ||
| using Microsoft.AspNetCore.Http; | ||
| using Ocelot.Infrastructure; | ||
| using Ocelot.Middleware; | ||
| using Ocelot.Responses; | ||
| using Ocelot.Values; | ||
|
|
||
| namespace Ocelot.LoadBalancer.LoadBalancers | ||
| namespace Ocelot.LoadBalancer.LoadBalancers; | ||
|
|
||
| public class CookieStickySessions : ILoadBalancer | ||
| { | ||
| public class CookieStickySessions : ILoadBalancer | ||
| private readonly int _keyExpiryInMs; | ||
| private readonly string _cookieName; | ||
| private readonly ILoadBalancer _loadBalancer; | ||
| private readonly IBus<StickySession> _bus; | ||
|
|
||
| private static readonly object Locker = new(); | ||
| private static readonly Dictionary<string, StickySession> Stored = new(); // TODO Inject instead of static sharing | ||
|
|
||
| public CookieStickySessions(ILoadBalancer loadBalancer, string cookieName, int keyExpiryInMs, IBus<StickySession> bus) | ||
| { | ||
| private readonly int _keyExpiryInMs; | ||
| private readonly string _key; | ||
| private readonly ILoadBalancer _loadBalancer; | ||
| private readonly ConcurrentDictionary<string, StickySession> _stored; | ||
| private readonly IBus<StickySession> _bus; | ||
| private readonly object _lock = new(); | ||
| _bus = bus; | ||
| _cookieName = cookieName; | ||
| _keyExpiryInMs = keyExpiryInMs; | ||
| _loadBalancer = loadBalancer; | ||
| _bus.Subscribe(CheckExpiry); | ||
| } | ||
|
|
||
| public CookieStickySessions(ILoadBalancer loadBalancer, string key, int keyExpiryInMs, IBus<StickySession> bus) | ||
| private void CheckExpiry(StickySession sticky) | ||
| { | ||
| // TODO Get test coverage for this | ||
| lock (Locker) | ||
| { | ||
| _bus = bus; | ||
| _key = key; | ||
| _keyExpiryInMs = keyExpiryInMs; | ||
| _loadBalancer = loadBalancer; | ||
| _stored = new ConcurrentDictionary<string, StickySession>(); | ||
| _bus.Subscribe(ss => | ||
| if (!Stored.TryGetValue(sticky.Key, out var session) || session.Expiry >= DateTime.UtcNow) | ||
| { | ||
| //todo - get test coverage for this. | ||
| if (_stored.TryGetValue(ss.Key, out var stickySession)) | ||
| { | ||
| lock (_lock) | ||
| { | ||
| if (stickySession.Expiry < DateTime.UtcNow) | ||
| { | ||
| _stored.TryRemove(stickySession.Key, out _); | ||
| _loadBalancer.Release(stickySession.HostAndPort); | ||
| } | ||
| } | ||
| } | ||
| }); | ||
| return; | ||
| } | ||
|
|
||
| Stored.Remove(session.Key); | ||
| _loadBalancer.Release(session.HostAndPort); | ||
| } | ||
| } | ||
|
|
||
| public async Task<Response<ServiceHostAndPort>> Lease(HttpContext httpContext) | ||
| public Task<Response<ServiceHostAndPort>> Lease(HttpContext httpContext) | ||
| { | ||
| var route = httpContext.Items.DownstreamRoute(); | ||
| var serviceName = route.LoadBalancerKey; | ||
| var cookie = httpContext.Request.Cookies[_cookieName]; | ||
| var key = $"{serviceName}:{cookie}"; // strong key name because of static store | ||
| lock (Locker) | ||
| { | ||
| var key = httpContext.Request.Cookies[_key]; | ||
|
|
||
| lock (_lock) | ||
| if (!string.IsNullOrEmpty(key) && Stored.TryGetValue(key, out StickySession cached)) | ||
| { | ||
| if (!string.IsNullOrEmpty(key) && _stored.ContainsKey(key)) | ||
| { | ||
| var cached = _stored[key]; | ||
|
|
||
| var updated = new StickySession(cached.HostAndPort, DateTime.UtcNow.AddMilliseconds(_keyExpiryInMs), key); | ||
|
|
||
| _stored[key] = updated; | ||
|
|
||
| _bus.Publish(updated, _keyExpiryInMs); | ||
|
|
||
| return new OkResponse<ServiceHostAndPort>(updated.HostAndPort); | ||
| } | ||
| var updated = new StickySession(cached.HostAndPort, DateTime.UtcNow.AddMilliseconds(_keyExpiryInMs), key); | ||
| Update(key, updated); | ||
| return Task.FromResult<Response<ServiceHostAndPort>>(new OkResponse<ServiceHostAndPort>(updated.HostAndPort)); | ||
| } | ||
|
|
||
| var next = await _loadBalancer.Lease(httpContext); | ||
|
|
||
| // There is no value in the store, so lease it now! | ||
| var next = _loadBalancer.Lease(httpContext).GetAwaiter().GetResult(); // unfortunately the operation must be synchronous | ||
| if (next.IsError) | ||
| { | ||
| return new ErrorResponse<ServiceHostAndPort>(next.Errors); | ||
| } | ||
|
|
||
| lock (_lock) | ||
| { | ||
| if (!string.IsNullOrEmpty(key) && !_stored.ContainsKey(key)) | ||
| { | ||
| var ss = new StickySession(next.Data, DateTime.UtcNow.AddMilliseconds(_keyExpiryInMs), key); | ||
| _stored[key] = ss; | ||
| _bus.Publish(ss, _keyExpiryInMs); | ||
| } | ||
| return Task.FromResult<Response<ServiceHostAndPort>>(new ErrorResponse<ServiceHostAndPort>(next.Errors)); | ||
| } | ||
|
|
||
| return new OkResponse<ServiceHostAndPort>(next.Data); | ||
| var ss = new StickySession(next.Data, DateTime.UtcNow.AddMilliseconds(_keyExpiryInMs), key); | ||
| Update(key, ss); | ||
| return Task.FromResult<Response<ServiceHostAndPort>>(new OkResponse<ServiceHostAndPort>(next.Data)); | ||
| } | ||
| } | ||
|
|
||
| public void Release(ServiceHostAndPort hostAndPort) | ||
| protected void Update(string key, StickySession value) | ||
| { | ||
| lock (Locker) | ||
| { | ||
| Stored[key] = value; | ||
| _bus.Publish(value, _keyExpiryInMs); | ||
| } | ||
| } | ||
|
|
||
| public void Release(ServiceHostAndPort hostAndPort) | ||
| { | ||
| } | ||
| } |
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.