fix(audit): maintenance-login roll-forward/backfill, archive orphan cleanup, head-export health (ScadaBridge#113 #114 #115)
#113: AuditLogPartitionMaintenance (monthly SPLIT RANGE) and the backfill-source-node endpoint now use the AuditLog:MaintenanceConnectionString login when set (unset = unchanged). Migration AuditMaintenanceGrants gives scadabridge_audit_purger ALTER ANY DATASPACE and column-level UPDATE (SourceNode). #114: archive cleanup deletes the writer's own *.tmp files and manifest-less data files older than 24 h inside {Directory}/{YYYY} (DeletedOrphan, dry-run aware). #115: head-export failures bump scadabridge.audit.chain_head_export.failures and Degrade the /health/ready entry audit-chain-head-export (failing or stale > 2 intervals). Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
@@ -184,7 +184,7 @@ spec for each is `docs/requirements/Component-<Name>.md`, and `README.md` carrie
|
||||
- Append-only enforced via DB roles (writer role has INSERT only, no UPDATE/DELETE).
|
||||
- **Clustered key `(OccurredAtUtc, EventId)` + aligned unique `UX_AuditLog_EventId_Occurred (EventId, OccurredAtUtc)` (migration `FlipAuditLogClusteredKey`, #111, B2 of the #78 benchmark).** The Audit Log page's newest-first page/keyset now reads an ordered range instead of the whole table (300k rows: 90,532 → 34 reads). Offline whole-table rebuild — script `docs/plans/sql/FlipAuditLogClusteredKey.sql` (+ `.rollback.sql`), both centrals stopped: the partition-switch staging table now reads the live key order, but binaries older than the flip hard-code the old one. DEV/QA not applied (sizing + window in #111).
|
||||
- **Parquet cold archive (T2, ScadaBridge#42) — default OFF (`AuditLog:Archive:Enabled` + absolute `Directory`, shared by both centrals).** The purge actor writes each expired partition to `{Directory}/{YYYY}/auditlog-{YYYY-MM}-{stamp}Z.parquet` (10 stored columns + `ChainSeq`/`RowHash`, tick-exact `OccurredAtUtcTicks`) plus `.tombstones.parquet` and a `.manifest.json` (written last; chain head at archive time, SHA-256 per file), read under the chain lock SHARED; then the switch runs with a guard (under the EXCLUSIVE lock: partition row count + each month's `LastSeq` must still equal the archive's, else THROW 51002/51003 and nothing is purged). Fail closed: a partition that cannot be archived is not purged. `audit archive list|verify` = `ListAuditArchives`/`VerifyAuditArchive` (verify-chain's gate): file hashes, chain recomputed from the archive, chain end vs the DB purge anchor (`Matches`/`Superseded`/`NotPurged`/`Mismatch`), CLI exit 3. `audit export --format parquet` streams the 24 export columns (was 501). No migration. **Owner decisions (#109):** destination stays a directory/UNC (no object-storage client; share + permissions named at enablement), parquet-tools-only reading (no re-import), unsigned like the chain. **Retention** `AuditLog:Archive:RetentionDays` (default 0 = never; else > `AuditLog:RetentionDays`, counted from the month's end) and **superseded stale-archive cleanup** (`DeleteSupersededStaleArchives`, default true: a retried purge's `Archived` archive is deleted only when a later `Purged`, verifying archive provably holds every row — same anchor + chain-end hash, unchained rows by EventId; else `KeptStale`) run after every purge tick; deletion touches only the archive name's own three files, manifest first; `audit archive cleanup [--dry-run]` = `CleanupAuditArchives` (Administrator).
|
||||
- **Hash-chain tamper evidence (T1, ScadaBridge#41) — default OFF (`AuditLog:HashChain:Enabled`).** One SHA-256 chain per UTC calendar month: `ChainSeq` + `RowHash = SHA-256(prev || canonical(row))` on `AuditLog` (migration `AddAuditLogHashChain`), linked by the central `AuditLogHashChainService` (both nodes, serialised by `sp_getapplock`) — never by the ingest path. `AuditLogChainHead` catches tail deletion and carries the partition-switch checkpoint anchor; the per-channel DELETE writes an `AuditLogChainTombstone` per chained row in the same transaction. `audit verify-chain --month` = management command `VerifyAuditChain` (OperationalAudit, unscoped callers only), first break with partition/ChainSeq/EventId, CLI exit 3. A chained row's `SourceNode` is frozen (backfill skips it). **Unsigned by owner decision (#107); external anchor = off-box head export** (`AuditLog:HashChain:HeadExport`, default off: both centrals append changed per-month heads — month, `LastSeq`/`LastHash`, purge anchor, row + tombstone counts — to append-only JSONL files `{Directory}/{yyyy}/audit-chain-heads-{date}-{node}.jsonl` or RFC 5424 syslog UDP/TCP); `audit verify-heads --path` = `VerifyAuditChainHeads` checks them against the DB (a re-hashed forgery passes verify-chain, fails verify-heads; exit 3). **Two-login hardening (#107 item 3):** `AuditLog:MaintenanceConnectionString` (`${secret:}`-capable, unset = runtime login) runs the chain job, verify/head export, partition switch-out + archive and per-channel purge on `AuditMaintenanceContext` (keyed `audit-maintenance` repository/archive); that login gets `scadabridge_audit_chainer` + `scadabridge_audit_purger`, the runtime login `scadabridge_audit_writer` (whose DENYs block chaining/purging) — pinned by `AuditMaintenanceLoginTests`.
|
||||
- **Hash-chain tamper evidence (T1, ScadaBridge#41) — default OFF (`AuditLog:HashChain:Enabled`).** One SHA-256 chain per UTC calendar month: `ChainSeq` + `RowHash = SHA-256(prev || canonical(row))` on `AuditLog` (migration `AddAuditLogHashChain`), linked by the central `AuditLogHashChainService` (both nodes, serialised by `sp_getapplock`) — never by the ingest path. `AuditLogChainHead` catches tail deletion and carries the partition-switch checkpoint anchor; the per-channel DELETE writes an `AuditLogChainTombstone` per chained row in the same transaction. `audit verify-chain --month` = management command `VerifyAuditChain` (OperationalAudit, unscoped callers only), first break with partition/ChainSeq/EventId, CLI exit 3. A chained row's `SourceNode` is frozen (backfill skips it). **Unsigned by owner decision (#107); external anchor = off-box head export** (`AuditLog:HashChain:HeadExport`, default off: both centrals append changed per-month heads — month, `LastSeq`/`LastHash`, purge anchor, row + tombstone counts — to append-only JSONL files `{Directory}/{yyyy}/audit-chain-heads-{date}-{node}.jsonl` or RFC 5424 syslog UDP/TCP); `audit verify-heads --path` = `VerifyAuditChainHeads` checks them against the DB (a re-hashed forgery passes verify-chain, fails verify-heads; exit 3). **Two-login hardening (#107 item 3):** `AuditLog:MaintenanceConnectionString` (`${secret:}`-capable, unset = runtime login) runs the chain job, verify/head export, partition switch-out + archive and per-channel purge on `AuditMaintenanceContext` (keyed `audit-maintenance` repository/archive); that login gets `scadabridge_audit_chainer` + `scadabridge_audit_purger`, the runtime login `scadabridge_audit_writer` (whose DENYs block chaining/purging) — pinned by `AuditMaintenanceLoginTests`. Since #113 the monthly roll-forward and `backfill-source-node` use it too (migration `AuditMaintenanceGrants`: purger gets `ALTER ANY DATASPACE` + `UPDATE (SourceNode)`). Head-export failures (#115) bump `scadabridge.audit.chain_head_export.failures` and Degrade `/health/ready` entry `audit-chain-head-export`; archive cleanup (#114) also deletes writer-named orphan temp / manifest-less files older than 24 h (`DeletedOrphan`).
|
||||
- Node-of-origin is captured alongside site-of-origin: `SourceNode` (`varchar(64)` NULL) on `AuditLog`, `Notifications`, and `SiteCalls` — `node-a`/`node-b` for site rows (qualified by `SourceSiteId`/`SourceSite`), `central-a`/`central-b` for central direct-write rows. Stamped at the writing node, carried verbatim through telemetry + reconciliation, and indexed via `IX_AuditLog_Node_Occurred (SourceNode, OccurredAtUtc)` on `AuditLog`.
|
||||
- Per-node stuck KPIs (M5.3 T6): Notification Outbox and Site Call Audit expose `PerNodeNotificationKpiRequest`/`PerNodeSiteCallKpiRequest` messages that group stuck/parked/delivered counts by `SourceNode`, surfacing per-node breakdowns on the Health dashboard.
|
||||
- `audit tree --execution-id <guid>` CLI command (M5.3 T8) + `GET /api/audit/tree` endpoint — resolves any node to its chain root and renders the full execution tree; backed by `IAuditLogRepository.GetExecutionTreeAsync`.
|
||||
|
||||
@@ -823,8 +823,14 @@ MS SQL for direct-write events). Unredacted secrets never persist.
|
||||
- **Safe deletion.** Cleanup deletes only the three files a well-formed archive name implies
|
||||
(`{name}.manifest.json`, `.parquet`, `.tombstones.parquet`) in that archive's own
|
||||
`{Directory}/{YYYY}` folder — never a path read from a manifest — manifest first, so an
|
||||
interrupted deletion leaves leftovers that are no longer an archive (orphans of an interrupted
|
||||
archive WRITE are not cleaned yet — #114). A cleanup failure is a
|
||||
interrupted deletion leaves leftovers that are no longer an archive. **Orphans of an interrupted
|
||||
archive WRITE (ScadaBridge#114):** the same pass also deletes, reporting action `DeletedOrphan`
|
||||
(dry-run aware), the writer's own temp files (`{name}.parquet.tmp`,
|
||||
`{name}.tombstones.parquet.tmp`, `{name}.manifest.json.tmp`) and data files whose
|
||||
`.manifest.json` was never written — only files named exactly like that for a well-formed
|
||||
archive name, directly inside a four-digit `{Directory}/{YYYY}` folder whose year matches the
|
||||
name, not reparse points, and last written more than 24 h ago (a running write is never
|
||||
touched). Disk-only; no correctness impact. A cleanup failure is a
|
||||
purge failure on the health surface. `scadabridge audit archive cleanup [--dry-run]`
|
||||
(`CleanupAuditArchives`, Administrator) runs the same pass on demand or reports what it
|
||||
would delete. Pinned by `AuditLogArchiveCleanupTests` (mutation-checked: each safety rule
|
||||
@@ -906,11 +912,17 @@ MS SQL for direct-write events). Unredacted secrets never persist.
|
||||
|
||||
`AuditMaintenanceLoginTests` runs exactly this setup on SQL Server: the writer ingests and is
|
||||
refused the chain UPDATE and the per-channel DELETE; the maintenance login chains, verifies,
|
||||
reads heads, purges per channel, enumerates partitions, archives and switches out. Not moved
|
||||
to the maintenance login: the monthly partition roll-forward (`ALTER PARTITION FUNCTION`,
|
||||
needs `ALTER ANY DATASPACE` on the runtime login, or run it as a DBA job) and the one-time
|
||||
`audit backfill-source-node` (needs `UPDATE` on `SourceNode` — a DBA-run step in a hardened
|
||||
setup) — both tracked in #113. Migrations themselves are applied by a DBA with the production SQL scripts.
|
||||
reads heads, purges per channel, enumerates partitions, archives and switches out. **Also on the
|
||||
maintenance login (ScadaBridge#113):** the monthly partition roll-forward
|
||||
(`ALTER PARTITION FUNCTION … SPLIT RANGE`, `AuditLogPartitionMaintenance` now built on
|
||||
`AuditMaintenanceContext`) and `audit backfill-source-node` (the management endpoint resolves the
|
||||
keyed maintenance `IAuditLogRepository`). Migration `AuditMaintenanceGrants` gives
|
||||
`scadabridge_audit_purger` the two grants they need that `ALTER ON SCHEMA::dbo` does not confer:
|
||||
database-level `ALTER ANY DATASPACE` (roll-forward) and column-level `UPDATE (SourceNode)` on
|
||||
`dbo.AuditLog` (backfill; no other column becomes updatable, the writer's DENY still wins).
|
||||
Unset `AuditLog:MaintenanceConnectionString` keeps both on the runtime login as before — which
|
||||
then needs those same two grants (or db_owner), exactly as the single-login default always did.
|
||||
`AuditMaintenanceLoginTests` pins both with a writer-refused negative control. Migrations themselves are applied by a DBA with the production SQL scripts.
|
||||
- **Authorization.** Reading the Audit Log requires the existing **Audit** role
|
||||
extended with a new **OperationalAudit** permission. Per-site row scoping
|
||||
reuses the existing site-permission model; bulk export requires an additional
|
||||
@@ -973,7 +985,12 @@ MS SQL for direct-write events). Unredacted secrets never persist.
|
||||
`FileMode.Append` only, one file per node per day; give the service account create +
|
||||
append-data rights and nothing else), or `Sink: Syslog` — RFC 5424 (`<PRI>1 … chainhead -
|
||||
{json}`, facility 13 "log audit", severity Notice) over UDP (RFC 5426) or TCP (RFC 6587
|
||||
octet counting). A failed write is logged and retried with the same heads next tick (no health counter yet — #115).
|
||||
octet counting). A failed write is logged and retried with the same heads next tick, and (ScadaBridge#115)
|
||||
is also a health signal: counter `scadabridge.audit.chain_head_export.failures` on the shared
|
||||
ScadaBridge meter, plus a `/health/ready` entry `audit-chain-head-export` that is Degraded (HTTP
|
||||
200, Overview-visible, not out of rotation) while the last tick failed or no tick has succeeded
|
||||
for more than two intervals; Healthy when the export is disabled. UDP syslog is fire-and-forget,
|
||||
so a dropped datagram is invisible by design — use `Protocol: Tcp` where that matters.
|
||||
`scadabridge audit verify-heads --path <file|dir>` reads those records back (file-sink
|
||||
lines or a syslog collector's log) and the `VerifyAuditChainHeads` command checks each
|
||||
against the database: the link stored at the exported sequence (the head, a live row's
|
||||
|
||||
+9
-1
@@ -31,6 +31,7 @@ public sealed class AuditChainHeadExportService : IHostedService, IDisposable
|
||||
private readonly string? _node;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly ILogger<AuditChainHeadExportService> _logger;
|
||||
private readonly AuditChainHeadExportStatus _status;
|
||||
private readonly Dictionary<DateTime, (long Seq, long? Purged, long Links)> _lastExported = new();
|
||||
private CancellationTokenSource? _cts;
|
||||
private Task? _loop;
|
||||
@@ -42,14 +43,17 @@ public sealed class AuditChainHeadExportService : IHostedService, IDisposable
|
||||
/// <param name="nodeName">This node's name for the records.</param>
|
||||
/// <param name="logger">Logger.</param>
|
||||
/// <param name="time">Clock.</param>
|
||||
/// <param name="status">Health status the tick outcome is recorded on (ScadaBridge#115).</param>
|
||||
public AuditChainHeadExportService(
|
||||
IServiceScopeFactory scopeFactory,
|
||||
IOptions<AuditLogHashChainOptions> options,
|
||||
IAuditChainHeadSink? sink,
|
||||
string? nodeName,
|
||||
ILogger<AuditChainHeadExportService> logger,
|
||||
TimeProvider? time = null)
|
||||
TimeProvider? time = null,
|
||||
AuditChainHeadExportStatus? status = null)
|
||||
{
|
||||
_status = status ?? new AuditChainHeadExportStatus();
|
||||
_scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory));
|
||||
_options = (options ?? throw new ArgumentNullException(nameof(options))).Value.HeadExport;
|
||||
_sink = sink;
|
||||
@@ -67,6 +71,7 @@ public sealed class AuditChainHeadExportService : IHostedService, IDisposable
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
_status.MarkStarted(_options.IntervalMinutes, _time.GetUtcNow().UtcDateTime);
|
||||
_logger.LogInformation("AuditLog chain-head export every {Minutes} min to {Sink}.", _options.IntervalMinutes, _sink.Describe);
|
||||
var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
_cts = cts;
|
||||
@@ -118,6 +123,7 @@ public sealed class AuditChainHeadExportService : IHostedService, IDisposable
|
||||
.ToList();
|
||||
if (changed.Count == 0)
|
||||
{
|
||||
_status.RecordSuccess(now);
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -128,6 +134,7 @@ public sealed class AuditChainHeadExportService : IHostedService, IDisposable
|
||||
_lastExported[h.Month] = (h.LastSeq, h.PurgedThroughSeq, h.ChainedRows + h.Tombstones);
|
||||
}
|
||||
|
||||
_status.RecordSuccess(now);
|
||||
_logger.LogDebug("Exported {Count} AuditLog chain head(s) to {Sink}.", records.Count, _sink.Describe);
|
||||
return records.Count;
|
||||
}
|
||||
@@ -139,6 +146,7 @@ public sealed class AuditChainHeadExportService : IHostedService, IDisposable
|
||||
{
|
||||
// Contained: the next tick retries the same heads (nothing was marked exported).
|
||||
_logger.LogError(ex, "AuditLog chain-head export to {Sink} failed; retrying next tick.", _sink.Describe);
|
||||
_status.RecordFailure(_time.GetUtcNow().UtcDateTime, ex.Message);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
using System.Diagnostics.Metrics;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central.HeadExport;
|
||||
|
||||
/// <summary>
|
||||
/// Health signal of the off-box chain-head export (ScadaBridge#115). Updated by
|
||||
/// <see cref="AuditChainHeadExportService"/> on every tick; read by the Host's
|
||||
/// <c>audit-chain-head-export</c> /health/ready check. A sink that stays down used to be a
|
||||
/// log line only, silently stopping the external anchor.
|
||||
/// </summary>
|
||||
public sealed class AuditChainHeadExportStatus
|
||||
{
|
||||
private static readonly Meter Meter = new(ScadaBridgeTelemetry.MeterName);
|
||||
|
||||
private static readonly Counter<long> FailuresCounter =
|
||||
Meter.CreateCounter<long>("scadabridge.audit.chain_head_export.failures", unit: "1",
|
||||
description: "AuditLog chain-head export ticks that failed to read or write the heads (sink offline / unreachable); retried next tick.");
|
||||
|
||||
private readonly object _gate = new();
|
||||
private long _totalFailures;
|
||||
private int _consecutiveFailures;
|
||||
private DateTime? _lastSuccessUtc;
|
||||
private DateTime? _lastFailureUtc;
|
||||
private string? _lastError;
|
||||
private DateTime? _startedUtc;
|
||||
|
||||
/// <summary>Whether the export is running at all (false = disabled, nothing to report).</summary>
|
||||
public bool Enabled { get; private set; }
|
||||
|
||||
/// <summary>Minutes between export ticks (for the staleness test).</summary>
|
||||
public int IntervalMinutes { get; private set; } = 60;
|
||||
|
||||
/// <summary>Total failed ticks since process start.</summary>
|
||||
public long TotalFailures { get { lock (_gate) { return _totalFailures; } } }
|
||||
|
||||
/// <summary>Failed ticks since the last success.</summary>
|
||||
public int ConsecutiveFailures { get { lock (_gate) { return _consecutiveFailures; } } }
|
||||
|
||||
/// <summary>UTC time of the last successful tick, or null.</summary>
|
||||
public DateTime? LastSuccessUtc { get { lock (_gate) { return _lastSuccessUtc; } } }
|
||||
|
||||
/// <summary>UTC time of the last failed tick, or null.</summary>
|
||||
public DateTime? LastFailureUtc { get { lock (_gate) { return _lastFailureUtc; } } }
|
||||
|
||||
/// <summary>Message of the last failure (never a secret: the exception message only), or null.</summary>
|
||||
public string? LastError { get { lock (_gate) { return _lastError; } } }
|
||||
|
||||
/// <summary>UTC time the export loop started, or null.</summary>
|
||||
public DateTime? StartedUtc { get { lock (_gate) { return _startedUtc; } } }
|
||||
|
||||
/// <summary>Marks the export as running.</summary>
|
||||
/// <param name="intervalMinutes">Tick interval.</param>
|
||||
/// <param name="nowUtc">Start time.</param>
|
||||
public void MarkStarted(int intervalMinutes, DateTime nowUtc)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
Enabled = true;
|
||||
IntervalMinutes = intervalMinutes;
|
||||
_startedUtc = nowUtc;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Records a successful tick.</summary>
|
||||
/// <param name="nowUtc">Time of the tick.</param>
|
||||
public void RecordSuccess(DateTime nowUtc)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
_consecutiveFailures = 0;
|
||||
_lastSuccessUtc = nowUtc;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Records a failed tick and bumps the failure counter metric.</summary>
|
||||
/// <param name="nowUtc">Time of the tick.</param>
|
||||
/// <param name="error">The failure message.</param>
|
||||
public void RecordFailure(DateTime nowUtc, string error)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
_totalFailures++;
|
||||
_consecutiveFailures++;
|
||||
_lastFailureUtc = nowUtc;
|
||||
_lastError = error;
|
||||
}
|
||||
|
||||
FailuresCounter.Add(1);
|
||||
}
|
||||
}
|
||||
@@ -419,6 +419,7 @@ public static class ServiceCollectionExtensions
|
||||
|
||||
// Off-box chain-head export (ScadaBridge#107 item 1, AuditLog:HashChain:HeadExport,
|
||||
// default off). The sink is built from the bound options; null when disabled.
|
||||
services.AddSingleton<Central.HeadExport.AuditChainHeadExportStatus>();
|
||||
services.AddHostedService(sp =>
|
||||
{
|
||||
var options = sp.GetRequiredService<Microsoft.Extensions.Options.IOptions<AuditLogHashChainOptions>>();
|
||||
@@ -434,7 +435,8 @@ public static class ServiceCollectionExtensions
|
||||
options,
|
||||
sink,
|
||||
node,
|
||||
sp.GetRequiredService<ILogger<Central.HeadExport.AuditChainHeadExportService>>());
|
||||
sp.GetRequiredService<ILogger<Central.HeadExport.AuditChainHeadExportService>>(),
|
||||
status: sp.GetRequiredService<Central.HeadExport.AuditChainHeadExportStatus>());
|
||||
});
|
||||
|
||||
// Parquet cold archive (ScadaBridge#42). Default OFF; when enabled the purge actor
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
|
||||
/// <summary>DI keys for the AuditLog maintenance-login registrations (ScadaBridge#107/#113).</summary>
|
||||
public static class AuditLogServiceKeys
|
||||
{
|
||||
/// <summary>
|
||||
/// Key of the maintenance-login <see cref="IAuditLogRepository"/> / <c>IAuditLogArchive</c>:
|
||||
/// the connection from <c>AuditLog:MaintenanceConnectionString</c>, or the runtime connection when unset.
|
||||
/// </summary>
|
||||
public const string Maintenance = "audit-maintenance";
|
||||
}
|
||||
@@ -169,6 +169,13 @@ public enum AuditArchiveCleanupAction
|
||||
/// so it may be the only copy of some rows (see Detail).
|
||||
/// </summary>
|
||||
KeptStale,
|
||||
|
||||
/// <summary>
|
||||
/// Deleted as an orphan of an interrupted archive write (ScadaBridge#114): a temp file of the
|
||||
/// archive writer (<c>*.parquet.tmp</c>, <c>*.tombstones.parquet.tmp</c>, <c>*.manifest.json.tmp</c>)
|
||||
/// or a data file whose <c>.manifest.json</c> was never written, older than the safe age.
|
||||
/// </summary>
|
||||
DeletedOrphan,
|
||||
}
|
||||
|
||||
/// <summary>One archive considered by cleanup.</summary>
|
||||
|
||||
+88
@@ -127,6 +127,8 @@ public sealed partial class AuditLogPartitionArchive
|
||||
}
|
||||
}
|
||||
|
||||
SweepOrphans(dryRun, items);
|
||||
|
||||
foreach (var item in items.Where(i => i.Action != AuditArchiveCleanupAction.KeptStale))
|
||||
{
|
||||
_logger.LogInformation("AuditLog archive cleanup{DryRun}: {Action} {Archive} — {Detail}",
|
||||
@@ -136,6 +138,92 @@ public sealed partial class AuditLogPartitionArchive
|
||||
return new AuditArchiveCleanupResult(dryRun, _options.RetentionDays, items);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Files younger than this are never treated as orphans: a running archive write may still be
|
||||
/// producing them (ScadaBridge#114).
|
||||
/// </summary>
|
||||
internal static readonly TimeSpan OrphanMinAge = TimeSpan.FromHours(24);
|
||||
|
||||
// Longest suffix first. A file is only ever considered through one of these exact suffixes
|
||||
// on a well-formed archive name, so nothing foreign in the directory can match.
|
||||
private static readonly string[] TempSuffixes = [".manifest.json.tmp", ".tombstones.parquet.tmp", ".parquet.tmp"];
|
||||
private static readonly string[] DataSuffixes = [".tombstones.parquet", ".parquet"];
|
||||
|
||||
/// <summary>
|
||||
/// Removes the leftovers of an interrupted archive write (ScadaBridge#114): the writer's own
|
||||
/// temp files, and data files whose manifest — the commit marker — was never written. Only
|
||||
/// files named exactly like the writer names them, directly inside a four-digit
|
||||
/// <c>{Directory}/{YYYY}</c> folder whose year matches the file's, and last written longer ago
|
||||
/// than <see cref="OrphanMinAge"/>. Reparse points are skipped.
|
||||
/// </summary>
|
||||
private void SweepOrphans(bool dryRun, List<AuditArchiveCleanupItem> items)
|
||||
{
|
||||
var root = _options.Directory!;
|
||||
if (!Directory.Exists(root))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var cutoff = _time.GetUtcNow().UtcDateTime - OrphanMinAge;
|
||||
foreach (var yearDir in Directory.EnumerateDirectories(root))
|
||||
{
|
||||
var year = Path.GetFileName(yearDir);
|
||||
if (year.Length != 4 || !year.All(char.IsAsciiDigit) || IsReparsePoint(yearDir))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
foreach (var path in Directory.EnumerateFiles(yearDir))
|
||||
{
|
||||
var file = Path.GetFileName(path);
|
||||
var tmpSuffix = TempSuffixes.FirstOrDefault(x => file.EndsWith(x, StringComparison.Ordinal));
|
||||
var isTemp = tmpSuffix is not null;
|
||||
var suffix = tmpSuffix ?? DataSuffixes.FirstOrDefault(x => file.EndsWith(x, StringComparison.Ordinal));
|
||||
if (suffix is null)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var stem = file[..^suffix.Length];
|
||||
if (!IsValidName(stem) || stem.Substring("auditlog-".Length, 4) != year || IsReparsePoint(path))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!isTemp && File.Exists(Path.Combine(yearDir, stem + ".manifest.json")))
|
||||
{
|
||||
continue; // a committed archive's file
|
||||
}
|
||||
|
||||
if (File.GetLastWriteTimeUtc(path) > cutoff)
|
||||
{
|
||||
continue; // possibly still being written
|
||||
}
|
||||
|
||||
if (!dryRun)
|
||||
{
|
||||
try
|
||||
{
|
||||
File.Delete(path);
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
|
||||
{
|
||||
_logger.LogWarning(ex, "AuditLog archive cleanup could not delete orphan {File}.", file);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
items.Add(new AuditArchiveCleanupItem(stem, stem.Substring("auditlog-".Length, 7), AuditArchiveCleanupAction.DeletedOrphan,
|
||||
isTemp
|
||||
? $"{file}: temp file of an interrupted archive write, older than {OrphanMinAge.TotalHours:0} h."
|
||||
: $"{file}: no {stem}.manifest.json (the archive write never committed), older than {OrphanMinAge.TotalHours:0} h."));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static bool IsReparsePoint(string path)
|
||||
=> (File.GetAttributes(path) & FileAttributes.ReparsePoint) != 0;
|
||||
|
||||
/// <summary>
|
||||
/// Proves <paramref name="later"/> holds every row of <paramref name="stale"/>; returns null
|
||||
/// when it does, else why not.
|
||||
|
||||
+2
-1
@@ -1,5 +1,6 @@
|
||||
using Microsoft.AspNetCore.DataProtection;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
|
||||
@@ -23,7 +24,7 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Maintenance;
|
||||
public sealed class AuditMaintenanceContext : IAsyncDisposable, IDisposable
|
||||
{
|
||||
/// <summary>DI key of the maintenance-login <c>IAuditLogRepository</c> and <c>IAuditLogArchive</c>.</summary>
|
||||
public const string ServiceKey = "audit-maintenance";
|
||||
public const string ServiceKey = AuditLogServiceKeys.Maintenance;
|
||||
|
||||
private readonly bool _owned;
|
||||
|
||||
|
||||
+2385
File diff suppressed because it is too large
Load Diff
+42
@@ -0,0 +1,42 @@
|
||||
using Microsoft.EntityFrameworkCore.Migrations;
|
||||
|
||||
#nullable disable
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
|
||||
{
|
||||
/// <summary>
|
||||
/// Grants for the two AuditLog maintenance paths that now run on the maintenance login
|
||||
/// (<c>AuditLog:MaintenanceConnectionString</c>, ScadaBridge#113). Permission DDL only.
|
||||
/// <list type="bullet">
|
||||
/// <item><c>ALTER ANY DATASPACE</c> to <c>scadabridge_audit_purger</c> — the monthly
|
||||
/// roll-forward's <c>ALTER PARTITION FUNCTION … SPLIT RANGE</c> / <c>ALTER PARTITION SCHEME
|
||||
/// … NEXT USED</c>; <c>ALTER ON SCHEMA::dbo</c> does not confer it.</item>
|
||||
/// <item>column-level <c>UPDATE (SourceNode)</c> to the same role — the one-time
|
||||
/// <c>audit backfill-source-node</c>. Column-scoped on purpose: no other AuditLog column
|
||||
/// becomes updatable.</item>
|
||||
/// </list>
|
||||
/// </summary>
|
||||
public partial class AuditMaintenanceGrants : Migration
|
||||
{
|
||||
/// <inheritdoc />
|
||||
protected override void Up(MigrationBuilder migrationBuilder)
|
||||
{
|
||||
migrationBuilder.Sql(@"
|
||||
IF DATABASE_PRINCIPAL_ID('scadabridge_audit_purger') IS NULL
|
||||
EXEC sp_executesql N'CREATE ROLE scadabridge_audit_purger';
|
||||
GRANT ALTER ANY DATASPACE TO scadabridge_audit_purger;
|
||||
GRANT UPDATE (SourceNode) ON dbo.AuditLog TO scadabridge_audit_purger;"); // AUDIT-PURGE-ALLOWED: role grant for the maintenance principal (SourceNode backfill)
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void Down(MigrationBuilder migrationBuilder)
|
||||
{
|
||||
migrationBuilder.Sql(@"
|
||||
IF DATABASE_PRINCIPAL_ID('scadabridge_audit_purger') IS NOT NULL
|
||||
BEGIN
|
||||
REVOKE ALTER ANY DATASPACE FROM scadabridge_audit_purger;
|
||||
REVOKE UPDATE (SourceNode) ON dbo.AuditLog FROM scadabridge_audit_purger;
|
||||
END"); // AUDIT-PURGE-ALLOWED: role grant reversal for the maintenance principal
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -115,7 +115,11 @@ public static class ServiceCollectionExtensions
|
||||
// hosted service. Scoped because the implementation reuses the per-scope
|
||||
// ScadaBridgeDbContext for raw-SQL execution; the hosted service opens a
|
||||
// fresh scope on each tick (mirrors AuditLogPurgeActor / AuditLogIngestActor).
|
||||
services.AddScoped<IPartitionMaintenance, AuditLogPartitionMaintenance>();
|
||||
// Runs on the maintenance connection when AuditLog:MaintenanceConnectionString is set
|
||||
// (ScadaBridge#113), else on the scope's own context as before.
|
||||
services.AddScoped<IPartitionMaintenance>(sp => new AuditLogPartitionMaintenance(
|
||||
sp.GetRequiredService<AuditMaintenanceContext>().Context,
|
||||
sp.GetService<Microsoft.Extensions.Logging.ILogger<AuditLogPartitionMaintenance>>()));
|
||||
|
||||
// AuditLog maintenance connection (ScadaBridge#107 item 3): the chain job and both purge
|
||||
// paths run on AuditMaintenanceContext — the scope's own context unless a separate
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
using Microsoft.Extensions.Diagnostics.HealthChecks;
|
||||
using ZB.MOM.WW.ScadaBridge.AuditLog.Central.HeadExport;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Host.Health;
|
||||
|
||||
/// <summary>
|
||||
/// /health/ready entry (ScadaBridge#115): Degraded while the off-box AuditLog chain-head export
|
||||
/// (<c>AuditLog:HashChain:HeadExport</c>) is failing, or has had no success for more than two
|
||||
/// intervals. Degraded maps to HTTP 200, so it shows on the Overview dashboard without taking
|
||||
/// the node out of rotation. Healthy when the export is disabled.
|
||||
/// </summary>
|
||||
public sealed class AuditChainHeadExportHealthCheck : IHealthCheck
|
||||
{
|
||||
private readonly AuditChainHeadExportStatus _status;
|
||||
private readonly TimeProvider _time;
|
||||
|
||||
/// <summary>Initializes the check.</summary>
|
||||
/// <param name="status">The export status singleton.</param>
|
||||
/// <param name="time">Clock (defaults to the system clock).</param>
|
||||
public AuditChainHeadExportHealthCheck(AuditChainHeadExportStatus status, TimeProvider? time = null)
|
||||
{
|
||||
_status = status ?? throw new ArgumentNullException(nameof(status));
|
||||
_time = time ?? TimeProvider.System;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<HealthCheckResult> CheckHealthAsync(
|
||||
HealthCheckContext context, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (!_status.Enabled)
|
||||
{
|
||||
return Task.FromResult(HealthCheckResult.Healthy("Chain-head export disabled."));
|
||||
}
|
||||
|
||||
var now = _time.GetUtcNow().UtcDateTime;
|
||||
var data = new Dictionary<string, object>(StringComparer.Ordinal)
|
||||
{
|
||||
["totalFailures"] = _status.TotalFailures,
|
||||
["consecutiveFailures"] = _status.ConsecutiveFailures,
|
||||
["intervalMinutes"] = _status.IntervalMinutes,
|
||||
};
|
||||
if (_status.LastSuccessUtc is { } ok)
|
||||
{
|
||||
data["lastSuccessUtc"] = ok;
|
||||
}
|
||||
|
||||
if (_status.ConsecutiveFailures > 0)
|
||||
{
|
||||
return Task.FromResult(HealthCheckResult.Degraded(
|
||||
$"Chain-head export failing ({_status.ConsecutiveFailures} consecutive): {_status.LastError}", data: data));
|
||||
}
|
||||
|
||||
var limit = TimeSpan.FromMinutes(2L * _status.IntervalMinutes);
|
||||
var reference = _status.LastSuccessUtc ?? _status.StartedUtc;
|
||||
if (reference is { } r && now - r > limit)
|
||||
{
|
||||
return Task.FromResult(HealthCheckResult.Degraded(
|
||||
$"No successful chain-head export for more than 2 intervals (last {(_status.LastSuccessUtc is null ? "never" : _status.LastSuccessUtc.Value.ToString("O"))}).", data: data));
|
||||
}
|
||||
|
||||
return Task.FromResult(HealthCheckResult.Healthy("Chain-head export OK.", data));
|
||||
}
|
||||
}
|
||||
@@ -579,6 +579,12 @@ try
|
||||
"bundle-signing-policy",
|
||||
failureStatus: null,
|
||||
tags: new[] { ZbHealthTags.Ready })
|
||||
// ScadaBridge#115: Degraded (HTTP 200) while the off-box chain-head export is failing
|
||||
// or stale; healthy when the export is disabled.
|
||||
.AddTypeActivatedCheck<AuditChainHeadExportHealthCheck>(
|
||||
"audit-chain-head-export",
|
||||
failureStatus: null,
|
||||
tags: new[] { ZbHealthTags.Ready })
|
||||
// Oldest-Up-member check (review 01 [High]), unscoped: every central member competes for
|
||||
// the one active slot. Since ZB.MOM.WW.Health 0.3.0 this IS the shared check — it selects
|
||||
// by age, so the host-local OldestNodeActiveHealthCheck that existed only to avoid the old
|
||||
|
||||
@@ -426,7 +426,10 @@ public static class AuditEndpoints
|
||||
statusCode: 400);
|
||||
}
|
||||
|
||||
var repo = context.RequestServices.GetRequiredService<IAuditLogRepository>();
|
||||
// Maintenance login when AuditLog:MaintenanceConnectionString is set (ScadaBridge#113);
|
||||
// otherwise the keyed registration resolves to the runtime connection's context.
|
||||
var repo = context.RequestServices.GetKeyedService<IAuditLogRepository>(AuditLogServiceKeys.Maintenance)
|
||||
?? context.RequestServices.GetRequiredService<IAuditLogRepository>();
|
||||
var rowsUpdated = await repo.BackfillSourceNodeAsync(sentinel, beforeUtc.Value, batchSize, context.RequestAborted);
|
||||
|
||||
return Results.Json(new
|
||||
|
||||
+45
-2
@@ -50,7 +50,8 @@ public sealed class AuditChainHeadExportServiceTests
|
||||
new DateTime(2026, month, 1, 0, 0, 0, DateTimeKind.Utc), seq, Enumerable.Repeat((byte)seq, 32).ToArray(),
|
||||
null, null, rows, 0, new DateTime(2026, month, 2, 0, 0, 0, DateTimeKind.Utc));
|
||||
|
||||
private static (AuditChainHeadExportService Service, FakeStore Store, RecordingSink Sink) Create(bool exportUnchanged = false)
|
||||
private static (AuditChainHeadExportService Service, FakeStore Store, RecordingSink Sink) Create(
|
||||
bool exportUnchanged = false, AuditChainHeadExportStatus? status = null, TimeProvider? time = null)
|
||||
{
|
||||
var store = new FakeStore();
|
||||
var sink = new RecordingSink();
|
||||
@@ -62,7 +63,7 @@ public sealed class AuditChainHeadExportServiceTests
|
||||
});
|
||||
var service = new AuditChainHeadExportService(
|
||||
services.BuildServiceProvider().GetRequiredService<IServiceScopeFactory>(),
|
||||
options, sink, "central-a", NullLogger<AuditChainHeadExportService>.Instance);
|
||||
options, sink, "central-a", NullLogger<AuditChainHeadExportService>.Instance, time, status);
|
||||
return (service, store, sink);
|
||||
}
|
||||
|
||||
@@ -120,6 +121,48 @@ public sealed class AuditChainHeadExportServiceTests
|
||||
Assert.Equal(10, Assert.Single(Assert.Single(sink.Writes)).LastSeq);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#115: a failed tick bumps the failure counter metric and the status
|
||||
/// (consecutive + total + last error); the next success clears the consecutive count.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task FailedTick_RecordsMetricAndStatus_SuccessResetsConsecutive()
|
||||
{
|
||||
var status = new AuditChainHeadExportStatus();
|
||||
var (service, store, sink) = Create(status: status);
|
||||
store.Heads.Add(Head(4, 10, 10));
|
||||
long measured = 0;
|
||||
using var listener = new System.Diagnostics.Metrics.MeterListener();
|
||||
listener.InstrumentPublished = (i, l) =>
|
||||
{
|
||||
if (i.Name == "scadabridge.audit.chain_head_export.failures")
|
||||
{
|
||||
l.EnableMeasurementEvents(i);
|
||||
}
|
||||
};
|
||||
listener.SetMeasurementEventCallback<long>((_, v, _, _) => Interlocked.Add(ref measured, v));
|
||||
listener.Start();
|
||||
|
||||
sink.Fail = true;
|
||||
await service.TickAsync(default);
|
||||
await service.TickAsync(default);
|
||||
|
||||
Assert.Equal(2, Interlocked.Read(ref measured));
|
||||
Assert.Equal(2, status.ConsecutiveFailures);
|
||||
Assert.Equal(2, status.TotalFailures);
|
||||
Assert.Equal("share offline", status.LastError);
|
||||
Assert.Null(status.LastSuccessUtc);
|
||||
|
||||
sink.Fail = false;
|
||||
await service.TickAsync(default);
|
||||
|
||||
Assert.Equal(0, status.ConsecutiveFailures);
|
||||
Assert.Equal(2, status.TotalFailures);
|
||||
Assert.NotNull(status.LastSuccessUtc);
|
||||
Assert.Equal(2, Interlocked.Read(ref measured));
|
||||
}
|
||||
|
||||
/// <summary>The validator ignores a disabled export and checks each sink's settings when enabled.</summary>
|
||||
/// <param name="configure">Which settings to apply.</param>
|
||||
/// <param name="expectedError">A fragment of the expected failure, or null when valid.</param>
|
||||
|
||||
+62
@@ -210,6 +210,68 @@ public sealed class AuditLogArchiveCleanupTests : IClassFixture<MsSqlMigrationFi
|
||||
Assert.Empty(Directory.EnumerateFiles(Path.Combine(dir, "2027")));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#114: leftovers of an interrupted archive write are removed — but only the
|
||||
/// writer's own temp / manifest-less data files, old enough, in a year folder. Fresh files,
|
||||
/// committed archives and foreign files survive; dry run deletes nothing.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task InterruptedWriteOrphans_AreRemoved_OnlyWhenOldAndWriterNamed()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
var dir = Dir("orphans");
|
||||
var month = Month(2027, 6);
|
||||
await InsertAsync(Enumerable.Range(1, 3).Select(i => NewEvent(month.AddDays(i))));
|
||||
await DrainAsync();
|
||||
var committed = await ArchiveAsync(dir, month);
|
||||
var year = Directory.CreateDirectory(Path.Combine(dir, "2027")).FullName;
|
||||
|
||||
const string orphanA = "auditlog-2027-05-20270601T000000001Z";
|
||||
const string freshB = "auditlog-2027-05-20270601T000000002Z";
|
||||
var old = DateTime.UtcNow.AddDays(-3);
|
||||
string Make(string name, DateTime? written = null)
|
||||
{
|
||||
var path = Path.Combine(year, name);
|
||||
File.WriteAllText(path, "x");
|
||||
File.SetLastWriteTimeUtc(path, written ?? old);
|
||||
return path;
|
||||
}
|
||||
|
||||
var oldTmpData = Make(orphanA + ".parquet.tmp");
|
||||
var oldTmpTomb = Make(orphanA + ".tombstones.parquet.tmp");
|
||||
var oldTmpManifest = Make(orphanA + ".manifest.json.tmp");
|
||||
var oldNoManifestData = Make(orphanA + ".parquet");
|
||||
var oldNoManifestTomb = Make(orphanA + ".tombstones.parquet");
|
||||
var freshTmp = Make(freshB + ".parquet.tmp", DateTime.UtcNow.AddHours(-1));
|
||||
var freshNoManifest = Make(freshB + ".parquet", DateTime.UtcNow.AddHours(-1));
|
||||
var foreign = Make("notes.tmp");
|
||||
var wrongName = Make("auditlog-bad.parquet.tmp");
|
||||
var wrongYear = Make("auditlog-2026-05-20260601T000000003Z.parquet.tmp");
|
||||
Directory.CreateDirectory(Path.Combine(dir, "misc"));
|
||||
var otherDir = Path.Combine(dir, "misc", orphanA + ".parquet.tmp");
|
||||
File.WriteAllText(otherDir, "x");
|
||||
File.SetLastWriteTimeUtc(otherDir, old);
|
||||
var committedFiles = new[] { ManifestPath(dir, committed), Path.Combine(year, committed.Name + ".parquet"), Path.Combine(year, committed.Name + ".tombstones.parquet") };
|
||||
foreach (var f in committedFiles)
|
||||
{
|
||||
File.SetLastWriteTimeUtc(f, old);
|
||||
}
|
||||
|
||||
var removed = new[] { oldTmpData, oldTmpTomb, oldTmpManifest, oldNoManifestData, oldNoManifestTomb };
|
||||
|
||||
var dry = await CleanupAsync(dir, dryRun: true);
|
||||
Assert.Equal(5, dry.Items.Count(i => i.Action == AuditArchiveCleanupAction.DeletedOrphan));
|
||||
Assert.All(removed, f => Assert.True(File.Exists(f)));
|
||||
|
||||
var real = await CleanupAsync(dir, dryRun: false);
|
||||
Assert.Equal(5, real.Items.Count(i => i.Action == AuditArchiveCleanupAction.DeletedOrphan));
|
||||
Assert.All(removed, f => Assert.False(File.Exists(f)));
|
||||
Assert.All(new[] { freshTmp, freshNoManifest, foreign, wrongName, wrongYear, otherDir }.Concat(committedFiles), f => Assert.True(File.Exists(f), f));
|
||||
|
||||
Assert.Empty((await CleanupAsync(dir, dryRun: false)).Items);
|
||||
}
|
||||
|
||||
private static readonly JsonSerializerOptions Json = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase };
|
||||
|
||||
private sealed class FixedClock(DateTime now) : TimeProvider
|
||||
|
||||
+49
@@ -160,6 +160,55 @@ IF SUSER_ID(N'{MaintenanceLogin}') IS NOT NULL DROP LOGIN [{MaintenanceLogin}];"
|
||||
Assert.Same(plainScope.ServiceProvider.GetRequiredService<ScadaBridgeDbContext>(), shared.Context);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#113: the monthly roll-forward (<c>ALTER PARTITION FUNCTION … SPLIT RANGE</c>) and
|
||||
/// the <c>SourceNode</c> backfill run on the maintenance login — and are refused to the writer
|
||||
/// login — with only the role grants the <c>AuditMaintenanceGrants</c> migration provisions.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[SkippableFact]
|
||||
public async Task RollForwardAndSourceNodeBackfill_RunOnTheMaintenanceLogin_NotTheWriter()
|
||||
{
|
||||
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
|
||||
var writerCs = As(WriterLogin);
|
||||
var maintenanceCs = As(MaintenanceLogin);
|
||||
var month = new DateTime(2027, 4, 1, 0, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
// A pre-feature row: no SourceNode, not chained.
|
||||
await using (var writerCtx = Context(writerCs))
|
||||
{
|
||||
var legacy = NewEvent(month.AddDays(1), AuditChannel.ApiOutbound) with { SourceNode = null };
|
||||
await new AuditLogRepository(writerCtx).InsertIfNotExistsAsync(legacy);
|
||||
}
|
||||
|
||||
// Negative controls: the writer login can neither split the partition function nor backfill.
|
||||
await using (var writerCtx = Context(writerCs))
|
||||
{
|
||||
await Assert.ThrowsAsync<SqlException>(() => new AuditLogPartitionMaintenance(writerCtx).EnsureLookaheadAsync(60));
|
||||
await Assert.ThrowsAsync<SqlException>(() =>
|
||||
new AuditLogRepository(writerCtx).BackfillSourceNodeAsync("unknown", month.AddYears(1), 100));
|
||||
}
|
||||
|
||||
var services = new ServiceCollection();
|
||||
services.AddLogging();
|
||||
services.AddConfigurationDatabase(writerCs, maintenanceCs);
|
||||
await using var provider = services.BuildServiceProvider();
|
||||
await using (var scope = provider.CreateAsyncScope())
|
||||
{
|
||||
var maintenance = scope.ServiceProvider.GetRequiredService<AuditMaintenanceContext>();
|
||||
var roll = scope.ServiceProvider.GetRequiredService<ZB.MOM.WW.ScadaBridge.Commons.Interfaces.IPartitionMaintenance>();
|
||||
var before = await roll.GetMaxBoundaryAsync();
|
||||
var added = await roll.EnsureLookaheadAsync(60);
|
||||
Assert.NotEmpty(added);
|
||||
Assert.True(await roll.GetMaxBoundaryAsync() > before);
|
||||
|
||||
var repo = scope.ServiceProvider.GetRequiredKeyedService<IAuditLogRepository>(
|
||||
ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories.AuditLogServiceKeys.Maintenance);
|
||||
Assert.Equal(1, await repo.BackfillSourceNodeAsync("unknown", month.AddYears(1), 100));
|
||||
Assert.Equal(MaintenanceLogin, await LoginNameAsync(maintenance.Context));
|
||||
}
|
||||
}
|
||||
|
||||
private string As(string login) => new SqlConnectionStringBuilder(_fixture.ConnectionString)
|
||||
{
|
||||
UserID = login,
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
using Microsoft.Extensions.Diagnostics.HealthChecks;
|
||||
using ZB.MOM.WW.ScadaBridge.AuditLog.Central.HeadExport;
|
||||
using ZB.MOM.WW.ScadaBridge.Host.Health;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
|
||||
|
||||
/// <summary>ScadaBridge#115: the chain-head export readiness entry.</summary>
|
||||
public sealed class AuditChainHeadExportHealthCheckTests
|
||||
{
|
||||
private sealed class Clock(DateTime now) : TimeProvider
|
||||
{
|
||||
public DateTime Now { get; set; } = now;
|
||||
public override DateTimeOffset GetUtcNow() => new(DateTime.SpecifyKind(Now, DateTimeKind.Utc));
|
||||
}
|
||||
|
||||
private static readonly DateTime T0 = new(2026, 11, 10, 12, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private static async Task<HealthStatus> Check(AuditChainHeadExportStatus s, Clock c)
|
||||
=> (await new AuditChainHeadExportHealthCheck(s, c).CheckHealthAsync(new HealthCheckContext())).Status;
|
||||
|
||||
/// <summary>Disabled export is healthy; failing or stale export is Degraded; recovery clears it.</summary>
|
||||
[Fact]
|
||||
public async Task Status_follows_export_outcome()
|
||||
{
|
||||
var clock = new Clock(T0);
|
||||
var status = new AuditChainHeadExportStatus();
|
||||
Assert.Equal(HealthStatus.Healthy, await Check(status, clock)); // disabled
|
||||
|
||||
status.MarkStarted(60, T0);
|
||||
Assert.Equal(HealthStatus.Healthy, await Check(status, clock));
|
||||
|
||||
status.RecordFailure(T0, "share offline");
|
||||
var result = await new AuditChainHeadExportHealthCheck(status, clock).CheckHealthAsync(new HealthCheckContext());
|
||||
Assert.Equal(HealthStatus.Degraded, result.Status);
|
||||
Assert.Contains("share offline", result.Description);
|
||||
|
||||
status.RecordSuccess(T0);
|
||||
Assert.Equal(HealthStatus.Healthy, await Check(status, clock));
|
||||
|
||||
clock.Now = T0.AddMinutes(121); // no success for > 2 intervals
|
||||
Assert.Equal(HealthStatus.Degraded, await Check(status, clock));
|
||||
status.RecordSuccess(clock.Now);
|
||||
Assert.Equal(HealthStatus.Healthy, await Check(status, clock));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user