feat(transport): observability for the import-session cap (register row 53)
Nothing counted open import sessions and nothing logged a rejection at MaxConcurrentImportSessions (default 8), so the register-row-43 session leak could only be inferred from rotating test failures. - Gauge scadabridge.central.transport.import_sessions.open on the existing ZB.MOM.WW.ScadaBridge meter — already in SiteServiceRegistration.ObservedMeters (the ZbTelemetryOptions.Meters allowlist), so no telemetry config change is needed and the series is not silently dropped. Provider-backed, mirroring the StoreAndForward queue-depth gauge: registered by BundleSessionEvictionService .StartAsync (Transport's only lifecycle seam) and identity-checked-cleared on StopAsync so a late stop cannot clobber a newer registrant. - IBundleSessionStore.OpenSessionCount counts only UNEXPIRED sessions. The raw dictionary count would report the node as pinned at the cap for up to a one-minute sweep interval after a burst had drained — the exact false positive the instrument exists to rule out. - BundleSessionStore.Open logs Warning with open count, cap and TTL when it rejects. Fires only on a real cap hit: a store full of expired sessions is reclaimed first and logs nothing. - Docs: docs/components/Transport.md + docs/requirements/Component-Transport.md. Tests: Transport 173 passed (+6 new — Warning content, two negative controls, unexpired-only counting, gauge register/clear, no-clobber-on-late-stop).
This commit is contained in:
Joseph Doherty
+248
@@ -0,0 +1,248 @@
|
||||
using System.Diagnostics.Metrics;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport;
|
||||
using ZB.MOM.WW.ScadaBridge.Transport.Import;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Transport.Tests.Import;
|
||||
|
||||
/// <summary>
|
||||
/// Register row 53: <c>MaxConcurrentImportSessions</c> had no observability — nothing
|
||||
/// counted open sessions and nothing logged a rejection at the cap, so the register-row-43
|
||||
/// session leak could only be inferred from rotating test failures. These tests pin the
|
||||
/// two signals that close that gap: the
|
||||
/// <c>scadabridge.central.transport.import_sessions.open</c> gauge and the Warning-level
|
||||
/// log emitted when <see cref="BundleSessionStore.Open"/> rejects at the cap.
|
||||
/// </summary>
|
||||
public sealed class BundleSessionObservabilityTests
|
||||
{
|
||||
private const string OpenSessionsGauge = "scadabridge.central.transport.import_sessions.open";
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Warning log on cap rejection
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public void Open_AtCap_LogsWarningWithCountAndCap()
|
||||
{
|
||||
// The rejection used to be visible only as the exception the caller saw. The
|
||||
// operator-facing signal must name BOTH the current count and the cap — that
|
||||
// pair is what distinguishes genuine concurrency from a leak.
|
||||
var clock = new TestClock(DateTimeOffset.UtcNow);
|
||||
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
|
||||
var logger = new CapturingLogger();
|
||||
var sut = new BundleSessionStore(opts, clock, logger);
|
||||
var future = clock.GetUtcNow().AddMinutes(30);
|
||||
|
||||
for (var i = 0; i < 3; i++)
|
||||
{
|
||||
sut.Open(SessionExpiringAt(future));
|
||||
}
|
||||
|
||||
Assert.Throws<InvalidOperationException>(() => sut.Open(SessionExpiringAt(future)));
|
||||
|
||||
var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning);
|
||||
Assert.Contains("Bundle import session rejected", warning.Message, StringComparison.Ordinal);
|
||||
Assert.Contains("3 of 3", warning.Message, StringComparison.Ordinal);
|
||||
Assert.Contains("MaxConcurrentImportSessions", warning.Message, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Open_UnderCap_LogsNothing()
|
||||
{
|
||||
// Negative control: the Warning must be a cap signal, not per-open noise.
|
||||
var clock = new TestClock(DateTimeOffset.UtcNow);
|
||||
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 });
|
||||
var logger = new CapturingLogger();
|
||||
var sut = new BundleSessionStore(opts, clock, logger);
|
||||
|
||||
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
|
||||
|
||||
Assert.Empty(logger.Entries);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Open_AtCapOfExpiredSessions_ReclaimsAndLogsNothing()
|
||||
{
|
||||
// A store that is full only of stale entries reclaims and admits the caller —
|
||||
// that is not a cap hit and must not produce a Warning, or the signal would fire
|
||||
// routinely on any node whose sweep has not yet run.
|
||||
var clock = new TestClock(DateTimeOffset.UtcNow);
|
||||
var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 });
|
||||
var logger = new CapturingLogger();
|
||||
var sut = new BundleSessionStore(opts, clock, logger);
|
||||
|
||||
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
|
||||
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
|
||||
clock.Advance(TimeSpan.FromMinutes(10));
|
||||
|
||||
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
|
||||
|
||||
Assert.Empty(logger.Entries);
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// OpenSessionCount / gauge
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public void OpenSessionCount_ExcludesExpiredButUnsweptSessions()
|
||||
{
|
||||
// The eviction sweep runs once a minute, so a raw dictionary count would report
|
||||
// the store as pinned at the cap for up to a sweep interval after a burst had
|
||||
// actually drained — exactly the false signal this instrument exists to rule out.
|
||||
var clock = new TestClock(DateTimeOffset.UtcNow);
|
||||
var sut = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
|
||||
|
||||
sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5)));
|
||||
var live = SessionExpiringAt(clock.GetUtcNow().AddMinutes(60));
|
||||
sut.Open(live);
|
||||
|
||||
Assert.Equal(2, sut.OpenSessionCount);
|
||||
|
||||
clock.Advance(TimeSpan.FromMinutes(10));
|
||||
|
||||
// The short session has expired but nothing has swept it yet.
|
||||
Assert.Equal(1, sut.OpenSessionCount);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EvictionService_RegistersTheGaugeProvider_AndClearsItOnStop()
|
||||
{
|
||||
// The eviction service is Transport's only lifecycle seam, so it owns the
|
||||
// process-global gauge slot. The gauge is read as a delta against a baseline
|
||||
// because the instrument is a process-wide static.
|
||||
var clock = new TestClock(DateTimeOffset.UtcNow);
|
||||
var store = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
|
||||
var service = new BundleSessionEvictionService(store, NullLogger<BundleSessionEvictionService>.Instance);
|
||||
|
||||
var baseline = ReadGauge();
|
||||
|
||||
await service.StartAsync(CancellationToken.None);
|
||||
try
|
||||
{
|
||||
store.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
|
||||
store.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
|
||||
|
||||
Assert.Equal(2, ReadGauge());
|
||||
}
|
||||
finally
|
||||
{
|
||||
await service.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
// After the clear the gauge falls back to reporting 0, not through a dead store.
|
||||
Assert.Equal(0, ReadGauge());
|
||||
Assert.Equal(0, baseline);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task EvictionService_StopDoesNotClobberANewerInstancesProvider()
|
||||
{
|
||||
// Identity-checked compare-and-clear: a late stop of a superseded instance must
|
||||
// leave a newer registrant's provider in place (mirrors the StoreAndForward
|
||||
// queue-depth gauge's contract).
|
||||
var clock = new TestClock(DateTimeOffset.UtcNow);
|
||||
var oldStore = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
|
||||
var newStore = new BundleSessionStore(Options.Create(new TransportOptions()), clock, NullLogger<BundleSessionStore>.Instance);
|
||||
var oldService = new BundleSessionEvictionService(oldStore, NullLogger<BundleSessionEvictionService>.Instance);
|
||||
var newService = new BundleSessionEvictionService(newStore, NullLogger<BundleSessionEvictionService>.Instance);
|
||||
|
||||
await oldService.StartAsync(CancellationToken.None);
|
||||
await newService.StartAsync(CancellationToken.None);
|
||||
try
|
||||
{
|
||||
newStore.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)));
|
||||
|
||||
// Late stop of the superseded instance.
|
||||
await oldService.StopAsync(CancellationToken.None);
|
||||
|
||||
Assert.Equal(1, ReadGauge());
|
||||
}
|
||||
finally
|
||||
{
|
||||
await newService.StopAsync(CancellationToken.None);
|
||||
}
|
||||
|
||||
Assert.Equal(0, ReadGauge());
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// Forces one synchronous observation of the open-import-sessions gauge and returns
|
||||
/// the measured value. Mirrors the MeterListener helper used by the Communication
|
||||
/// and StoreAndForward telemetry suites.
|
||||
/// </summary>
|
||||
/// <returns>The gauge's current value, or 0 if it never reported.</returns>
|
||||
private static long ReadGauge()
|
||||
{
|
||||
long observed = 0;
|
||||
using var listener = new MeterListener();
|
||||
listener.InstrumentPublished = (instrument, l) =>
|
||||
{
|
||||
if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName
|
||||
&& instrument.Name == OpenSessionsGauge)
|
||||
{
|
||||
l.EnableMeasurementEvents(instrument);
|
||||
}
|
||||
};
|
||||
listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => observed = measurement);
|
||||
listener.Start();
|
||||
listener.RecordObservableInstruments();
|
||||
return observed;
|
||||
}
|
||||
|
||||
private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new()
|
||||
{
|
||||
SessionId = Guid.NewGuid(),
|
||||
Manifest = StubManifest(),
|
||||
DecryptedContent = Array.Empty<byte>(),
|
||||
ExpiresAt = expiresAt,
|
||||
};
|
||||
|
||||
private static BundleManifest StubManifest() => new(
|
||||
BundleFormatVersion: 1,
|
||||
SchemaVersion: "1.0",
|
||||
CreatedAtUtc: DateTimeOffset.UnixEpoch,
|
||||
SourceEnvironment: "test",
|
||||
ExportedBy: "t",
|
||||
ScadaBridgeVersion: "1",
|
||||
ContentHash: "0",
|
||||
Encryption: null,
|
||||
Summary: new BundleSummary(0, 0, 0, 0, 0, 0, 0, 0),
|
||||
Contents: Array.Empty<ManifestContentEntry>());
|
||||
|
||||
/// <summary>Minimal mutable-clock <see cref="TimeProvider"/> for deterministic expiry.</summary>
|
||||
private sealed class TestClock : TimeProvider
|
||||
{
|
||||
private DateTimeOffset _now;
|
||||
public TestClock(DateTimeOffset start) { _now = start; }
|
||||
public override DateTimeOffset GetUtcNow() => _now;
|
||||
public void Advance(TimeSpan by) => _now = _now.Add(by);
|
||||
}
|
||||
|
||||
/// <summary>Captures formatted log entries so the cap-rejection Warning can be asserted.</summary>
|
||||
private sealed class CapturingLogger : ILogger<BundleSessionStore>
|
||||
{
|
||||
public List<(LogLevel Level, string Message)> Entries { get; } = [];
|
||||
|
||||
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||
|
||||
public bool IsEnabled(LogLevel logLevel) => true;
|
||||
|
||||
public void Log<TState>(
|
||||
LogLevel logLevel,
|
||||
EventId eventId,
|
||||
TState state,
|
||||
Exception? exception,
|
||||
Func<TState, Exception?, string> formatter)
|
||||
{
|
||||
Entries.Add((logLevel, formatter(state, exception)));
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user