LNK-2883: Added Kafka health check to .NET services

DotNet/Account/Program.cs

@@ -21,6 +21,7 @@
 using LantanaGroup.Link.Shared.Application.Extensions;
 using LantanaGroup.Link.Shared.Application.Extensions.ExternalServices;
 using LantanaGroup.Link.Shared.Application.Extensions.Security;
+using LantanaGroup.Link.Shared.Application.Health;
 using LantanaGroup.Link.Shared.Application.Middleware;
 using LantanaGroup.Link.Shared.Application.Models.Configs;
 using LantanaGroup.Link.Shared.Application.Services;
@@ -188,8 +189,11 @@ static void RegisterServices(WebApplicationBuilder builder)
     builder.Services.AddScoped<IRoleRepository, RoleRepository>();
 
     //Add health checks
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, AccountConstants.ServiceName).GetHealthCheckOptions();
+
     builder.Services.AddHealthChecks()
-        .AddCheck<DatabaseHealthCheck>("Database");
+        .AddCheck<DatabaseHealthCheck>("Database")
+        .AddKafka(kafkaHealthOptions);
 
     // Add tenant API service
     builder.Services.AddHttpClient();

DotNet/Audit/Program.cs

@@ -42,6 +42,7 @@
 using LantanaGroup.Link.Audit.Domain.Managers;
 using LantanaGroup.Link.Shared.Application.Error.Handlers;
 using LantanaGroup.Link.Audit.Application.Services;
+using LantanaGroup.Link.Shared.Application.Health;
 
 var builder = WebApplication.CreateBuilder(args);
 
@@ -181,8 +182,11 @@ static void RegisterServices(WebApplicationBuilder builder)
     }
 
     //Add health checks
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, AuditConstants.ServiceName).GetHealthCheckOptions();
+
     builder.Services.AddHealthChecks()
-        .AddCheck<DatabaseHealthCheck>("Database");
+        .AddCheck<DatabaseHealthCheck>("Database")
+        .AddKafka(kafkaHealthOptions);
 
     //configure CORS
     builder.Services.AddLinkCorsService(options => {

DotNet/Census/Program.cs

@@ -44,6 +44,7 @@
 using LantanaGroup.Link.Census.Domain.Managers;
 using LantanaGroup.Link.Shared.Application.Utilities;
 using LantanaGroup.Link.Shared.Application.Listeners;
+using LantanaGroup.Link.Shared.Application.Health;
 
 var builder = WebApplication.CreateBuilder(args);
 
@@ -183,8 +184,11 @@ static void RegisterServices(WebApplicationBuilder builder)
     });
 
     //Add health checks
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, CensusConstants.ServiceName).GetHealthCheckOptions();
+
     builder.Services.AddHealthChecks()
-        .AddCheck<DatabaseHealthCheck>("Database");
+        .AddCheck<DatabaseHealthCheck>("Database")
+        .AddKafka(kafkaHealthOptions);
 
     builder.Services.AddControllers().AddJsonOptions(options =>
     {

DotNet/DataAcquisition/Program.cs

@@ -51,6 +51,7 @@
 using LantanaGroup.Link.Shared.Application.Utilities;
 using Hl7.Fhir.Serialization;
 using Hl7.Fhir.Model;
+using LantanaGroup.Link.Shared.Application.Health;
 
 var builder = WebApplication.CreateBuilder(args);
 
@@ -338,8 +339,12 @@ static void RegisterServices(WebApplicationBuilder builder)
     });
 
     //Add Health Check
+    var kafkaConnection = builder.Configuration.GetRequiredSection(KafkaConstants.SectionName).Get<KafkaConnection>();
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, DataAcquisitionConstants.ServiceName).GetHealthCheckOptions();
+
     builder.Services.AddHealthChecks()
-        .AddDbContextCheck<DataAcquisitionDbContext>();
+        .AddDbContextCheck<DataAcquisitionDbContext>()
+        .AddKafka(kafkaHealthOptions);
 
     builder.Services.AddSingleton(TimeProvider.System);
     builder.Services.AddSingleton<IDataAcquisitionServiceMetrics, DataAcquisitionServiceMetrics>();

DotNet/Normalization/Program.cs

@@ -39,6 +39,7 @@
 using LantanaGroup.Link.Normalization.Application.Managers;
 using AuditEventMessage = LantanaGroup.Link.Shared.Application.Models.Kafka.AuditEventMessage;
 using Confluent.Kafka;
+using LantanaGroup.Link.Shared.Application.Health;
 
 var builder = WebApplication.CreateBuilder(args);
 
@@ -211,8 +212,13 @@ static void RegisterServices(WebApplicationBuilder builder)
     }
 
     //Add health checks
+    var kafkaConnection = builder.Configuration.GetRequiredSection(KafkaConstants.SectionName).Get<KafkaConnection>();
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, NormalizationConstants.ServiceName).GetHealthCheckOptions();
+
+
     builder.Services.AddHealthChecks()
-        .AddCheck<DatabaseHealthCheck>("Database");
+        .AddCheck<DatabaseHealthCheck>("Database")
+        .AddKafka(kafkaHealthOptions);
 
     builder.Services.AddSwaggerGen(c =>
     {

DotNet/Notification/Program.cs

@@ -38,6 +38,7 @@
 using LantanaGroup.Link.Shared.Application.Models;
 using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.OpenApi.Models;
+using LantanaGroup.Link.Shared.Application.Health;
 
 var builder = WebApplication.CreateBuilder(args);
 
@@ -198,8 +199,11 @@ static void RegisterServices(WebApplicationBuilder builder)
     builder.Services.AddScoped<INotificationRepository, NotificationRepository>();
 
     //Add health checks
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, NotificationConstants.ServiceName).GetHealthCheckOptions();
+
     builder.Services.AddHealthChecks()
-        .AddCheck<DatabaseHealthCheck>("Database");
+        .AddCheck<DatabaseHealthCheck>("Database")
+        .AddKafka(kafkaHealthOptions);
 
     //Add Hosted Services
     builder.Services.AddHostedService<NotificationRequestedListener>();

DotNet/QueryDispatch/Program.cs

@@ -45,7 +45,8 @@
 using QueryDispatch.Domain;
 using LantanaGroup.Link.QueryDispatch.Domain.Entities;
 using QueryDispatch.Application.Extensions;
-
+using HealthChecks.Kafka;
+using LantanaGroup.Link.Shared.Application.Health;
 
 var builder = WebApplication.CreateBuilder(args);
 
@@ -244,8 +245,11 @@
 });
 
 //Add health checks
+var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, QueryDispatchConstants.ServiceName).GetHealthCheckOptions();
+
 builder.Services.AddHealthChecks()
-    .AddDbContextCheck<QueryDispatchDbContext>();
+    .AddDbContextCheck<QueryDispatchDbContext>()
+    .AddKafka(kafkaHealthOptions);
 
 // Logging using Serilog
 builder.Logging.AddSerilog();

DotNet/Report/Program.cs

@@ -17,6 +17,7 @@
 using LantanaGroup.Link.Shared.Application.Extensions;
 using LantanaGroup.Link.Shared.Application.Extensions.Security;
 using LantanaGroup.Link.Shared.Application.Factories;
+using LantanaGroup.Link.Shared.Application.Health;
 using LantanaGroup.Link.Shared.Application.Interfaces;
 using LantanaGroup.Link.Shared.Application.Listeners;
 using LantanaGroup.Link.Shared.Application.Middleware;
@@ -168,8 +169,11 @@ static void RegisterServices(WebApplicationBuilder builder)
     builder.Services.AddControllers();
 
     //Add health checks
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, ReportConstants.ServiceName).GetHealthCheckOptions();
+
     builder.Services.AddHealthChecks()
-        .AddCheck<DatabaseHealthCheck>("Database");
+        .AddCheck<DatabaseHealthCheck>("Database")
+        .AddKafka(kafkaHealthOptions);
 
     // Add swagger
     builder.Services.AddEndpointsApiExplorer();

DotNet/Shared/Application/Health/KafkaHealthConfiguration.cs

@@ -0,0 +1,48 @@
+using Confluent.Kafka;
+using HealthChecks.Kafka;
+using LantanaGroup.Link.Shared.Application.Models.Configs;
+
+namespace LantanaGroup.Link.Shared.Application.Health
+{
+    public class KafkaHealthCheckConfiguration
+    {
+        private KafkaConnection _connection;
+        private string _serviceName;
+
+        public KafkaHealthCheckConfiguration(KafkaConnection connection, string serviceName)
+        {
+            _connection = connection;
+            _serviceName = serviceName;
+        }
+
+        public KafkaHealthCheckOptions GetHealthCheckOptions()
+        {
+            var producerConfig = new ProducerConfig()
+            {
+                BootstrapServers = string.Join(", ", _connection.BootstrapServers),
+                MessageTimeoutMs = 3000
+            };
+
+            if (_connection.SaslProtocolEnabled)
+            {
+                producerConfig.SaslMechanism = SaslMechanism.Plain;
+                producerConfig.SecurityProtocol = SecurityProtocol.SaslPlaintext;
+                producerConfig.SaslUsername = _connection.SaslUsername;
+                producerConfig.SaslPassword = _connection.SaslPassword;
+            }
+
+            return new KafkaHealthCheckOptions()
+            {
+                Configuration = producerConfig,
+                MessageBuilder = MessageBuilder,
+                Topic = "Service-Healthcheck"
+            };
+        }
+
+        private Message<string, string> MessageBuilder(KafkaHealthCheckOptions options)
+        {
+            var utcDate = DateTime.UtcNow;  
+            return new Message<string, string>() { Key = _serviceName, Value = $"Service health check on {utcDate} ({utcDate.Kind})" };
+        }
+    }
+}

DotNet/Shared/Shared.csproj

@@ -18,6 +18,7 @@
 	</PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="AspNetCore.HealthChecks.Kafka" Version="8.0.1" />
     <PackageReference Include="AutoMapper" Version="13.*" />
     <PackageReference Include="Azure.Monitor.OpenTelemetry.AspNetCore" Version="1.1.0" />
     <PackageReference Include="Azure.Security.KeyVault.Secrets" Version="4.6.0" />

DotNet/Submission/Program.cs

@@ -5,6 +5,7 @@
 using LantanaGroup.Link.Shared.Application.Extensions;
 using LantanaGroup.Link.Shared.Application.Extensions.Security;
 using LantanaGroup.Link.Shared.Application.Factories;
+using LantanaGroup.Link.Shared.Application.Health;
 using LantanaGroup.Link.Shared.Application.Interfaces;
 using LantanaGroup.Link.Shared.Application.Listeners;
 using LantanaGroup.Link.Shared.Application.Middleware;
@@ -149,6 +150,15 @@ static void RegisterServices(WebApplicationBuilder builder)
     builder.Services.AddTransient<IKafkaProducerFactory<string, string>, KafkaProducerFactory<string, string>>();
     builder.Services.AddTransient<IRetryEntityFactory, RetryEntityFactory>();
 
+
+    //Add health checks
+    var kafkaConnection = builder.Configuration.GetRequiredSection(KafkaConstants.SectionName).Get<KafkaConnection>();
+    var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, SubmissionConstants.ServiceName).GetHealthCheckOptions();
+
+    builder.Services.AddHealthChecks()
+        .AddKafka(kafkaHealthOptions);
+
+
     // Add repositories
     // TODO
 
@@ -216,6 +226,12 @@ static void SetupMiddleware(WebApplication app)
         app.UseMiddleware<UserScopeMiddleware>();
     }
     app.UseAuthorization();
+
+    //map health check middleware
+    app.MapHealthChecks("/health", new HealthCheckOptions
+    {
+        ResponseWriter = UIResponseWriter.WriteHealthCheckUIResponse
+    }).RequireCors("HealthCheckPolicy");
 }
 
 #endregion

DotNet/Tenant/Program.cs

@@ -30,6 +30,7 @@
 using LantanaGroup.Link.Shared.Application.Middleware;
 using LantanaGroup.Link.Shared.Application.Factories;
 using Confluent.Kafka;
+using LantanaGroup.Link.Shared.Application.Health;
 
 namespace Tenant
 {
@@ -172,8 +173,11 @@ static void RegisterServices(WebApplicationBuilder builder)
 
 
             //Add health checks
+            var kafkaHealthOptions = new KafkaHealthCheckConfiguration(kafkaConnection, TenantConstants.ServiceName).GetHealthCheckOptions();
+
             builder.Services.AddHealthChecks()
-                .AddCheck<DatabaseHealthCheck>("Database");
+                .AddCheck<DatabaseHealthCheck>("Database")
+                .AddKafka(kafkaHealthOptions);
 
             // Add Link Security
             bool allowAnonymousAccess = builder.Configuration.GetValue<bool>("Authentication:EnableAnonymousAccess");

docker-compose.yml

@@ -264,6 +264,7 @@ services:
         kafka-topics --bootstrap-server kafka_b:9094 --command-config /etc/kafka-client.properties --create --if-not-exists --topic AuditableEventOccurred --replication-factor 1 --partitions 1
         kafka-topics --bootstrap-server kafka_b:9094 --command-config /etc/kafka-client.properties --create --if-not-exists --topic AuditableEventOccurred-Error --replication-factor 1 --partitions 1
         kafka-topics --bootstrap-server kafka_b:9094 --command-config /etc/kafka-client.properties --create --if-not-exists --topic AuditableEventOccurred-Retry --replication-factor 1 --partitions 1
+        kafka-topics --bootstrap-server kafka_b:9094 --command-config /etc/kafka-client.properties --create --if-not-exists --topic Service-Healthcheck --replication-factor 1 --partitions 1 --config retention.ms=1
       "
 
   kafka-ui: