Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(#2945): AsyncDaemonHealthCheck - add maxSameLagTime option #1

Merged
merged 5 commits into from
Mar 21, 2024
Merged

feat(#2945): AsyncDaemonHealthCheck - add maxSameLagTime option #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
19c9daa
add time provider service
MateuszNaKodach Feb 6, 2024
7c85c41
add test
MateuszNaKodach Feb 6, 2024
f64b4a7
add maxSameLagTime implementation
MateuszNaKodach Feb 7, 2024
960f3ed
update docs
MateuszNaKodach Feb 7, 2024
31da385
fix: AsyncDaemonHealthCheck - without events should return Healthy
MateuszNaKodach Feb 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 36 additions & 2 deletions docs/events/projections/healthchecks.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,15 @@
The healthcheck is available in the [Marten.AspNetCore](https://www.nuget.org/packages/Marten.AspNetCore) package.
:::

Marten supports a customizable [HealthChecks](https://learn.microsoft.com/en-us/aspnet/core/host-and-deploy/health-checks?view=aspnetcore-7.0). This can be useful when running the async daemon in a containerized environment such as Kubernetes. The check will verify that no projection's progression lags more than `maxEventLag` behind the `HighWaterMark`. The default `maxEventLag` is 100. Read more about events progression tracking and `HighWaterMark` in [Async Daemon documentation](/events/projections/async-daemon).
Marten supports a customizable [HealthChecks](https://learn.microsoft.com/en-us/aspnet/core/host-and-deploy/health-checks?view=aspnetcore-7.0).
This can be useful when running the async daemon in a containerized environment such as Kubernetes.
Especially if you experience `ProgressionProgressOutOfOrderException` errors in async projections.

The `maxEventLag` setting controls how far behind the `HighWaterMark` any async projection is allowed to lag before it's considered unhealthy. E.g. if the `HighWaterMark` is 1000 and an a system with 3 async projections `ProjA`, `ProjB` and `ProjC` are processed respectively to sequence number 899, 901 and 901 then the system will be considered unhealthy with a `maxEventLag` of 100 (1000 - 899 = 101), BUT healthy with a `mavEventLag` of 101 or higher.
The check will verify that no projection's progression lags more than `maxEventLag` behind the `HighWaterMark`.
The default `maxEventLag` is 100. Read more about events progression tracking and `HighWaterMark` in [Async Daemon documentation](/events/projections/async-daemon).

The `maxEventLag` setting controls how far behind the `HighWaterMark` any async projection is allowed to lag before it's considered unhealthy.
E.g. if the `HighWaterMark` is 1000 and an a system with 3 async projections `ProjA`, `ProjB` and `ProjC` are processed respectively to sequence number 899, 901 and 901 then the system will be considered unhealthy with a `maxEventLag` of 100 (1000 - 899 = 101), BUT healthy with a `mavEventLag` of 101 or higher.

::: tip INFO
The healthcheck will only be checked against `Async` projections
Expand All @@ -21,3 +27,31 @@ Services.AddHealthChecks().AddMartenAsyncDaemonHealthCheck(maxEventLag: 500);
// Map HealthCheck Endpoint
app.MapHealthChecks("/health");
```

If you want to add some time toleration for the healthcheck, you may use additional parameter `maxSameLagTime`.
It treats as unhealthy projections same as described below, but ONLY IF the same projection lag remains for the given time.

### Example use case #1
Assuming that `maxEventLag` = `100` and `maxSameLagTime` = `TimeSpan.FromSeconds(30)`:
- `HighWaterMark` is 1000 and async projection was processed to sequence number 850 at 2024-02-07 01:30:00 -> 'Healthy'
- `HighWaterMark` is 1000 and async projection was processed to sequence number 850 at 2024-02-07 01:30:30 -> 'Unhealthy'

It's unhealty, because the projection haven't progressed since last healthcheck and `maxSameLagTime` elapsed on the same sequence number.


### Example use case #2
Assuming that `maxEventLag` = `100` and `maxSameLagTime` = `TimeSpan.FromSeconds(30)`:
- `HighWaterMark` is 1000 and async projection was processed to sequence number 850 at 2024-02-07 01:30:00 -> 'Healthy'
- `HighWaterMark` is 1000 and async projection was processed to sequence number 851 at 2024-02-07 01:30:30 -> 'Healthy'

It's healthy, because the projection progressed since last healthcheck.

## Example configuration:

```cs
// Add HealthCheck
Services.AddHealthChecks().AddMartenAsyncDaemonHealthCheck(maxEventLag: 500, maxSameLagTime: TimeSpan.FromSeconds(30));

// Map HealthCheck Endpoint
app.MapHealthChecks("/health");
```
130 changes: 102 additions & 28 deletions src/Marten.AspNetCore/Daemon/AsyncDaemonHealthCheckExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,14 @@
#nullable enable
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using JasperFx.Core;
using Marten.Events.Projections;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Microsoft.Extensions.Diagnostics.HealthChecks;

namespace Marten.Events.Daemon;
Expand All @@ -25,19 +28,28 @@ public static class AsyncDaemonHealthCheckExtensions
/// </summary>
/// <param name="builder"><see cref="IHealthChecksBuilder"/></param>
/// <param name="maxEventLag">(optional) Acceptable lag of an eventprojection before it's considered unhealthy - defaults to 100</param>
/// <param name="maxSameLagTime">(optional) Treat projection as healthy if maxEventLag exceeded, but projection sequence changed since last check in given time - defaults to null (uses just maxEventLag)</param>
/// <returns>If healthy: <see cref="HealthCheckResult.Healthy"/> - else <see cref="HealthCheckResult.Unhealthy"/></returns>
public static IHealthChecksBuilder AddMartenAsyncDaemonHealthCheck(this IHealthChecksBuilder builder, int maxEventLag = 100)
public static IHealthChecksBuilder AddMartenAsyncDaemonHealthCheck(
this IHealthChecksBuilder builder,
int maxEventLag = 100,
TimeSpan? maxSameLagTime = null
)
{
builder.Services.AddSingleton(new AsyncDaemonHealthCheckSettings(maxEventLag));
return builder.AddCheck<AsyncDaemonHealthCheck>(nameof(AsyncDaemonHealthCheck), tags: new[] { "Marten", "AsyncDaemon" });
builder.Services.AddSingleton(new AsyncDaemonHealthCheckSettings(maxEventLag, maxSameLagTime));
builder.Services.TryAddSingleton(TimeProvider.System);
return builder.AddCheck<AsyncDaemonHealthCheck>(
nameof(AsyncDaemonHealthCheck),
tags: new[] { "Marten", "AsyncDaemon" }
);
}

/// <summary>
/// Internal class used to DI settings to async daemon health check
/// </summary>
/// <param name="MaxEventLag"></param>
/// <returns></returns>
internal record AsyncDaemonHealthCheckSettings(int MaxEventLag);
internal record AsyncDaemonHealthCheckSettings(int MaxEventLag, TimeSpan? MaxSameLagTime = null);

/// <summary>
/// Health check implementation
Expand All @@ -54,37 +66,99 @@ internal class AsyncDaemonHealthCheck: IHealthCheck
/// </summary>
private readonly int _maxEventLag;

public AsyncDaemonHealthCheck(IDocumentStore store, AsyncDaemonHealthCheckSettings settings)
/// <summary>
/// The allowed time for event projection is lagging (by maxEventLag).
/// If not provided every projection is considered lagging if HighWaterMark - projection.Position >= maxEventLag.
/// If provided only if projection.Position is still the same for given time.
/// When you want to rely only on time just set _maxEventLag=1 and maxSameLagTime to desired value.
/// </summary>
private readonly TimeSpan? _maxSameLagTime;

private readonly TimeProvider _timeProvider;

private readonly ConcurrentDictionary<string, (DateTime CheckedAt, long Sequence)>
_lastProjectionsChecks = new();

public AsyncDaemonHealthCheck(IDocumentStore store, AsyncDaemonHealthCheckSettings settings,
TimeProvider timeProvider)
{
_store = store;
_timeProvider = timeProvider;
_maxEventLag = settings.MaxEventLag;
_maxSameLagTime = settings.MaxSameLagTime;
}
public async Task<HealthCheckResult> CheckHealthAsync(HealthCheckContext context,
CancellationToken cancellationToken = default)

public async Task<HealthCheckResult> CheckHealthAsync(
HealthCheckContext context,
CancellationToken cancellationToken = default
)
{
try
{
var projectionsToCheck = _store.Options
.Events
.Projections()
.Where(x => x.Lifecycle == ProjectionLifecycle.Async) // Only check async projections to avoid issus where inline progression counter is set.
.Select(x => $"{x.ProjectionName}:All")
.ToHashSet();

var allProgress = await _store.Advanced.AllProjectionProgress(token: cancellationToken).ConfigureAwait(true);

var highWaterMark = allProgress.First(x => string.Equals("HighWaterMark", x.ShardName));
var projectionMarks = allProgress.Where(x => !string.Equals("HighWaterMark", x.ShardName));

var unhealthy = projectionMarks
.Where(x => projectionsToCheck.Contains(x.ShardName))
.Where(x => x.Sequence <= highWaterMark.Sequence - _maxEventLag)
.Select(x => x.ShardName)
.ToArray();

return unhealthy.Any()
? HealthCheckResult.Unhealthy($"Unhealthy: Async projection sequence is more than {_maxEventLag} events behind for projection(s): {unhealthy.Join(", ")}")
: HealthCheckResult.Healthy("Healthy");
var projectionsToCheck = _store.Options.Events.Projections()
.Where(x => x.Lifecycle == ProjectionLifecycle.Async)
.Select(x => $"{x.ProjectionName}:All")
.ToHashSet();

var allProgress = await _store.Advanced.AllProjectionProgress(token: cancellationToken)
.ConfigureAwait(true);

var highWaterMark = allProgress.FirstOrDefault(x => string.Equals("HighWaterMark", x.ShardName));
if (highWaterMark is null)
{
return HealthCheckResult.Healthy("Healthy");
}

var projectionMarks = allProgress.Where(x => !string.Equals("HighWaterMark", x.ShardName)).ToArray();

var projectionsSequences = projectionMarks.Where(x => projectionsToCheck.Contains(x.ShardName))
.Select(x => new { x.ShardName, x.Sequence })
.ToArray();

var laggingProjections = projectionsSequences
.Where(x => x.Sequence <= highWaterMark.Sequence - _maxEventLag)
.ToArray();

if (_maxSameLagTime is null)
{
return laggingProjections.Any()
? HealthCheckResult.Unhealthy(
$"Unhealthy: Async projection sequence is more than {_maxEventLag} events behind for projection(s): {laggingProjections.Select(x => x.ShardName).Join(", ")}"
)
: HealthCheckResult.Healthy("Healthy");
}

var now = _timeProvider.GetUtcNow().UtcDateTime;

var projectionsLaggingWithSamePositionForGivenTime = laggingProjections.Where(
x =>
{
var (laggingSince, lastKnownPosition) =
_lastProjectionsChecks.GetValueOrDefault(x.ShardName, (now, x.Sequence));

var isLaggingWithSamePositionForGivenTime =
now.Subtract(laggingSince) >= _maxSameLagTime &&
x.Sequence == lastKnownPosition;

return isLaggingWithSamePositionForGivenTime;
}
)
.ToArray();

foreach (var laggingProjection in laggingProjections)
{
_lastProjectionsChecks.AddOrUpdate(
laggingProjection.ShardName,
_ => (now, laggingProjection.Sequence),
(_, _) => (now, laggingProjection.Sequence)
);
}

return projectionsLaggingWithSamePositionForGivenTime.Any()
? HealthCheckResult.Unhealthy(
$"Unhealthy: Async projection sequence is more than {_maxEventLag} events behind with same sequence for more than {_maxSameLagTime} for projection(s): {projectionsLaggingWithSamePositionForGivenTime.Select(x => x.ShardName).Join(", ")}"
)
: HealthCheckResult.Healthy("Healthy");
}
catch (Exception ex)
{
Expand Down
Loading
Loading