Adding Akka.Persistence health checks

Directory.Build.props

@@ -118,7 +118,7 @@ A basic lambda function can handle it - the only reason this function is as comp
     <TestSdkVersion>17.11.1</TestSdkVersion>
     <CoverletVersion>6.0.3</CoverletVersion>
      <XunitRunneVisualstudio>3.1.4</XunitRunneVisualstudio>
-    <AkkaVersion>1.5.50</AkkaVersion>
+    <AkkaVersion>1.5.51-alpha-1759266431</AkkaVersion>
     <MicrosoftExtensionsVersion>[6.0.0,)</MicrosoftExtensionsVersion>
     <SystemTextJsonVersion>[6.0.10,)</SystemTextJsonVersion>
   </PropertyGroup>

nuget.config

@@ -4,8 +4,13 @@
     <!--To inherit the global NuGet package sources remove the <clear/> line below -->
     <clear />
     <add key="nuget.org" value="https://api.nuget.org/v3/index.json" />
+    <add key="testlab" value="https://nuget.testlab.petabridge.net/v3/index.json"/>
   </packageSources>
   <packageSourceMapping>
+    <packageSource key="testlab">
+      <package pattern="Akka"/>
+      <package pattern="Akka.*"/>
+    </packageSource>
     <packageSource key="nuget.org">
       <package pattern="*" />
     </packageSource>

src/Akka.Persistence.Hosting/AkkaPersistenceHostingExtensions.cs

@@ -6,6 +6,7 @@
 using Akka.Persistence.Journal;
 using Akka.Util;
 using Akka.Actor;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
 
 #nullable enable
 namespace Akka.Persistence.Hosting
@@ -37,13 +38,29 @@ public sealed class AkkaPersistenceJournalBuilder
         internal readonly AkkaConfigurationBuilder Builder;
         internal readonly Dictionary<Type, HashSet<string>> Bindings = new Dictionary<Type, HashSet<string>>();
         internal readonly Dictionary<string, Type> Adapters = new Dictionary<string, Type>();
+        internal AkkaHealthCheckRegistration? HealthCheckRegistration = null;
 
         public AkkaPersistenceJournalBuilder(string journalId, AkkaConfigurationBuilder builder)
         {
             JournalId = journalId;
             Builder = builder;
         }
 
+        /// <summary>
+        /// Uses the built-in journal health check on the Akka.Persistence.Journal.
+        /// </summary>
+        /// <param name="unHealthyStatus">Default status to return when the plugin reports <see cref="PersistenceHealthStatus.Unhealthy"/>
+        /// or <see cref="PersistenceHealthStatus.Degraded"/>. Defaults to degraded.</param>
+        /// <param name="name">Optional name to add to the health check.</param>
+        /// <returns></returns>
+        public AkkaPersistenceJournalBuilder WithHealthCheck(HealthStatus unHealthyStatus = HealthStatus.Degraded,
+            string? name = null)
+        {
+            var registration = AddHealthCheck(name, unHealthyStatus);
+            HealthCheckRegistration = registration;
+            return this;
+        }
+
         public AkkaPersistenceJournalBuilder AddEventAdapter<TAdapter>(string eventAdapterName,
             IEnumerable<Type> boundTypes) where TAdapter : IEventAdapter
         {
@@ -79,6 +96,16 @@ private void AddAdapter<TAdapter>(string eventAdapterName, IEnumerable<Type> bou
             }
         }
 
+        private AkkaHealthCheckRegistration AddHealthCheck(string? name, HealthStatus unHealthyStatus)
+        {
+            var registration = new AkkaHealthCheckRegistration(
+                name ?? $"Akka.Persistence.Journal.{JournalId}",
+                new JournalHealthCheck(JournalId),
+                unHealthyStatus,
+                ["akka", "persistence", "journal"]);
+            return registration;
+        }
+
         /// <summary>
         /// INTERNAL API - Builds the HOCON and then injects it.
         /// </summary>
@@ -98,6 +125,10 @@ internal void Build()
             var finalHocon = ConfigurationFactory.ParseString(adapters.ToString())
                 .WithFallback(Persistence.DefaultConfig()); // add the default config as a fallback
             Builder.AddHocon(finalHocon, HoconAddMode.Prepend);
+
+            // add the health checks if specified
+            if(HealthCheckRegistration != null)
+                Builder.WithHealthCheck(HealthCheckRegistration);
         }
 
         internal void AppendAdapters(StringBuilder sb)

src/Akka.Persistence.Hosting/HealthChecks.cs

@@ -0,0 +1,64 @@
+using System.Threading;
+using System.Threading.Tasks;
+using Akka.Hosting;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
+
+namespace Akka.Persistence.Hosting;
+
+internal static class HealthCheckExt
+{
+    public static HealthCheckResult ToHealthCheckResult(this PersistenceHealthCheckResult persistenceHealthStatus)
+        => new(persistenceHealthStatus.Status.ToHealthStatus(), persistenceHealthStatus.Description,
+            persistenceHealthStatus.Exception, persistenceHealthStatus.Data);
+
+    public static HealthStatus ToHealthStatus(this PersistenceHealthStatus persistenceHealthStatus) => persistenceHealthStatus switch
+    {
+        PersistenceHealthStatus.Healthy => HealthStatus.Healthy,
+        PersistenceHealthStatus.Degraded => HealthStatus.Degraded,
+        _ => HealthStatus.Unhealthy
+    };
+}
+
+/// <summary>
+/// INTERNAL API
+///
+/// Leverages internal Akka.Persistence APIs to perform a health check on a journal.
+/// </summary>
+internal sealed class JournalHealthCheck : IAkkaHealthCheck
+{
+    private readonly string _journalPluginId;
+
+    public JournalHealthCheck(string journalPluginId)
+    {
+        _journalPluginId = journalPluginId;
+    }
+
+    public async Task<HealthCheckResult> CheckHealthAsync(AkkaHealthCheckContext context, CancellationToken cancellationToken = default)
+    {
+        var persistence = Persistence.Instance.Apply(context.ActorSystem);
+        var journalResult = await persistence.CheckJournalHealthAsync(_journalPluginId, cancellationToken);
+        return journalResult.ToHealthCheckResult();
+    }
+}
+
+/// <summary>
+/// INTERNAL API
+///
+/// Leverages internal Akka.Persistence APIs to perform a health check on the snapshot store.
+/// </summary>
+internal sealed class SnapshotStoreHealthCheck : IAkkaHealthCheck
+{
+    private readonly string _snapshotStorePluginId;
+
+    public SnapshotStoreHealthCheck(string snapshotStorePluginId)
+    {
+        _snapshotStorePluginId = snapshotStorePluginId;
+    }
+
+    public async Task<HealthCheckResult> CheckHealthAsync(AkkaHealthCheckContext context, CancellationToken cancellationToken = default)
+    {
+        var persistence = Persistence.Instance.Apply(context.ActorSystem);
+        var ssResult = await persistence.CheckSnapshotStoreHealthAsync(_snapshotStorePluginId, cancellationToken);
+        return ssResult.ToHealthCheckResult();
+    }
+}