Health check circuit breaker

Signed-off-by: Tomasz Maruszak <[email protected]>
zarusz · Jan 1, 2025 · dfdda2f · dfdda2f
1 parent 90468dc
commit dfdda2f
Show file tree

Hide file tree

Showing 37 changed files with 442 additions and 302 deletions.
diff --git a/README.md b/README.md
@@ -93,6 +93,7 @@ SlimMessageBus is a client façade for message brokers for .NET. It comes with i
 | `.Host.Outbox.Sql`                   | Transactional Outbox using MSSQL                                                                                    | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.Outbox.Sql.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql)                                     |
 | `.Host.Outbox.Sql.DbContext`         | Transactional Outbox using MSSQL with EF DataContext integration                                                    | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.Outbox.Sql.DbContext.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.Sql.DbContext)                 |
 | `.Host.AsyncApi`                     | [AsyncAPI](https://www.asyncapi.com/) specification generation via [Saunter](https://github.com/tehmantra/saunter)  | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.AsyncApi.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.AsyncApi)                                         |
+| `.Host.CircuitBreaker.HealthCheck`   | Consumer circuit breaker based on [health checks](docs/intro.md#health-check-circuit-breaker)                       | [![NuGet](https://img.shields.io/nuget/v/SlimMessageBus.Host.CircuitBreaker.HealthCheck.svg)](https://www.nuget.org/packages/SlimMessageBus.Host.CircuitBreaker.HealthCheck)     |
 
 Typically the application layers (domain model, business logic) only need to depend on `SlimMessageBus` which is the facade, and ultimately the application hosting layer (ASP.NET, Console App, Windows Service) will reference and configure the other packages (`SlimMessageBus.Host.*`) which are the messaging transport providers and additional plugins.
 

diff --git a/docs/NuGet.md b/docs/NuGet.md
@@ -24,5 +24,6 @@ Plugins:
 - Transactional Outbox pattern (SQL, DbContext)
 - Serialization using JSON, Avro, ProtoBuf
 - AsyncAPI specification generation
+- Consumer Circuit Breaker based on Health Checks
 
 Find out more [https://github.com/zarusz/SlimMessageBus](https://github.com/zarusz/SlimMessageBus).
diff --git a/src/Host.Plugin.Properties.xml b/src/Host.Plugin.Properties.xml
@@ -4,7 +4,7 @@
   <Import Project="Common.NuGet.Properties.xml" />
 
   <PropertyGroup>
-    <Version>3.0.0-rc901</Version>
+    <Version>3.0.0-rc902</Version>
   </PropertyGroup>
 
 </Project>
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/Consumers/AddConsumer.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/Consumers/AddConsumer.cs
@@ -1,17 +1,10 @@
 namespace Sample.CircuitBreaker.HealthCheck.Consumers;
 
-public class AddConsumer : IConsumer<Add>
-{
-    private readonly ILogger<AddConsumer> _logger;
-
-    public AddConsumer(ILogger<AddConsumer> logger)
-    {
-        _logger = logger;
-    }
-
+public class AddConsumer(ILogger<AddConsumer> logger) : IConsumer<Add>
+{
     public Task OnHandle(Add message, CancellationToken cancellationToken)
     {
-        _logger.LogInformation("{A} + {B} = {C}", message.a, message.b, message.a + message.b);
+        logger.LogInformation("{A} + {B} = {C}", message.A, message.B, message.A + message.B);
         return Task.CompletedTask;
     }
 }
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/Consumers/SubtractConsumer.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/Consumers/SubtractConsumer.cs
@@ -1,17 +1,10 @@
 namespace Sample.CircuitBreaker.HealthCheck.Consumers;
 
-public class SubtractConsumer : IConsumer<Subtract>
-{
-    private readonly ILogger<SubtractConsumer> _logger;
-
-    public SubtractConsumer(ILogger<SubtractConsumer> logger)
-    {
-        _logger = logger;
-    }
-
+public class SubtractConsumer(ILogger<SubtractConsumer> logger) : IConsumer<Subtract>
+{
     public Task OnHandle(Subtract message, CancellationToken cancellationToken)
     {
-        _logger.LogInformation("{A} - {B} = {C}", message.a, message.b, message.a - message.b);
+        logger.LogInformation("{A} - {B} = {C}", message.A, message.B, message.A - message.B);
         return Task.CompletedTask;
     }
 }
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/GlobalUsings.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/GlobalUsings.cs
@@ -14,4 +14,4 @@
 global using SlimMessageBus;
 global using SlimMessageBus.Host;
 global using SlimMessageBus.Host.RabbitMQ;
-global using SlimMessageBus.Host.Serialization.SystemTextJson;
+global using SlimMessageBus.Host.Serialization.SystemTextJson;
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/HealthChecks/AddRandomHealthCheck.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/HealthChecks/AddRandomHealthCheck.cs
@@ -1,11 +1,5 @@
 namespace Sample.CircuitBreaker.HealthCheck.HealthChecks;
 
-using Microsoft.Extensions.Logging;
-
-public class AddRandomHealthCheck : RandomHealthCheck
+public class AddRandomHealthCheck(ILogger<AddRandomHealthCheck> logger) : RandomHealthCheck(logger)
 {
-    public AddRandomHealthCheck(ILogger<AddRandomHealthCheck> logger) 
-        : base(logger)
-    {
-    }
 }
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/HealthChecks/RandomHealthCheck.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/HealthChecks/RandomHealthCheck.cs
@@ -2,19 +2,12 @@
 
 using Microsoft.Extensions.Diagnostics.HealthChecks;
 
-public abstract class RandomHealthCheck : IHealthCheck
-{
-    private readonly ILogger _logger;
-
-    protected RandomHealthCheck(ILogger logger)
-    {
-        _logger = logger;
-    }
-
+public abstract class RandomHealthCheck(ILogger logger) : IHealthCheck
+{
     public Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default)
     {
         var value = (HealthStatus)Random.Shared.Next(3);
-        _logger.LogInformation("{HealthCheck} evaluated as {HealthStatus}", this.GetType(), value);
+        logger.LogInformation("{HealthCheck} evaluated as {HealthStatus}", GetType(), value);
         return Task.FromResult(new HealthCheckResult(value, value.ToString()));
     }
 }
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/HealthChecks/SubtractRandomHealthCheck.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/HealthChecks/SubtractRandomHealthCheck.cs
@@ -1,11 +1,5 @@
 namespace Sample.CircuitBreaker.HealthCheck.HealthChecks;
 
-using Microsoft.Extensions.Logging;
-
-public class SubtractRandomHealthCheck : RandomHealthCheck
+public class SubtractRandomHealthCheck(ILogger<SubtractRandomHealthCheck> logger) : RandomHealthCheck(logger)
 {
-    public SubtractRandomHealthCheck(ILogger<SubtractRandomHealthCheck> logger) 
-        : base(logger)
-    {
-    }
 }
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/IntermittentMessagePublisher.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/IntermittentMessagePublisher.cs
@@ -1,27 +1,20 @@
 namespace Sample.CircuitBreaker.HealthCheck;
-public class IntermittentMessagePublisher : BackgroundService
+
+public class IntermittentMessagePublisher(ILogger<IntermittentMessagePublisher> logger, IMessageBus messageBus)
+    : BackgroundService
 {
-    private readonly ILogger _logger;
-    private readonly IMessageBus _messageBus;
-
-    public IntermittentMessagePublisher(ILogger<IntermittentMessagePublisher> logger, IMessageBus messageBus)
-    {
-        _logger = logger;
-        _messageBus = messageBus;
-    }
-
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         while (!stoppingToken.IsCancellationRequested)
         {
             var a = Random.Shared.Next(10);
             var b = Random.Shared.Next(10);
 
-            //_logger.LogInformation("Emitting {A} +- {B} = ?", a, b);
+            logger.LogInformation("Emitting {A} +- {B} = ?", a, b);
 
             await Task.WhenAll(
-                _messageBus.Publish(new Add(a, b)),
-                _messageBus.Publish(new Subtract(a, b)),
+                messageBus.Publish(new Add(a, b), cancellationToken: stoppingToken),
+                messageBus.Publish(new Subtract(a, b), cancellationToken: stoppingToken),
                 Task.Delay(1000, stoppingToken));
         }
     }

diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/Models/Add.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/Models/Add.cs
@@ -1,3 +1,3 @@
 namespace Sample.CircuitBreaker.HealthCheck.Models;
 
-public record Add(int a, int b);
+public record Add(int A, int B);
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/Models/Subtract.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/Models/Subtract.cs
@@ -1,3 +1,3 @@
 namespace Sample.CircuitBreaker.HealthCheck.Models;
 
-public record Subtract(int a, int b);
+public record Subtract(int A, int B);
diff --git a/src/Samples/Sample.CircuitBreaker.HealthCheck/Program.cs b/src/Samples/Sample.CircuitBreaker.HealthCheck/Program.cs
@@ -1,9 +1,10 @@
 namespace Sample.CircuitBreaker.HealthCheck;
+
 using Microsoft.Extensions.Diagnostics.HealthChecks;
 
 using Sample.CircuitBreaker.HealthCheck.HealthChecks;
 
-using SlimMessageBus.Host.CircuitBreaker.HealthCheck.Config;
+using SlimMessageBus.Host.CircuitBreaker.HealthCheck;
 
 public static class Program
 {

diff --git a/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/Config/AbstractConsumerProperties.cs b/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/Config/AbstractConsumerProperties.cs
@@ -0,0 +1,7 @@
+namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck;
+
+static internal class AbstractConsumerProperties
+{
+    static readonly internal ProviderExtensionProperty<bool> IsPaused = new("CircuitBreaker_IsPaused");
+    static readonly internal ProviderExtensionProperty<List<IConsumerCircuitBreaker>?> Breakers = new("CircuitBreaker_Breakers");
+}
diff --git a/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/Config/ConsumerBuilderExtensions.cs b/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/Config/ConsumerBuilderExtensions.cs
@@ -1,4 +1,4 @@
-namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck.Config;
+namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck;
 
 using Microsoft.Extensions.DependencyInjection.Extensions;
 
@@ -32,16 +32,18 @@ public static T PauseOnDegradedHealthCheck<T>(this T builder, params string[] ta
 
     private static void RegisterHealthServices(AbstractConsumerBuilder builder)
     {
-        builder.ConsumerSettings.CircuitBreakers.TryAdd<HealthCheckCircuitBreaker>();
-        builder.PostConfigurationActions.Add(
-            services =>
-            {
-                services.TryAddSingleton<HealthCheckBackgroundService>();
-                services.TryAddEnumerable(ServiceDescriptor.Singleton<IHealthCheckPublisher, HealthCheckBackgroundService>(sp => sp.GetRequiredService<HealthCheckBackgroundService>()));
-                services.TryAdd(ServiceDescriptor.Singleton<IHealthCheckHostBreaker, HealthCheckBackgroundService>(sp => sp.GetRequiredService<HealthCheckBackgroundService>()));
-                services.AddHostedService(sp => sp.GetRequiredService<HealthCheckBackgroundService>());
-
-                services.TryAddSingleton<HealthCheckCircuitBreaker>();
-            });
+        var circuitBreakers = builder.ConsumerSettings.GetOrCreate(ConsumerSettingsProperties.CircuitBreakerTypes, () => []);
+        circuitBreakers.TryAdd<HealthCheckCircuitBreaker>();
+
+        builder.PostConfigurationActions.Add(services =>
+        {
+            services.TryAddSingleton<HealthCheckBackgroundService>();
+            services.TryAddEnumerable(ServiceDescriptor.Singleton<IHealthCheckPublisher, HealthCheckBackgroundService>(sp => sp.GetRequiredService<HealthCheckBackgroundService>()));
+            services.TryAdd(ServiceDescriptor.Singleton<IHealthCheckHostBreaker, HealthCheckBackgroundService>(sp => sp.GetRequiredService<HealthCheckBackgroundService>()));
+            services.AddHostedService(sp => sp.GetRequiredService<HealthCheckBackgroundService>());
+
+            services.TryAddTransient<HealthCheckCircuitBreaker>();
+            services.TryAddEnumerable(ServiceDescriptor.Singleton<IAbstractConsumerInterceptor, HealthCheckCircuitBreakerAbstractConsumerInterceptor>());
+        });
     }
 }
diff --git a/....HealthCheck/Config/SettingsExtensions.cs → ...heck/Config/ConsumerSettingsExtensions.cs b/....HealthCheck/Config/SettingsExtensions.cs → ...heck/Config/ConsumerSettingsExtensions.cs
@@ -1,9 +1,7 @@
 namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck;
 
-static internal class SettingsExtensions
+static internal class ConsumerSettingsExtensions
 {
-    private const string _key = nameof(HealthCheckCircuitBreaker);
-
     public static T PauseOnDegraded<T>(this T consumerSettings, params string[] tags)
         where T : AbstractConsumerSettings
     {
@@ -15,7 +13,6 @@ public static T PauseOnDegraded<T>(this T consumerSettings, params string[] tags
                 dict[tag] = HealthStatus.Degraded;
             }
         }
-
         return consumerSettings;
     }
 
@@ -30,18 +27,9 @@ public static T PauseOnUnhealthy<T>(this T consumerSettings, params string[] tag
                 dict[tag] = HealthStatus.Unhealthy;
             }
         }
-
         return consumerSettings;
     }
 
-    static internal IDictionary<string, HealthStatus> HealthBreakerTags(this AbstractConsumerSettings consumerSettings)
-    {
-        if (!consumerSettings.Properties.TryGetValue(_key, out var rawValue) || rawValue is not IDictionary<string, HealthStatus> value)
-        {
-            value = new Dictionary<string, HealthStatus>();
-            consumerSettings.Properties[_key] = value;
-        }
-
-        return value;
-    }
+    static internal IDictionary<string, HealthStatus> HealthBreakerTags(this AbstractConsumerSettings consumerSettings)
+        => consumerSettings.GetOrCreate(ConsumerSettingsProperties.HealthStatusTags, () => []);
 }
diff --git a/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/Config/ConsumerSettingsProperties.cs b/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/Config/ConsumerSettingsProperties.cs
@@ -0,0 +1,11 @@
+namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck;
+
+static internal class ConsumerSettingsProperties
+{
+    /// <summary>
+    /// <see cref="IConsumerCircuitBreaker"/> to be used with the consumer.
+    /// </summary>
+    static readonly internal ProviderExtensionProperty<TypeCollection<IConsumerCircuitBreaker>> CircuitBreakerTypes = new("CircuitBreakerTypes");
+
+    static readonly internal ProviderExtensionProperty<Dictionary<string, HealthStatus>> HealthStatusTags = new("CircuitBreaker_HealthStatusTags");
+}
diff --git a/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/HealthCheckBackgroundService.cs b/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/HealthCheckBackgroundService.cs
@@ -38,10 +38,8 @@ public async Task PublishAsync(HealthReport report, CancellationToken cancellati
         }
     }
 
-    public Task StartAsync(CancellationToken cancellationToken)
-    {
-        return Task.CompletedTask;
-    }
+    public Task StartAsync(CancellationToken cancellationToken)
+        => Task.CompletedTask;
 
     public Task StopAsync(CancellationToken cancellationToken)
     {

diff --git a/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/HealthCheckCircuitBreaker.cs b/src/SlimMessageBus.Host.CircuitBreaker.HealthCheck/HealthCheckCircuitBreaker.cs
@@ -1,5 +1,5 @@
-namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck;
-
+namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck;
+
 internal sealed class HealthCheckCircuitBreaker : IConsumerCircuitBreaker
 {
     private readonly IEnumerable<AbstractConsumerSettings> _settings;

diff --git a/...s.Host.CircuitBreaker.HealthCheck/HealthCheckCircuitBreakerAbstractConsumerInterceptor.cs b/...s.Host.CircuitBreaker.HealthCheck/HealthCheckCircuitBreakerAbstractConsumerInterceptor.cs
@@ -0,0 +1,92 @@
+namespace SlimMessageBus.Host.CircuitBreaker.HealthCheck;
+
+internal sealed class HealthCheckCircuitBreakerAbstractConsumerInterceptor : IAbstractConsumerInterceptor
+{
+    public int Order => 100;
+
+    public async Task<bool> CanStart(AbstractConsumer consumer)
+    {
+        var breakerTypes = consumer.Settings.SelectMany(x => x.GetOrDefault(ConsumerSettingsProperties.CircuitBreakerTypes, [])).ToHashSet();
+        if (breakerTypes.Count == 0)
+        {
+            // no breakers, allow to pass
+            return true;
+        }
+
+        var breakers = consumer.GetOrCreate(AbstractConsumerProperties.Breakers, () => [])!;
+
+        async Task BreakerChanged(Circuit state)
+        {
+            if (!consumer.IsStarted)
+            {
+                return;
+            }
+
+            var isPaused = consumer.IsPaused();
+            var shouldPause = state == Circuit.Closed || breakers.Exists(x => x.State == Circuit.Closed);
+            if (shouldPause != isPaused)
+            {
+                var settings = consumer.Settings.Count > 0 ? consumer.Settings[0] : null;
+                var path = settings?.Path ?? "[unknown path]";
+                var bus = settings?.MessageBusSettings?.Name ?? "default";
+                if (shouldPause)
+                {
+                    consumer.Logger.LogWarning("Circuit breaker tripped for '{Path}' on '{Bus}' bus. Consumer paused.", path, bus);
+                    await consumer.DoStop().ConfigureAwait(false);
+                }
+                else
+                {
+                    consumer.Logger.LogInformation("Circuit breaker restored for '{Path}' on '{Bus}' bus. Consumer resumed.", path, bus);
+                    await consumer.DoStart().ConfigureAwait(false);
+                }
+                consumer.SetIsPaused(shouldPause);
+            }
+        }
+
+        var sp = consumer.Settings.Select(x => x.MessageBusSettings.ServiceProvider).First(x => x != null);
+        foreach (var breakerType in breakerTypes)
+        {
+            var breaker = (IConsumerCircuitBreaker)ActivatorUtilities.CreateInstance(sp, breakerType, consumer.Settings);
+            breakers.Add(breaker);
+
+            await breaker.Subscribe(BreakerChanged);
+        }
+
+        var isPaused = breakers.Exists(x => x.State == Circuit.Closed);
+        consumer.SetIsPaused(isPaused);
+        return !isPaused;
+    }
+
+    public async Task<bool> CanStop(AbstractConsumer consumer)
+    {
+        var breakers = consumer.GetOrDefault(AbstractConsumerProperties.Breakers, null);
+        if (breakers == null || breakers.Count == 0)
+        {
+            // no breakers, allow to pass
+            return true;
+        }
+
+        foreach (var breaker in breakers)
+        {
+            breaker.Unsubscribe();
+
+            if (breaker is IAsyncDisposable asyncDisposable)
+            {
+                await asyncDisposable.DisposeAsync();
+            }
+            else if (breaker is IDisposable disposable)
+            {
+                disposable.Dispose();
+            }
+        }
+        breakers.Clear();
+
+        return !consumer.IsPaused();
+    }
+}
+
+internal static class AbstractConsumerExtensions
+{
+    public static bool IsPaused(this AbstractConsumer consumer) => consumer.GetOrDefault(AbstractConsumerProperties.IsPaused, false);
+    public static void SetIsPaused(this AbstractConsumer consumer, bool isPaused) => AbstractConsumerProperties.IsPaused.Set(consumer, isPaused);
+}