perf(audit): bounded async audit writer with batched inserts, one-time bootstrap, retention sweep
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,333 @@
|
||||
using System.Threading.Channels;
|
||||
using ZB.MOM.WW.Audit;
|
||||
using ZB.MOM.WW.MxGateway.Server.Configuration;
|
||||
|
||||
namespace ZB.MOM.WW.MxGateway.Server.Security.Audit;
|
||||
|
||||
/// <summary>
|
||||
/// Bounded, non-blocking <see cref="IAuditWriter"/>: <see cref="WriteAsync"/> enqueues onto a
|
||||
/// fixed-capacity channel and returns, leaving <see cref="AuditDrainService"/> to batch the
|
||||
/// events onto the durable <see cref="IAuditEventSink"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The canonical <see cref="IAuditWriter"/> contract is already best-effort — a failed audit
|
||||
/// write is swallowed rather than propagated. The channel makes the <em>bound</em> on that
|
||||
/// promise explicit: audit can cost the calling RPC at most one enqueue, never a SQLite
|
||||
/// round-trip, and at most <see cref="ChannelCapacity"/> events of memory. This matters on the
|
||||
/// constraint-denial path, where a partially denied bulk RPC previously awaited one insert per
|
||||
/// denied tag, serially, against the same database file the authentication hot path reads.
|
||||
/// <para>
|
||||
/// When the channel is full the newest write is dropped (<see cref="BoundedChannelFullMode.DropWrite"/>)
|
||||
/// and counted in <see cref="DroppedCount"/>. Dropping is the deliberate choice over blocking:
|
||||
/// a stalled audit database must degrade audit completeness, not stall the gateway. Drops are
|
||||
/// logged, and <see cref="AuditDrainService"/> reports the running total on its sweep.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Until a drain attaches (<see cref="AttachDrain"/>), and again after it detaches, writes go
|
||||
/// straight through to <see cref="CanonicalAuditWriter"/>. Enqueueing into a channel nobody will
|
||||
/// ever read would silently discard audit in the processes that have no hosted services — the
|
||||
/// <c>apikey</c> admin CLI and the DI-only tests — so those keep the original synchronous path.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class ChannelAuditWriter : IAuditWriter
|
||||
{
|
||||
/// <summary>
|
||||
/// Maximum number of audit events buffered before writes start being dropped. Sized to
|
||||
/// absorb a fully denied bulk RPC (the gateway's bulk request cap) plus headroom, so a
|
||||
/// realistic burst is buffered rather than lost.
|
||||
/// </summary>
|
||||
public const int ChannelCapacity = 4096;
|
||||
|
||||
private readonly CanonicalAuditWriter _directWriter;
|
||||
private readonly ILogger<ChannelAuditWriter> _logger;
|
||||
private readonly Channel<AuditEvent> _channel;
|
||||
|
||||
private long _droppedCount;
|
||||
private int _drainAttached;
|
||||
private int _dropLogged;
|
||||
|
||||
/// <summary>Creates the writer and its bounded buffer.</summary>
|
||||
/// <param name="directWriter">The synchronous writer used when no drain is attached.</param>
|
||||
/// <param name="logger">Logger for drop diagnostics.</param>
|
||||
public ChannelAuditWriter(CanonicalAuditWriter directWriter, ILogger<ChannelAuditWriter> logger)
|
||||
{
|
||||
_directWriter = directWriter;
|
||||
_logger = logger;
|
||||
|
||||
// DropWrite discards the incoming item and still reports success to the producer, so the
|
||||
// itemDropped callback is the only place a drop can be observed and counted.
|
||||
_channel = Channel.CreateBounded<AuditEvent>(
|
||||
new BoundedChannelOptions(ChannelCapacity)
|
||||
{
|
||||
FullMode = BoundedChannelFullMode.DropWrite,
|
||||
SingleReader = true,
|
||||
SingleWriter = false,
|
||||
},
|
||||
itemDropped: RecordDrop);
|
||||
}
|
||||
|
||||
/// <summary>Gets the number of audit events dropped because the channel was full.</summary>
|
||||
public long DroppedCount => Interlocked.Read(ref _droppedCount);
|
||||
|
||||
/// <summary>Gets the reader the drain service consumes buffered events from.</summary>
|
||||
public ChannelReader<AuditEvent> Reader => _channel.Reader;
|
||||
|
||||
/// <summary>
|
||||
/// Marks a drain as running, so subsequent writes enqueue instead of writing through.
|
||||
/// Called by <see cref="AuditDrainService"/> once its one-time bootstrap has completed.
|
||||
/// </summary>
|
||||
public void AttachDrain() => Volatile.Write(ref _drainAttached, 1);
|
||||
|
||||
/// <summary>
|
||||
/// Marks the drain as no longer running, so writes revert to the synchronous path. Called
|
||||
/// at shutdown so late audit is still persisted rather than buffered into a dead channel.
|
||||
/// </summary>
|
||||
public void DetachDrain() => Volatile.Write(ref _drainAttached, 0);
|
||||
|
||||
/// <summary>
|
||||
/// Enqueues a canonical audit event for the drain to persist. Never blocks, never throws,
|
||||
/// and never touches the store on the caller's thread while a drain is attached.
|
||||
/// </summary>
|
||||
/// <param name="auditEvent">The canonical audit event to persist.</param>
|
||||
/// <param name="cancellationToken">Token honoured only by the direct write-through path.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public Task WriteAsync(AuditEvent auditEvent, CancellationToken cancellationToken = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(auditEvent);
|
||||
|
||||
if (Volatile.Read(ref _drainAttached) == 0)
|
||||
{
|
||||
return _directWriter.WriteAsync(auditEvent, cancellationToken);
|
||||
}
|
||||
|
||||
// Under DropWrite a full channel still reports success (the discard surfaces through the
|
||||
// itemDropped callback); a false here means the channel is completed, which is also a drop.
|
||||
if (!_channel.Writer.TryWrite(auditEvent))
|
||||
{
|
||||
RecordDrop(auditEvent);
|
||||
}
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>Signals that no further events will be enqueued, so the drain loop can finish.</summary>
|
||||
public void CompleteWriting() => _channel.Writer.TryComplete();
|
||||
|
||||
private void RecordDrop(AuditEvent auditEvent)
|
||||
{
|
||||
Interlocked.Increment(ref _droppedCount);
|
||||
|
||||
// Log the first drop only; the running total is reported on the drain's periodic sweep,
|
||||
// so a sustained overload cannot turn audit pressure into a log flood.
|
||||
if (Interlocked.Exchange(ref _dropLogged, 1) == 0)
|
||||
{
|
||||
_logger.LogWarning(
|
||||
"Audit channel is full ({Capacity} events); dropping audit event {EventId} (action {Action}). "
|
||||
+ "Audit is best-effort and bounded; further drops are reported in aggregate.",
|
||||
ChannelCapacity,
|
||||
auditEvent.EventId,
|
||||
auditEvent.Action);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Drains <see cref="ChannelAuditWriter"/> onto the durable <see cref="IAuditEventSink"/>,
|
||||
/// owns the one-time schema bootstrap, and sweeps audit rows past their retention window.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Batching is the point: up to <see cref="MaxBatchSize"/> buffered events are committed in a
|
||||
/// single transaction, so a burst of constraint denials costs a handful of commits instead of
|
||||
/// one per denied tag. The bootstrap runs here — before the writer starts enqueueing — so no
|
||||
/// audit write ever pays a <c>CREATE TABLE IF NOT EXISTS</c> round-trip.
|
||||
/// </remarks>
|
||||
/// <param name="writer">The channel writer whose buffered events are drained.</param>
|
||||
/// <param name="sink">The durable sink events are committed to.</param>
|
||||
/// <param name="security">Security options carrying the audit retention window.</param>
|
||||
/// <param name="timeProvider">Clock used for the retention cutoff and sweep interval.</param>
|
||||
/// <param name="logger">Logger for bootstrap, drain and sweep diagnostics.</param>
|
||||
public sealed class AuditDrainService(
|
||||
ChannelAuditWriter writer,
|
||||
IAuditEventSink sink,
|
||||
SecurityOptions security,
|
||||
TimeProvider timeProvider,
|
||||
ILogger<AuditDrainService> logger) : BackgroundService
|
||||
{
|
||||
/// <summary>Maximum number of audit events committed in one transaction per drain pass.</summary>
|
||||
public const int MaxBatchSize = 64;
|
||||
|
||||
/// <summary>How often the retention sweep runs while the gateway is up.</summary>
|
||||
public static readonly TimeSpan RetentionSweepInterval = TimeSpan.FromHours(1);
|
||||
|
||||
/// <summary>Upper bound on how long shutdown waits for the remaining buffered events.</summary>
|
||||
private static readonly TimeSpan ShutdownDrainCap = TimeSpan.FromSeconds(2);
|
||||
|
||||
/// <summary>
|
||||
/// Bootstraps the audit table, runs one retention sweep, then attaches the drain so the
|
||||
/// writer switches from synchronous write-through to enqueueing.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Token to observe for cancellation.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public override async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
await sink.EnsureInitializedAsync(cancellationToken).ConfigureAwait(false);
|
||||
await SweepRetentionAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
// Audit is best-effort: a bootstrap failure must not take the gateway down. The
|
||||
// sink's own latch will retry the schema check on the first write.
|
||||
logger.LogWarning(exception, "Audit store bootstrap failed; audit writes will retry the schema check.");
|
||||
}
|
||||
|
||||
writer.AttachDrain();
|
||||
|
||||
await base.StartAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Detaches the drain (so late writes go straight to the sink) and gives the buffered
|
||||
/// events a bounded window to reach the store.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Token to observe for cancellation.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public override async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
writer.DetachDrain();
|
||||
writer.CompleteWriting();
|
||||
|
||||
await base.StopAsync(cancellationToken).ConfigureAwait(false);
|
||||
|
||||
using CancellationTokenSource drainCap = new(ShutdownDrainCap);
|
||||
try
|
||||
{
|
||||
await DrainPendingAsync(drainCap.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
logger.LogWarning(
|
||||
"Shutdown drain exceeded {CapSeconds}s; remaining buffered audit events were not persisted.",
|
||||
ShutdownDrainCap.TotalSeconds);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Commits every event currently buffered, in batches of at most <see cref="MaxBatchSize"/>.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Token to observe for cancellation.</param>
|
||||
/// <returns>The number of events persisted.</returns>
|
||||
public async Task<int> DrainPendingAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
int persisted = 0;
|
||||
List<AuditEvent> batch = new(MaxBatchSize);
|
||||
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
batch.Clear();
|
||||
while (batch.Count < MaxBatchSize && writer.Reader.TryRead(out AuditEvent? auditEvent))
|
||||
{
|
||||
batch.Add(auditEvent);
|
||||
}
|
||||
|
||||
if (batch.Count == 0)
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await sink.InsertBatchAsync(batch, cancellationToken).ConfigureAwait(false);
|
||||
persisted += batch.Count;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
// Best-effort: the batch is lost, but the drain keeps going so one poisoned
|
||||
// batch cannot stall every later audit event behind it.
|
||||
logger.LogWarning(
|
||||
exception,
|
||||
"Failed to persist a batch of {Count} audit events; audit writes are best-effort and were suppressed.",
|
||||
batch.Count);
|
||||
}
|
||||
}
|
||||
|
||||
return persisted;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deletes audit rows older than <c>MxGateway:Security:AuditRetentionDays</c>, and reports
|
||||
/// any audit events dropped by channel pressure since the last sweep.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Token to observe for cancellation.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public async Task SweepRetentionAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
DateTimeOffset cutoff = timeProvider.GetUtcNow() - TimeSpan.FromDays(security.AuditRetentionDays);
|
||||
|
||||
try
|
||||
{
|
||||
int deleted = await sink.DeleteOlderThanAsync(cutoff, cancellationToken).ConfigureAwait(false);
|
||||
if (deleted > 0)
|
||||
{
|
||||
logger.LogInformation(
|
||||
"Audit retention sweep removed {Deleted} events older than {Cutoff:o} ({RetentionDays} days).",
|
||||
deleted,
|
||||
cutoff,
|
||||
security.AuditRetentionDays);
|
||||
}
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogWarning(exception, "Audit retention sweep failed; it will be retried on the next interval.");
|
||||
}
|
||||
|
||||
long dropped = writer.DroppedCount;
|
||||
if (dropped > 0)
|
||||
{
|
||||
logger.LogWarning(
|
||||
"{Dropped} audit events have been dropped since startup because the audit channel was full.",
|
||||
dropped);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await Task.WhenAll(
|
||||
DrainLoopAsync(stoppingToken),
|
||||
RetentionLoopAsync(stoppingToken)).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task DrainLoopAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
while (await writer.Reader.WaitToReadAsync(stoppingToken).ConfigureAwait(false))
|
||||
{
|
||||
await DrainPendingAsync(stoppingToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Shutdown; StopAsync performs the final bounded drain.
|
||||
}
|
||||
}
|
||||
|
||||
private async Task RetentionLoopAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
using PeriodicTimer timer = new(RetentionSweepInterval, timeProvider);
|
||||
|
||||
try
|
||||
{
|
||||
while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
|
||||
{
|
||||
await SweepRetentionAsync(stoppingToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Shutdown.
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user